clientStmt.c 32.9 KB
Newer Older
D
dapan1121 已提交
1 2 3 4 5

#include "clientInt.h"
#include "clientLog.h"
#include "tdef.h"

X
Xiaoyu Wang 已提交
6 7
#include "clientStmt.h"

dengyihao's avatar
dengyihao 已提交
8 9
char* gStmtStatusStr[] = {"unknown",     "init", "prepare", "settbname", "settags",
                          "fetchFields", "bind", "bindCol", "addBatch",  "exec"};
10

11
static int32_t stmtCreateRequest(STscStmt* pStmt) {
D
dapan1121 已提交
12
  int32_t code = 0;
H
Hongze Cheng 已提交
13 14

  if (pStmt->exec.pRequest == NULL) {
dengyihao's avatar
dengyihao 已提交
15 16
    code = buildRequest(pStmt->taos->id, pStmt->sql.sqlStr, pStmt->sql.sqlLen, NULL, false, &pStmt->exec.pRequest,
                        pStmt->reqid);
dengyihao's avatar
dengyihao 已提交
17
    if (pStmt->reqid != 0) {
dengyihao's avatar
dengyihao 已提交
18 19
      pStmt->reqid++;
    }
D
dapan1121 已提交
20 21 22
    if (TSDB_CODE_SUCCESS == code) {
      pStmt->exec.pRequest->syncQuery = true;
    }
23
  }
D
dapan1121 已提交
24 25

  return code;
26 27
}

D
stmt  
dapan1121 已提交
28
int32_t stmtSwitchStatus(STscStmt* pStmt, STMT_STATUS newStatus) {
D
dapan1121 已提交
29
  int32_t code = 0;
X
Xiaoyu Wang 已提交
30

31 32 33 34
  if (newStatus >= STMT_INIT && newStatus < STMT_MAX) {
    STMT_LOG_SEQ(newStatus);
  }

D
stmt  
dapan1121 已提交
35
  switch (newStatus) {
D
dapan1121 已提交
36 37
    case STMT_PREPARE:
      break;
D
stmt  
dapan1121 已提交
38
    case STMT_SETTBNAME:
D
dapan1121 已提交
39
      if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND) || STMT_STATUS_EQ(BIND_COL)) {
D
dapan1121 已提交
40 41 42 43
        code = TSDB_CODE_TSC_STMT_API_ERROR;
      }
      break;
    case STMT_SETTAGS:
D
dapan1121 已提交
44
      if (STMT_STATUS_NE(SETTBNAME) && STMT_STATUS_NE(FETCH_FIELDS)) {
D
dapan1121 已提交
45 46 47 48 49 50 51 52 53 54 55 56
        code = TSDB_CODE_TSC_STMT_API_ERROR;
      }
      break;
    case STMT_FETCH_FIELDS:
      if (STMT_STATUS_EQ(INIT)) {
        code = TSDB_CODE_TSC_STMT_API_ERROR;
      }
      break;
    case STMT_BIND:
      if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND_COL)) {
        code = TSDB_CODE_TSC_STMT_API_ERROR;
      }
X
Xiaoyu Wang 已提交
57 58 59 60 61
      /*
            if ((pStmt->sql.type == STMT_TYPE_MULTI_INSERT) && ()) {
              code = TSDB_CODE_TSC_STMT_API_ERROR;
            }
      */
D
dapan1121 已提交
62 63 64 65 66 67 68 69 70 71
      break;
    case STMT_BIND_COL:
      if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND)) {
        code = TSDB_CODE_TSC_STMT_API_ERROR;
      }
      break;
    case STMT_ADD_BATCH:
      if (STMT_STATUS_NE(BIND) && STMT_STATUS_NE(BIND_COL) && STMT_STATUS_NE(FETCH_FIELDS)) {
        code = TSDB_CODE_TSC_STMT_API_ERROR;
      }
D
stmt  
dapan1121 已提交
72
      break;
D
dapan1121 已提交
73
    case STMT_EXECUTE:
D
dapan1121 已提交
74
      if (STMT_TYPE_QUERY == pStmt->sql.type) {
X
Xiaoyu Wang 已提交
75 76
        if (STMT_STATUS_NE(ADD_BATCH) && STMT_STATUS_NE(FETCH_FIELDS) && STMT_STATUS_NE(BIND) &&
            STMT_STATUS_NE(BIND_COL)) {
D
dapan1121 已提交
77 78 79 80 81 82
          code = TSDB_CODE_TSC_STMT_API_ERROR;
        }
      } else {
        if (STMT_STATUS_NE(ADD_BATCH) && STMT_STATUS_NE(FETCH_FIELDS)) {
          code = TSDB_CODE_TSC_STMT_API_ERROR;
        }
D
dapan1121 已提交
83
      }
D
dapan1121 已提交
84
      break;
D
stmt  
dapan1121 已提交
85
    default:
D
dapan1121 已提交
86
      code = TSDB_CODE_TSC_APP_ERROR;
D
stmt  
dapan1121 已提交
87 88 89
      break;
  }

D
dapan1121 已提交
90
  STMT_ERR_RET(code);
D
stmt  
dapan1121 已提交
91 92 93 94 95 96

  pStmt->sql.status = newStatus;

  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
97
int32_t stmtGetTbName(TAOS_STMT* stmt, char** tbName) {
D
stmt  
dapan1121 已提交
98 99
  STscStmt* pStmt = (STscStmt*)stmt;

D
stmt  
dapan1121 已提交
100
  pStmt->sql.type = STMT_TYPE_MULTI_INSERT;
101

D
dapan 已提交
102
  if ('\0' == pStmt->bInfo.tbName[0]) {
D
stmt  
dapan1121 已提交
103 104 105 106
    tscError("no table name set");
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_TBNAME_ERROR);
  }

D
stmt  
dapan1121 已提交
107
  *tbName = pStmt->bInfo.tbName;
D
stmt  
dapan1121 已提交
108 109 110 111

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
112
int32_t stmtBackupQueryFields(STscStmt* pStmt) {
X
Xiaoyu Wang 已提交
113
  SStmtQueryResInfo* pRes = &pStmt->sql.queryRes;
D
dapan1121 已提交
114 115
  pRes->numOfCols = pStmt->exec.pRequest->body.resInfo.numOfCols;
  pRes->precision = pStmt->exec.pRequest->body.resInfo.precision;
X
Xiaoyu Wang 已提交
116

D
dapan1121 已提交
117 118 119 120
  int32_t size = pRes->numOfCols * sizeof(TAOS_FIELD);
  pRes->fields = taosMemoryMalloc(size);
  pRes->userFields = taosMemoryMalloc(size);
  if (NULL == pRes->fields || NULL == pRes->userFields) {
D
dapan1121 已提交
121 122
    STMT_ERR_RET(TSDB_CODE_TSC_OUT_OF_MEMORY);
  }
D
dapan1121 已提交
123 124 125 126 127 128 129
  memcpy(pRes->fields, pStmt->exec.pRequest->body.resInfo.fields, size);
  memcpy(pRes->userFields, pStmt->exec.pRequest->body.resInfo.userFields, size);

  return TSDB_CODE_SUCCESS;
}

int32_t stmtRestoreQueryFields(STscStmt* pStmt) {
X
Xiaoyu Wang 已提交
130 131 132
  SStmtQueryResInfo* pRes = &pStmt->sql.queryRes;
  int32_t            size = pRes->numOfCols * sizeof(TAOS_FIELD);

D
dapan1121 已提交
133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150
  pStmt->exec.pRequest->body.resInfo.numOfCols = pRes->numOfCols;
  pStmt->exec.pRequest->body.resInfo.precision = pRes->precision;

  if (NULL == pStmt->exec.pRequest->body.resInfo.fields) {
    pStmt->exec.pRequest->body.resInfo.fields = taosMemoryMalloc(size);
    if (NULL == pStmt->exec.pRequest->body.resInfo.fields) {
      STMT_ERR_RET(TSDB_CODE_TSC_OUT_OF_MEMORY);
    }
    memcpy(pStmt->exec.pRequest->body.resInfo.fields, pRes->fields, size);
  }

  if (NULL == pStmt->exec.pRequest->body.resInfo.userFields) {
    pStmt->exec.pRequest->body.resInfo.userFields = taosMemoryMalloc(size);
    if (NULL == pStmt->exec.pRequest->body.resInfo.userFields) {
      STMT_ERR_RET(TSDB_CODE_TSC_OUT_OF_MEMORY);
    }
    memcpy(pStmt->exec.pRequest->body.resInfo.userFields, pRes->userFields, size);
  }
D
dapan1121 已提交
151 152 153 154

  return TSDB_CODE_SUCCESS;
}

155
int32_t stmtUpdateBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, SName* tbName, const char* sTableName, bool autoCreateTbl) {
D
stmt  
dapan1121 已提交
156
  STscStmt* pStmt = (STscStmt*)stmt;
157 158
  char           tbFName[TSDB_TABLE_FNAME_LEN];
  tNameExtractFullName(tbName, tbFName);
D
stmt  
dapan1121 已提交
159

160
  memcpy(&pStmt->bInfo.sname, tbName, sizeof(*tbName));
D
dapan 已提交
161 162
  strncpy(pStmt->bInfo.tbFName, tbFName, sizeof(pStmt->bInfo.tbFName) - 1);
  pStmt->bInfo.tbFName[sizeof(pStmt->bInfo.tbFName) - 1] = 0;
163

164
  pStmt->bInfo.tbUid = autoCreateTbl ? 0 : pTableMeta->uid;
D
stmt  
dapan1121 已提交
165 166 167
  pStmt->bInfo.tbSuid = pTableMeta->suid;
  pStmt->bInfo.tbType = pTableMeta->tableType;
  pStmt->bInfo.boundTags = tags;
D
dapan1121 已提交
168
  pStmt->bInfo.tagsCached = false;
D
dapan1121 已提交
169
  tstrncpy(pStmt->bInfo.stbFName, sTableName, sizeof(pStmt->bInfo.stbFName));
D
stmt  
dapan1121 已提交
170 171 172 173

  return TSDB_CODE_SUCCESS;
}

D
dapan 已提交
174
int32_t stmtUpdateExecInfo(TAOS_STMT* stmt, SHashObj* pVgHash, SHashObj* pBlockHash, bool autoCreateTbl) {
D
stmt  
dapan1121 已提交
175 176
  STscStmt* pStmt = (STscStmt*)stmt;

D
dapan1121 已提交
177
  pStmt->sql.pVgHash = pVgHash;
D
stmt  
dapan1121 已提交
178
  pStmt->exec.pBlockHash = pBlockHash;
D
dapan 已提交
179
  pStmt->exec.autoCreateTbl = autoCreateTbl;
D
stmt  
dapan1121 已提交
180 181 182 183

  return TSDB_CODE_SUCCESS;
}

184
int32_t stmtUpdateInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, SName* tbName, bool autoCreateTbl,
wmmhello's avatar
wmmhello 已提交
185
                       SHashObj* pVgHash, SHashObj* pBlockHash, const char* sTableName) {
D
dapan 已提交
186 187
  STscStmt* pStmt = (STscStmt*)stmt;

188
  STMT_ERR_RET(stmtUpdateBindInfo(stmt, pTableMeta, tags, tbName, sTableName, autoCreateTbl));
D
dapan 已提交
189
  STMT_ERR_RET(stmtUpdateExecInfo(stmt, pVgHash, pBlockHash, autoCreateTbl));
D
dapan 已提交
190

D
dapan 已提交
191
  pStmt->sql.autoCreateTbl = autoCreateTbl;
192

D
dapan 已提交
193 194 195
  return TSDB_CODE_SUCCESS;
}

D
stmt  
dapan1121 已提交
196 197 198
int32_t stmtGetExecInfo(TAOS_STMT* stmt, SHashObj** pVgHash, SHashObj** pBlockHash) {
  STscStmt* pStmt = (STscStmt*)stmt;

D
dapan1121 已提交
199
  *pVgHash = pStmt->sql.pVgHash;
D
stmt  
dapan1121 已提交
200
  *pBlockHash = pStmt->exec.pBlockHash;
D
stmt  
dapan1121 已提交
201 202 203 204

  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
205
int32_t stmtCacheBlock(STscStmt* pStmt) {
D
stmt  
dapan1121 已提交
206 207 208
  if (pStmt->sql.type != STMT_TYPE_MULTI_INSERT) {
    return TSDB_CODE_SUCCESS;
  }
D
stmt  
dapan1121 已提交
209

210
  uint64_t uid = pStmt->bInfo.tbUid;
D
dapan 已提交
211
  uint64_t cacheUid = (TSDB_CHILD_TABLE == pStmt->bInfo.tbType) ? pStmt->bInfo.tbSuid : uid;
D
stmt  
dapan1121 已提交
212

D
dapan 已提交
213
  if (taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid))) {
D
stmt  
dapan1121 已提交
214 215 216
    return TSDB_CODE_SUCCESS;
  }

D
dapan 已提交
217
  STableDataBlocks** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
dengyihao's avatar
dengyihao 已提交
218
  if (!pSrc) {
wmmhello's avatar
wmmhello 已提交
219 220
    return TSDB_CODE_OUT_OF_MEMORY;
  }
dengyihao's avatar
dengyihao 已提交
221
  STableDataBlocks* pDst = NULL;
222

D
stmt  
dapan1121 已提交
223
  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc));
D
stmt  
dapan1121 已提交
224 225

  SStmtTableCache cache = {
X
Xiaoyu Wang 已提交
226 227
      .pDataBlock = pDst,
      .boundTags = pStmt->bInfo.boundTags,
D
stmt  
dapan1121 已提交
228 229
  };

D
dapan 已提交
230
  if (taosHashPut(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid), &cache, sizeof(cache))) {
D
stmt  
dapan1121 已提交
231 232
    return TSDB_CODE_OUT_OF_MEMORY;
  }
D
stmt  
dapan1121 已提交
233

D
dapan 已提交
234 235 236 237 238
  if (pStmt->sql.autoCreateTbl) {
    pStmt->bInfo.tagsCached = true;
  } else {
    pStmt->bInfo.boundTags = NULL;
  }
239

D
stmt  
dapan1121 已提交
240 241 242
  return TSDB_CODE_SUCCESS;
}

D
stmt  
dapan1121 已提交
243 244
int32_t stmtParseSql(STscStmt* pStmt) {
  SStmtCallback stmtCb = {
245 246 247 248
      .pStmt = pStmt,
      .getTbNameFn = stmtGetTbName,
      .setInfoFn = stmtUpdateInfo,
      .getExecInfoFn = stmtGetExecInfo,
D
stmt  
dapan1121 已提交
249
  };
D
stmt  
dapan1121 已提交
250

251
  STMT_ERR_RET(stmtCreateRequest(pStmt));
H
Hongze Cheng 已提交
252

D
stmt  
dapan1121 已提交
253 254
  STMT_ERR_RET(parseSql(pStmt->exec.pRequest, false, &pStmt->sql.pQuery, &stmtCb));

D
stmt  
dapan1121 已提交
255
  pStmt->bInfo.needParse = false;
X
Xiaoyu Wang 已提交
256

D
dapan1121 已提交
257 258 259 260 261 262
  if (pStmt->sql.pQuery->pRoot && 0 == pStmt->sql.type) {
    pStmt->sql.type = STMT_TYPE_INSERT;
  } else if (pStmt->sql.pQuery->pPrepareRoot) {
    pStmt->sql.type = STMT_TYPE_QUERY;
  }

D
stmt  
dapan1121 已提交
263 264
  return TSDB_CODE_SUCCESS;
}
D
stmt  
dapan1121 已提交
265

D
stmt  
dapan1121 已提交
266
int32_t stmtCleanBindInfo(STscStmt* pStmt) {
D
stmt  
dapan1121 已提交
267 268 269 270
  pStmt->bInfo.tbUid = 0;
  pStmt->bInfo.tbSuid = 0;
  pStmt->bInfo.tbType = 0;
  pStmt->bInfo.needParse = true;
D
dapan 已提交
271
  pStmt->bInfo.inExecCache = false;
D
stmt  
dapan1121 已提交
272

D
dapan 已提交
273 274
  pStmt->bInfo.tbName[0] = 0;
  pStmt->bInfo.tbFName[0] = 0;
D
dapan1121 已提交
275 276 277 278
  if (!pStmt->bInfo.tagsCached) {
    destroyBoundColumnInfo(pStmt->bInfo.boundTags);
    taosMemoryFreeClear(pStmt->bInfo.boundTags);
  }
wmmhello's avatar
wmmhello 已提交
279
  memset(pStmt->bInfo.stbFName, 0, TSDB_TABLE_FNAME_LEN);
D
stmt  
dapan1121 已提交
280
  return TSDB_CODE_SUCCESS;
D
stmt  
dapan1121 已提交
281 282
}

D
dapan1121 已提交
283 284
int32_t stmtCleanExecInfo(STscStmt* pStmt, bool keepTable, bool deepClean) {
  if (STMT_TYPE_QUERY != pStmt->sql.type || deepClean) {
D
dapan1121 已提交
285 286 287
    taos_free_result(pStmt->exec.pRequest);
    pStmt->exec.pRequest = NULL;
  }
D
stmt  
dapan1121 已提交
288

D
dapan 已提交
289
  size_t keyLen = 0;
290
  void*  pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
D
stmt  
dapan1121 已提交
291
  while (pIter) {
292 293 294 295
    STableDataBlocks* pBlocks = *(STableDataBlocks**)pIter;
    char*             key = taosHashGetKey(pIter, &keyLen);
    STableMeta*       pMeta = qGetTableMetaInDataBlock(pBlocks);

D
dapan 已提交
296
    if (keepTable && (strlen(pStmt->bInfo.tbFName) == keyLen) && strncmp(pStmt->bInfo.tbFName, key, keyLen) == 0) {
D
stmt  
dapan1121 已提交
297
      STMT_ERR_RET(qResetStmtDataBlock(pBlocks, true));
X
Xiaoyu Wang 已提交
298

D
stmt  
dapan1121 已提交
299 300 301 302
      pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
      continue;
    }

D
dapan1121 已提交
303 304 305 306 307
    if (STMT_TYPE_MULTI_INSERT == pStmt->sql.type) {
      qFreeStmtDataBlock(pBlocks);
    } else {
      qDestroyStmtDataBlock(pBlocks);
    }
D
dapan 已提交
308
    taosHashRemove(pStmt->exec.pBlockHash, key, keyLen);
D
stmt  
dapan1121 已提交
309 310 311 312

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

D
dapan 已提交
313
  pStmt->exec.autoCreateTbl = false;
314

D
stmt  
dapan1121 已提交
315 316 317 318 319 320
  if (keepTable) {
    return TSDB_CODE_SUCCESS;
  }

  taosHashCleanup(pStmt->exec.pBlockHash);
  pStmt->exec.pBlockHash = NULL;
D
stmt  
dapan1121 已提交
321 322 323 324 325 326 327

  STMT_ERR_RET(stmtCleanBindInfo(pStmt));

  return TSDB_CODE_SUCCESS;
}

int32_t stmtCleanSQLInfo(STscStmt* pStmt) {
D
dapan1121 已提交
328 329
  taosMemoryFree(pStmt->sql.queryRes.fields);
  taosMemoryFree(pStmt->sql.queryRes.userFields);
D
stmt  
dapan1121 已提交
330 331
  taosMemoryFree(pStmt->sql.sqlStr);
  qDestroyQuery(pStmt->sql.pQuery);
D
dapan1121 已提交
332
  taosArrayDestroy(pStmt->sql.nodeList);
D
dapan1121 已提交
333 334
  taosHashCleanup(pStmt->sql.pVgHash);
  pStmt->sql.pVgHash = NULL;
X
Xiaoyu Wang 已提交
335 336

  void* pIter = taosHashIterate(pStmt->sql.pTableCache, NULL);
D
stmt  
dapan1121 已提交
337
  while (pIter) {
X
Xiaoyu Wang 已提交
338
    SStmtTableCache* pCache = (SStmtTableCache*)pIter;
D
stmt  
dapan1121 已提交
339

D
stmt  
dapan1121 已提交
340
    qDestroyStmtDataBlock(pCache->pDataBlock);
D
stmt  
dapan1121 已提交
341
    destroyBoundColumnInfo(pCache->boundTags);
D
dapan1121 已提交
342
    taosMemoryFreeClear(pCache->boundTags);
X
Xiaoyu Wang 已提交
343

D
stmt  
dapan1121 已提交
344 345 346 347 348
    pIter = taosHashIterate(pStmt->sql.pTableCache, pIter);
  }
  taosHashCleanup(pStmt->sql.pTableCache);
  pStmt->sql.pTableCache = NULL;

D
dapan1121 已提交
349
  STMT_ERR_RET(stmtCleanExecInfo(pStmt, false, true));
D
stmt  
dapan1121 已提交
350
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
D
stmt  
dapan1121 已提交
351

D
dapan1121 已提交
352 353
  memset(&pStmt->sql, 0, sizeof(pStmt->sql));

D
stmt  
dapan1121 已提交
354 355 356
  return TSDB_CODE_SUCCESS;
}

357
int32_t stmtRebuildDataBlock(STscStmt* pStmt, STableDataBlocks* pDataBlock, STableDataBlocks** newBlock, uint64_t uid) {
358 359 360
  SEpSet           ep = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
  SVgroupInfo      vgInfo = {0};
  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
D
dapan1121 已提交
361 362 363
                           .requestId = pStmt->exec.pRequest->requestId,
                           .requestObjRefId = pStmt->exec.pRequest->self,
                           .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
364

D
dapan1121 已提交
365
  STMT_ERR_RET(catalogGetTableHashVgroup(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &vgInfo));
366
  STMT_ERR_RET(
D
dapan1121 已提交
367
      taosHashPut(pStmt->sql.pVgHash, (const char*)&vgInfo.vgId, sizeof(vgInfo.vgId), (char*)&vgInfo, sizeof(vgInfo)));
368

D
dapan 已提交
369 370 371 372 373
  STMT_ERR_RET(qRebuildStmtDataBlock(newBlock, pDataBlock, uid, vgInfo.vgId));

  return TSDB_CODE_SUCCESS;
}

D
stmt  
dapan1121 已提交
374
int32_t stmtGetFromCache(STscStmt* pStmt) {
D
stmt  
dapan1121 已提交
375
  pStmt->bInfo.needParse = true;
D
dapan 已提交
376
  pStmt->bInfo.inExecCache = false;
377 378 379

  STableDataBlocks* pBlockInExec =
      taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
D
dapan 已提交
380
  if (pBlockInExec) {
D
dapan1121 已提交
381
    pStmt->bInfo.needParse = false;
D
dapan 已提交
382 383 384
    pStmt->bInfo.inExecCache = true;

    if (pStmt->sql.autoCreateTbl) {
D
dapan1121 已提交
385
      tscDebug("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
D
dapan 已提交
386 387 388
      return TSDB_CODE_SUCCESS;
    }
  }
D
stmt  
dapan1121 已提交
389

D
stmt  
dapan1121 已提交
390
  if (NULL == pStmt->sql.pTableCache || taosHashGetSize(pStmt->sql.pTableCache) <= 0) {
D
dapan 已提交
391 392 393
    if (pStmt->bInfo.inExecCache) {
      ASSERT(taosHashGetSize(pStmt->exec.pBlockHash) == 1);
      pStmt->bInfo.needParse = false;
D
dapan1121 已提交
394
      tscDebug("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
D
dapan 已提交
395 396
      return TSDB_CODE_SUCCESS;
    }
397

D
dapan1121 已提交
398
    tscDebug("no stmt block cache for tb %s", pStmt->bInfo.tbFName);
D
stmt  
dapan1121 已提交
399 400 401
    return TSDB_CODE_SUCCESS;
  }

D
dapan 已提交
402 403 404 405
  if (NULL == pStmt->pCatalog) {
    STMT_ERR_RET(catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog));
  }

D
dapan 已提交
406 407 408 409
  if (pStmt->sql.autoCreateTbl) {
    SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &pStmt->bInfo.tbSuid, sizeof(pStmt->bInfo.tbSuid));
    if (pCache) {
      pStmt->bInfo.needParse = false;
D
dapan1121 已提交
410 411 412
      pStmt->exec.autoCreateTbl = true;

      pStmt->bInfo.tbUid = 0;
413

D
dapan 已提交
414
      STableDataBlocks* pNewBlock = NULL;
D
dapan 已提交
415
      STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataBlock, &pNewBlock, 0));
416 417 418

      if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
                      POINTER_BYTES)) {
D
dapan 已提交
419 420
        STMT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
      }
421

H
Hongze Cheng 已提交
422
      tscDebug("reuse stmt block for tb %s in sqlBlock, suid:0x%" PRIx64, pStmt->bInfo.tbFName, pStmt->bInfo.tbSuid);
D
dapan1121 已提交
423

D
dapan 已提交
424 425
      return TSDB_CODE_SUCCESS;
    }
426

D
dapan 已提交
427 428
    STMT_RET(stmtCleanBindInfo(pStmt));
  }
D
stmt  
dapan1121 已提交
429

430 431
  STableMeta*      pTableMeta = NULL;
  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
D
dapan1121 已提交
432 433 434
                           .requestId = pStmt->exec.pRequest->requestId,
                           .requestObjRefId = pStmt->exec.pRequest->self,
                           .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
435
  int32_t          code = catalogGetTableMeta(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &pTableMeta);
D
dapan1121 已提交
436 437
  if (TSDB_CODE_PAR_TABLE_NOT_EXIST == code) {
    STMT_ERR_RET(stmtCleanBindInfo(pStmt));
X
Xiaoyu Wang 已提交
438

D
dapan1121 已提交
439 440
    tscDebug("tb %s not exist", pStmt->bInfo.tbFName);

D
dapan1121 已提交
441 442 443 444
    return TSDB_CODE_SUCCESS;
  }

  STMT_ERR_RET(code);
X
Xiaoyu Wang 已提交
445

D
dapan1121 已提交
446 447
  uint64_t uid = pTableMeta->uid;
  uint64_t suid = pTableMeta->suid;
X
Xiaoyu Wang 已提交
448
  int8_t   tableType = pTableMeta->tableType;
D
dapan1121 已提交
449
  taosMemoryFree(pTableMeta);
D
dapan 已提交
450
  uint64_t cacheUid = (TSDB_CHILD_TABLE == tableType) ? suid : uid;
451

D
dapan1121 已提交
452
  if (uid == pStmt->bInfo.tbUid) {
D
stmt  
dapan1121 已提交
453
    pStmt->bInfo.needParse = false;
D
dapan1121 已提交
454

D
dapan1121 已提交
455 456
    tscDebug("tb %s is current table", pStmt->bInfo.tbFName);

D
stmt  
dapan1121 已提交
457 458 459
    return TSDB_CODE_SUCCESS;
  }

D
dapan 已提交
460
  if (pStmt->bInfo.inExecCache) {
D
dapan 已提交
461
    SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid));
D
stmt  
dapan1121 已提交
462
    if (NULL == pCache) {
463 464 465
      tscError("table [%s, %" PRIx64 ", %" PRIx64 "] found in exec blockHash, but not in sql blockHash",
               pStmt->bInfo.tbFName, uid, cacheUid);

D
stmt  
dapan1121 已提交
466 467
      STMT_ERR_RET(TSDB_CODE_TSC_APP_ERROR);
    }
X
Xiaoyu Wang 已提交
468

D
stmt  
dapan1121 已提交
469
    pStmt->bInfo.needParse = false;
X
Xiaoyu Wang 已提交
470

D
dapan1121 已提交
471 472 473
    pStmt->bInfo.tbUid = uid;
    pStmt->bInfo.tbSuid = suid;
    pStmt->bInfo.tbType = tableType;
D
stmt  
dapan1121 已提交
474
    pStmt->bInfo.boundTags = pCache->boundTags;
D
dapan1121 已提交
475
    pStmt->bInfo.tagsCached = true;
D
dapan1121 已提交
476

D
dapan1121 已提交
477 478
    tscDebug("tb %s in execBlock list, set to current", pStmt->bInfo.tbFName);

D
stmt  
dapan1121 已提交
479 480 481
    return TSDB_CODE_SUCCESS;
  }

D
dapan 已提交
482
  SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid));
D
stmt  
dapan1121 已提交
483
  if (pCache) {
D
stmt  
dapan1121 已提交
484
    pStmt->bInfo.needParse = false;
D
stmt  
dapan1121 已提交
485

D
dapan1121 已提交
486 487 488
    pStmt->bInfo.tbUid = uid;
    pStmt->bInfo.tbSuid = suid;
    pStmt->bInfo.tbType = tableType;
D
stmt  
dapan1121 已提交
489
    pStmt->bInfo.boundTags = pCache->boundTags;
D
dapan1121 已提交
490
    pStmt->bInfo.tagsCached = true;
D
stmt  
dapan1121 已提交
491

D
stmt  
dapan1121 已提交
492
    STableDataBlocks* pNewBlock = NULL;
D
dapan 已提交
493
    STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataBlock, &pNewBlock, uid));
D
stmt  
dapan1121 已提交
494

495 496
    if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
                    POINTER_BYTES)) {
D
stmt  
dapan1121 已提交
497 498
      STMT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
    }
X
Xiaoyu Wang 已提交
499

D
dapan1121 已提交
500 501
    tscDebug("tb %s in sqlBlock list, set to current", pStmt->bInfo.tbFName);

D
stmt  
dapan1121 已提交
502 503 504
    return TSDB_CODE_SUCCESS;
  }

D
stmt  
dapan1121 已提交
505 506
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));

D
stmt  
dapan1121 已提交
507 508 509
  return TSDB_CODE_SUCCESS;
}

D
stmt  
dapan1121 已提交
510 511 512 513 514 515 516 517 518 519 520 521 522 523
int32_t stmtResetStmt(STscStmt* pStmt) {
  STMT_ERR_RET(stmtCleanSQLInfo(pStmt));

  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
  if (NULL == pStmt->sql.pTableCache) {
    terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
    STMT_ERR_RET(terrno);
  }

  pStmt->sql.status = STMT_INIT;

  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
524
TAOS_STMT* stmtInit(STscObj* taos, int64_t reqid) {
X
Xiaoyu Wang 已提交
525
  STscObj*  pObj = (STscObj*)taos;
D
dapan1121 已提交
526 527 528
  STscStmt* pStmt = NULL;

  pStmt = taosMemoryCalloc(1, sizeof(STscStmt));
D
stmt  
dapan1121 已提交
529
  if (NULL == pStmt) {
D
dapan1121 已提交
530 531 532
    terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
    return NULL;
  }
D
stmt  
dapan1121 已提交
533

D
stmt  
dapan1121 已提交
534 535
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
  if (NULL == pStmt->sql.pTableCache) {
D
stmt  
dapan1121 已提交
536 537 538 539
    terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
    taosMemoryFree(pStmt);
    return NULL;
  }
D
stmt  
dapan1121 已提交
540

D
dapan1121 已提交
541
  pStmt->taos = pObj;
D
stmt  
dapan1121 已提交
542
  pStmt->bInfo.needParse = true;
D
stmt  
dapan1121 已提交
543
  pStmt->sql.status = STMT_INIT;
dengyihao's avatar
dengyihao 已提交
544
  pStmt->reqid = reqid;
X
Xiaoyu Wang 已提交
545

546
  STMT_LOG_SEQ(STMT_INIT);
dengyihao's avatar
dengyihao 已提交
547

548 549
  tscDebug("stmt:%p initialized", pStmt);

D
stmt  
dapan1121 已提交
550 551
  return pStmt;
}
D
dapan1121 已提交
552

X
Xiaoyu Wang 已提交
553
int stmtPrepare(TAOS_STMT* stmt, const char* sql, unsigned long length) {
D
stmt  
dapan1121 已提交
554 555
  STscStmt* pStmt = (STscStmt*)stmt;

556
  STMT_DLOG_E("start to prepare");
D
dapan1121 已提交
557

D
stmt  
dapan1121 已提交
558
  if (pStmt->sql.status >= STMT_PREPARE) {
D
stmt  
dapan1121 已提交
559
    STMT_ERR_RET(stmtResetStmt(pStmt));
D
stmt  
dapan1121 已提交
560 561
  }

D
stmt  
dapan1121 已提交
562 563 564 565 566
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_PREPARE));

  if (length <= 0) {
    length = strlen(sql);
  }
X
Xiaoyu Wang 已提交
567

D
stmt  
dapan1121 已提交
568 569
  pStmt->sql.sqlStr = strndup(sql, length);
  pStmt->sql.sqlLen = length;
D
stmt  
dapan1121 已提交
570 571 572 573

  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
574
int stmtSetTbName(TAOS_STMT* stmt, const char* tbName) {
D
stmt  
dapan1121 已提交
575 576
  STscStmt* pStmt = (STscStmt*)stmt;

577
  STMT_DLOG("start to set tbName: %s", tbName);
D
dapan1121 已提交
578

D
stmt  
dapan1121 已提交
579
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTBNAME));
D
stmt  
dapan1121 已提交
580

D
dapan1121 已提交
581 582 583 584 585 586 587
  int32_t insert = 0;
  stmtIsInsert(stmt, &insert);
  if (0 == insert) {
    tscError("set tb name not available for none insert statement");
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
  }

588
  STMT_ERR_RET(stmtCreateRequest(pStmt));
589 590 591

  STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
                            pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
D
dapan 已提交
592
  tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName);
593

D
stmt  
dapan1121 已提交
594
  STMT_ERR_RET(stmtGetFromCache(pStmt));
D
stmt  
dapan1121 已提交
595

D
stmt  
dapan1121 已提交
596
  if (pStmt->bInfo.needParse) {
D
dapan 已提交
597 598
    strncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName) - 1);
    pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
D
dapan1121 已提交
599 600

    STMT_ERR_RET(stmtParseSql(pStmt));
D
stmt  
dapan1121 已提交
601 602
  }

D
stmt  
dapan1121 已提交
603 604 605
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
606
int stmtSetTbTags(TAOS_STMT* stmt, TAOS_MULTI_BIND* tags) {
D
stmt  
dapan1121 已提交
607
  STscStmt* pStmt = (STscStmt*)stmt;
D
dapan1121 已提交
608

609
  STMT_DLOG_E("start to set tbTags");
D
dapan1121 已提交
610

D
stmt  
dapan1121 已提交
611
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
D
stmt  
dapan1121 已提交
612

D
dapan 已提交
613 614 615
  if (pStmt->bInfo.inExecCache) {
    return TSDB_CODE_SUCCESS;
  }
D
dapan 已提交
616

617 618
  STableDataBlocks** pDataBlock =
      (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
D
stmt  
dapan1121 已提交
619
  if (NULL == pDataBlock) {
D
dapan 已提交
620
    tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
D
stmt  
dapan1121 已提交
621 622
    STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
  }
X
Xiaoyu Wang 已提交
623

D
dapan1121 已提交
624
  tscDebug("start to bind stmt tag values");
H
Hongze Cheng 已提交
625 626 627
  STMT_ERR_RET(qBindStmtTagsValue(*pDataBlock, pStmt->bInfo.boundTags, pStmt->bInfo.tbSuid, pStmt->bInfo.stbFName,
                                  pStmt->bInfo.sname.tname, tags, pStmt->exec.pRequest->msgBuf,
                                  pStmt->exec.pRequest->msgBufLen));
D
stmt  
dapan1121 已提交
628

629 630
  pStmt->exec.autoCreateTbl = true;

D
stmt  
dapan1121 已提交
631 632 633
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
634
int stmtFetchTagFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields) {
D
stmt  
dapan1121 已提交
635 636 637 638 639
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
    tscError("invalid operation to get query tag fileds");
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
  }

640 641
  STableDataBlocks** pDataBlock =
      (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
D
stmt  
dapan1121 已提交
642
  if (NULL == pDataBlock) {
D
dapan 已提交
643
    tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
D
stmt  
dapan1121 已提交
644 645 646
    STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
  }

D
stmt  
dapan1121 已提交
647
  STMT_ERR_RET(qBuildStmtTagFields(*pDataBlock, pStmt->bInfo.boundTags, fieldNum, fields));
D
stmt  
dapan1121 已提交
648

D
stmt  
dapan1121 已提交
649 650
  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
651

D
dapan1121 已提交
652
int stmtFetchColFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields) {
D
stmt  
dapan1121 已提交
653 654 655 656 657
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
    tscError("invalid operation to get query column fileds");
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
  }

658 659
  STableDataBlocks** pDataBlock =
      (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
D
stmt  
dapan1121 已提交
660
  if (NULL == pDataBlock) {
D
dapan 已提交
661
    tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
D
stmt  
dapan1121 已提交
662 663 664
    STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
  }

D
stmt  
dapan1121 已提交
665
  STMT_ERR_RET(qBuildStmtColFields(*pDataBlock, fieldNum, fields));
D
stmt  
dapan1121 已提交
666

X
Xiaoyu Wang 已提交
667
  return TSDB_CODE_SUCCESS;
D
stmt  
dapan1121 已提交
668 669
}

X
Xiaoyu Wang 已提交
670
int stmtBindBatch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind, int32_t colIdx) {
D
stmt  
dapan1121 已提交
671 672
  STscStmt* pStmt = (STscStmt*)stmt;

673
  STMT_DLOG("start to bind stmt data, colIdx: %d", colIdx);
D
dapan1121 已提交
674

D
dapan1121 已提交
675 676
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_BIND));

X
Xiaoyu Wang 已提交
677 678
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
D
stmt  
dapan1121 已提交
679
    pStmt->bInfo.needParse = false;
D
stmt  
dapan1121 已提交
680
  }
D
stmt  
dapan1121 已提交
681

D
dapan1121 已提交
682 683 684 685
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
    taos_free_result(pStmt->exec.pRequest);
    pStmt->exec.pRequest = NULL;
  }
X
Xiaoyu Wang 已提交
686

687
  STMT_ERR_RET(stmtCreateRequest(pStmt));
D
stmt  
dapan1121 已提交
688

D
stmt  
dapan1121 已提交
689
  if (pStmt->bInfo.needParse) {
D
stmt  
dapan1121 已提交
690 691
    STMT_ERR_RET(stmtParseSql(pStmt));
  }
D
stmt  
dapan1121 已提交
692

D
dapan1121 已提交
693
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
694 695
    STMT_ERR_RET(qStmtBindParams(pStmt->sql.pQuery, bind, colIdx));

D
dapan1121 已提交
696 697 698 699 700 701 702 703 704 705 706
    SParseContext ctx = {.requestId = pStmt->exec.pRequest->requestId,
                         .acctId = pStmt->taos->acctId,
                         .db = pStmt->exec.pRequest->pDb,
                         .topicQuery = false,
                         .pSql = pStmt->sql.sqlStr,
                         .sqlLen = pStmt->sql.sqlLen,
                         .pMsg = pStmt->exec.pRequest->msgBuf,
                         .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
                         .pTransporter = pStmt->taos->pAppInfo->pTransporter,
                         .pStmtCb = NULL,
                         .pUser = pStmt->taos->user};
D
dapan1121 已提交
707 708
    ctx.mgmtEpSet = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
    STMT_ERR_RET(catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &ctx.pCatalog));
709

D
dapan1121 已提交
710
    STMT_ERR_RET(qStmtParseQuerySql(&ctx, pStmt->sql.pQuery));
X
Xiaoyu Wang 已提交
711

D
dapan1121 已提交
712
    if (pStmt->sql.pQuery->haveResultSet) {
713 714
      setResSchemaInfo(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->pResSchema,
                       pStmt->sql.pQuery->numOfResCols);
D
dapan1121 已提交
715
      taosMemoryFreeClear(pStmt->sql.pQuery->pResSchema);
D
dapan1121 已提交
716 717
      setResPrecision(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->precision);
    }
718

D
dapan1121 已提交
719
    TSWAP(pStmt->exec.pRequest->dbList, pStmt->sql.pQuery->pDbList);
720
    TSWAP(pStmt->exec.pRequest->tableList, pStmt->sql.pQuery->pTableList);
D
dapan1121 已提交
721
    TSWAP(pStmt->exec.pRequest->targetTableList, pStmt->sql.pQuery->pTargetTableList);
D
dapan1121 已提交
722

723 724 725
    // if (STMT_TYPE_QUERY == pStmt->sql.queryRes) {
    //   STMT_ERR_RET(stmtRestoreQueryFields(pStmt));
    // }
D
dapan1121 已提交
726

727
    // STMT_ERR_RET(stmtBackupQueryFields(pStmt));
D
dapan1121 已提交
728

D
dapan1121 已提交
729
    return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
730
  }
731 732 733

  STableDataBlocks** pDataBlock =
      (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
D
stmt  
dapan1121 已提交
734
  if (NULL == pDataBlock) {
D
dapan 已提交
735
    tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
D
stmt  
dapan1121 已提交
736 737
    STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
  }
D
stmt  
dapan1121 已提交
738 739

  if (colIdx < 0) {
D
dapan1121 已提交
740 741 742 743 744
    int32_t code = qBindStmtColsValue(*pDataBlock, bind, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen);
    if (code) {
      tscError("qBindStmtColsValue failed, error:%s", tstrerror(code));
      STMT_ERR_RET(code);
    }
D
stmt  
dapan1121 已提交
745 746 747 748 749 750 751
  } else {
    if (colIdx != (pStmt->bInfo.sBindLastIdx + 1) && colIdx != 0) {
      tscError("bind column index not in sequence");
      STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
    }

    pStmt->bInfo.sBindLastIdx = colIdx;
X
Xiaoyu Wang 已提交
752

D
stmt  
dapan1121 已提交
753 754 755
    if (0 == colIdx) {
      pStmt->bInfo.sBindRowNum = bind->num;
    }
X
Xiaoyu Wang 已提交
756 757 758

    qBindStmtSingleColValue(*pDataBlock, bind, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen, colIdx,
                            pStmt->bInfo.sBindRowNum);
D
stmt  
dapan1121 已提交
759
  }
X
Xiaoyu Wang 已提交
760

D
stmt  
dapan1121 已提交
761
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
762 763
}

X
Xiaoyu Wang 已提交
764
int stmtAddBatch(TAOS_STMT* stmt) {
D
stmt  
dapan1121 已提交
765 766
  STscStmt* pStmt = (STscStmt*)stmt;

767
  STMT_DLOG_E("start to add batch");
D
dapan1121 已提交
768

D
stmt  
dapan1121 已提交
769
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_ADD_BATCH));
D
stmt  
dapan1121 已提交
770

D
stmt  
dapan1121 已提交
771
  STMT_ERR_RET(stmtCacheBlock(pStmt));
X
Xiaoyu Wang 已提交
772

D
stmt  
dapan1121 已提交
773 774 775
  return TSDB_CODE_SUCCESS;
}

776
int stmtUpdateTableUid(STscStmt* pStmt, SSubmitRsp* pRsp) {
D
dapan1121 已提交
777
  tscDebug("stmt start to update tbUid, blockNum: %d", pRsp->nBlocks);
D
dapan1121 已提交
778

779 780 781
  int32_t code = 0;
  int32_t finalCode = 0;
  size_t  keyLen = 0;
782
  STableDataBlocks** pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
D
dapan 已提交
783
  while (pIter) {
784 785 786 787
    STableDataBlocks* pBlock = *pIter;
    char*             key = taosHashGetKey(pIter, &keyLen);

    STableMeta* pMeta = qGetTableMetaInDataBlock(pBlock);
D
dapan 已提交
788 789 790 791 792
    if (pMeta->uid) {
      pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
      continue;
    }

793 794
    SSubmitBlkRsp* blkRsp = NULL;
    int32_t        i = 0;
D
dapan 已提交
795 796 797 798 799
    for (; i < pRsp->nBlocks; ++i) {
      blkRsp = pRsp->pBlocks + i;
      if (strlen(blkRsp->tblFName) != keyLen) {
        continue;
      }
800

D
dapan 已提交
801 802 803
      if (strncmp(blkRsp->tblFName, key, keyLen)) {
        continue;
      }
804

D
dapan 已提交
805 806 807 808
      break;
    }

    if (i < pRsp->nBlocks) {
809 810 811
      tscDebug("auto created table %s uid updated from %" PRIx64 " to %" PRIx64, blkRsp->tblFName, pMeta->uid,
               blkRsp->uid);

D
dapan 已提交
812 813 814
      pMeta->uid = blkRsp->uid;
      pStmt->bInfo.tbUid = blkRsp->uid;
    } else {
815 816
      tscDebug("table %s not found in submit rsp, will update from catalog", pStmt->bInfo.tbFName);
      if (NULL == pStmt->pCatalog) {
817 818 819 820 821 822
        code = catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog);
        if (code) {
          pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
          finalCode = code;
          continue;
        }
823 824
      }

825 826 827 828 829 830
      code = stmtCreateRequest(pStmt);
      if (code) {
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
        finalCode = code;
        continue;
      }
831 832 833 834 835 836 837 838 839 840

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

      taos_free_result(pStmt->exec.pRequest);
      pStmt->exec.pRequest = NULL;
841 842 843 844

      if (code || NULL == pTableMeta) {
        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
        finalCode = code;
D
dapan1121 已提交
845
        taosMemoryFree(pTableMeta);
846
        continue;
847 848 849 850
      }

      pMeta->uid = pTableMeta->uid;
      pStmt->bInfo.tbUid = pTableMeta->uid;
D
dapan1121 已提交
851
      taosMemoryFree(pTableMeta);      
D
dapan 已提交
852 853 854 855 856
    }

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

857
  return finalCode;
D
dapan 已提交
858 859
}

860 861 862 863 864
int stmtExec(TAOS_STMT* stmt) {
  STscStmt*   pStmt = (STscStmt*)stmt;
  int32_t     code = 0;
  SSubmitRsp* pRsp = NULL;
  bool        autoCreateTbl = pStmt->exec.autoCreateTbl;
D
stmt  
dapan1121 已提交
865

866
  STMT_DLOG_E("start to exec");
D
dapan1121 已提交
867

D
stmt  
dapan1121 已提交
868
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
D
stmt  
dapan1121 已提交
869

D
dapan1121 已提交
870
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
871
    launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
D
dapan1121 已提交
872
  } else {
D
dapan1121 已提交
873
    STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pBlockHash));
874
    launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, (autoCreateTbl ? (void**)&pRsp : NULL));
D
dapan1121 已提交
875
  }
D
fix bug  
dapan1121 已提交
876 877 878 879 880 881

  if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
    code = refreshMeta(pStmt->exec.pRequest->pTscObj, pStmt->exec.pRequest);
    if (code) {
      pStmt->exec.pRequest->code = code;
    } else {
D
dapan1121 已提交
882
      tFreeSSubmitRsp(pRsp);
D
fix bug  
dapan1121 已提交
883 884 885 886
      STMT_ERR_RET(stmtResetStmt(pStmt));
      STMT_ERR_RET(TSDB_CODE_NEED_RETRY);
    }
  }
X
Xiaoyu Wang 已提交
887

D
stmt  
dapan1121 已提交
888
  STMT_ERR_JRET(pStmt->exec.pRequest->code);
D
stmt  
dapan1121 已提交
889

D
dapan1121 已提交
890 891
  pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
  pStmt->affectedRows += pStmt->exec.affectedRows;
D
stmt  
dapan1121 已提交
892

D
stmt  
dapan1121 已提交
893 894
_return:

D
dapan1121 已提交
895
  stmtCleanExecInfo(pStmt, (code ? false : true), false);
D
dapan 已提交
896 897 898 899

  if (TSDB_CODE_SUCCESS == code && autoCreateTbl) {
    if (NULL == pRsp) {
      tscError("no submit resp got for auto create table");
D
dapan1121 已提交
900 901 902
      code = TSDB_CODE_TSC_APP_ERROR;
    } else {
      code = stmtUpdateTableUid(pStmt, pRsp);
D
dapan 已提交
903 904
    }
  }
X
Xiaoyu Wang 已提交
905

D
dapan1121 已提交
906
  tFreeSSubmitRsp(pRsp);
907

D
stmt  
dapan1121 已提交
908
  ++pStmt->sql.runTimes;
X
Xiaoyu Wang 已提交
909

D
stmt  
dapan1121 已提交
910
  STMT_RET(code);
D
stmt  
dapan1121 已提交
911 912
}

X
Xiaoyu Wang 已提交
913
int stmtClose(TAOS_STMT* stmt) {
D
stmt  
dapan1121 已提交
914 915
  STscStmt* pStmt = (STscStmt*)stmt;

D
dapan1121 已提交
916
  stmtCleanSQLInfo(pStmt);
D
dapan1121 已提交
917
  taosMemoryFree(stmt);
D
dapan1121 已提交
918 919

  return TSDB_CODE_SUCCESS;
D
stmt  
dapan1121 已提交
920 921
}

X
Xiaoyu Wang 已提交
922
const char* stmtErrstr(TAOS_STMT* stmt) {
D
stmt  
dapan1121 已提交
923
  STscStmt* pStmt = (STscStmt*)stmt;
D
stmt  
dapan1121 已提交
924

D
fix bug  
dapan1121 已提交
925
  if (stmt == NULL || NULL == pStmt->exec.pRequest) {
X
Xiaoyu Wang 已提交
926
    return (char*)tstrerror(terrno);
D
stmt  
dapan1121 已提交
927 928
  }

D
fix bug  
dapan1121 已提交
929
  pStmt->exec.pRequest->code = terrno;
D
stmt  
dapan1121 已提交
930

D
stmt  
dapan1121 已提交
931
  return taos_errstr(pStmt->exec.pRequest);
D
stmt  
dapan1121 已提交
932 933
}

X
Xiaoyu Wang 已提交
934
int stmtAffectedRows(TAOS_STMT* stmt) { return ((STscStmt*)stmt)->affectedRows; }
D
stmt  
dapan1121 已提交
935

X
Xiaoyu Wang 已提交
936
int stmtAffectedRowsOnce(TAOS_STMT* stmt) { return ((STscStmt*)stmt)->exec.affectedRows; }
D
dapan1121 已提交
937

X
Xiaoyu Wang 已提交
938
int stmtIsInsert(TAOS_STMT* stmt, int* insert) {
D
stmt  
dapan1121 已提交
939 940
  STscStmt* pStmt = (STscStmt*)stmt;

941 942
  STMT_DLOG_E("start is insert");

D
stmt  
dapan1121 已提交
943 944 945
  if (pStmt->sql.type) {
    *insert = (STMT_TYPE_INSERT == pStmt->sql.type || STMT_TYPE_MULTI_INSERT == pStmt->sql.type);
  } else {
946
    *insert = qIsInsertValuesSql(pStmt->sql.sqlStr, pStmt->sql.sqlLen);
D
stmt  
dapan1121 已提交
947
  }
X
Xiaoyu Wang 已提交
948

D
stmt  
dapan1121 已提交
949 950 951
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
952 953 954
int stmtGetTagFields(TAOS_STMT* stmt, int* nums, TAOS_FIELD_E** fields) {
  STscStmt* pStmt = (STscStmt*)stmt;

955 956
  STMT_DLOG_E("start to get tag fields");

D
dapan1121 已提交
957 958 959
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
    STMT_RET(TSDB_CODE_TSC_STMT_API_ERROR);
  }
960

D
dapan1121 已提交
961 962 963 964 965 966 967 968 969 970 971 972
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));

  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
    pStmt->bInfo.needParse = false;
  }

  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
    taos_free_result(pStmt->exec.pRequest);
    pStmt->exec.pRequest = NULL;
  }

973
  STMT_ERR_RET(stmtCreateRequest(pStmt));
D
dapan1121 已提交
974 975 976 977 978 979 980 981 982 983 984 985 986

  if (pStmt->bInfo.needParse) {
    STMT_ERR_RET(stmtParseSql(pStmt));
  }

  STMT_ERR_RET(stmtFetchTagFields(stmt, nums, fields));

  return TSDB_CODE_SUCCESS;
}

int stmtGetColFields(TAOS_STMT* stmt, int* nums, TAOS_FIELD_E** fields) {
  STscStmt* pStmt = (STscStmt*)stmt;

987 988
  STMT_DLOG_E("start to get col fields");

D
dapan1121 已提交
989 990 991
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
    STMT_RET(TSDB_CODE_TSC_STMT_API_ERROR);
  }
992

D
dapan1121 已提交
993 994 995 996 997 998 999 1000 1001 1002 1003 1004
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));

  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
    pStmt->bInfo.needParse = false;
  }

  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
    taos_free_result(pStmt->exec.pRequest);
    pStmt->exec.pRequest = NULL;
  }

1005
  STMT_ERR_RET(stmtCreateRequest(pStmt));
D
dapan1121 已提交
1006 1007 1008 1009 1010 1011 1012 1013 1014 1015

  if (pStmt->bInfo.needParse) {
    STMT_ERR_RET(stmtParseSql(pStmt));
  }

  STMT_ERR_RET(stmtFetchColFields(stmt, nums, fields));

  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
1016
int stmtGetParamNum(TAOS_STMT* stmt, int* nums) {
D
dapan1121 已提交
1017 1018
  STscStmt* pStmt = (STscStmt*)stmt;

1019 1020
  STMT_DLOG_E("start to get param num");

D
dapan1121 已提交
1021 1022
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));

X
Xiaoyu Wang 已提交
1023 1024
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
D
dapan1121 已提交
1025 1026 1027 1028 1029 1030 1031
    pStmt->bInfo.needParse = false;
  }

  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
    taos_free_result(pStmt->exec.pRequest);
    pStmt->exec.pRequest = NULL;
  }
X
Xiaoyu Wang 已提交
1032

1033
  STMT_ERR_RET(stmtCreateRequest(pStmt));
D
dapan1121 已提交
1034 1035 1036 1037 1038
  if (pStmt->bInfo.needParse) {
    STMT_ERR_RET(stmtParseSql(pStmt));
  }

  if (STMT_TYPE_QUERY == pStmt->sql.type) {
D
dapan1121 已提交
1039
    *nums = taosArrayGetSize(pStmt->sql.pQuery->pPlaceholderValues);
D
dapan1121 已提交
1040 1041 1042
  } else {
    STMT_ERR_RET(stmtFetchColFields(stmt, nums, NULL));
  }
X
Xiaoyu Wang 已提交
1043

D
stmt  
dapan1121 已提交
1044 1045 1046
  return TSDB_CODE_SUCCESS;
}

1047
int stmtGetParam(TAOS_STMT* stmt, int idx, int* type, int* bytes) {
D
dapan1121 已提交
1048 1049
  STscStmt* pStmt = (STscStmt*)stmt;

1050 1051
  STMT_DLOG_E("start to get param");

D
dapan1121 已提交
1052 1053 1054
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
    STMT_RET(TSDB_CODE_TSC_STMT_API_ERROR);
  }
1055

D
dapan1121 已提交
1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));

  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
      STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
    pStmt->bInfo.needParse = false;
  }

  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
    taos_free_result(pStmt->exec.pRequest);
    pStmt->exec.pRequest = NULL;
  }

1068
  STMT_ERR_RET(stmtCreateRequest(pStmt));
D
dapan1121 已提交
1069 1070 1071 1072 1073

  if (pStmt->bInfo.needParse) {
    STMT_ERR_RET(stmtParseSql(pStmt));
  }

1074 1075
  int32_t       nums = 0;
  TAOS_FIELD_E* pField = NULL;
D
dapan1121 已提交
1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090
  STMT_ERR_RET(stmtFetchColFields(stmt, &nums, &pField));
  if (idx >= nums) {
    tscError("idx %d is too big", idx);
    taosMemoryFree(pField);
    STMT_ERR_RET(TSDB_CODE_INVALID_PARA);
  }

  *type = pField[idx].type;
  *bytes = pField[idx].bytes;

  taosMemoryFree(pField);

  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
1091
TAOS_RES* stmtUseResult(TAOS_STMT* stmt) {
D
dapan1121 已提交
1092 1093
  STscStmt* pStmt = (STscStmt*)stmt;

1094 1095
  STMT_DLOG_E("start to use result");

D
dapan1121 已提交
1096 1097 1098 1099 1100 1101
  if (STMT_TYPE_QUERY != pStmt->sql.type) {
    tscError("useResult only for query statement");
    return NULL;
  }

  return pStmt->exec.pRequest;
D
stmt  
dapan1121 已提交
1102
}