clientStmt.c 17.4 KB
Newer Older
D
dapan1121 已提交
1 2 3

#include "clientInt.h"
#include "clientLog.h"
D
stmt  
dapan1121 已提交
4
#include "clientStmt.h"
D
dapan1121 已提交
5 6
#include "tdef.h"

D
stmt  
dapan1121 已提交
7
int32_t stmtSwitchStatus(STscStmt* pStmt, STMT_STATUS newStatus) {
D
dapan1121 已提交
8 9
  int32_t code = 0;
  
D
stmt  
dapan1121 已提交
10
  switch (newStatus) {
D
dapan1121 已提交
11 12
    case STMT_PREPARE:
      break;
D
stmt  
dapan1121 已提交
13
    case STMT_SETTBNAME:
D
dapan1121 已提交
14
      if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND) || STMT_STATUS_EQ(BIND_COL)) {
D
dapan1121 已提交
15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41
        code = TSDB_CODE_TSC_STMT_API_ERROR;
      }
      break;
    case STMT_SETTAGS:
      if (STMT_STATUS_NE(SETTBNAME)) {
        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;
      }
      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 已提交
42
      break;
D
dapan1121 已提交
43 44 45 46
    case STMT_EXECUTE:
      if (STMT_STATUS_NE(ADD_BATCH) && STMT_STATUS_NE(FETCH_FIELDS)) {
        code = TSDB_CODE_TSC_STMT_API_ERROR;
      }
D
dapan1121 已提交
47
      break;
D
stmt  
dapan1121 已提交
48
    default:
D
dapan1121 已提交
49
      code = TSDB_CODE_TSC_APP_ERROR;
D
stmt  
dapan1121 已提交
50 51 52
      break;
  }

D
dapan1121 已提交
53
  STMT_ERR_RET(code);
D
stmt  
dapan1121 已提交
54 55 56 57 58 59 60

  pStmt->sql.status = newStatus;

  return TSDB_CODE_SUCCESS;
}


D
stmt  
dapan1121 已提交
61 62 63
int32_t stmtGetTbName(TAOS_STMT *stmt, char **tbName) {
  STscStmt* pStmt = (STscStmt*)stmt;

D
stmt  
dapan1121 已提交
64
  pStmt->sql.type = STMT_TYPE_MULTI_INSERT;
D
stmt  
dapan1121 已提交
65
  
D
stmt  
dapan1121 已提交
66
  if (NULL == pStmt->bInfo.tbName) {
D
stmt  
dapan1121 已提交
67 68 69 70
    tscError("no table name set");
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_TBNAME_ERROR);
  }

D
stmt  
dapan1121 已提交
71
  *tbName = pStmt->bInfo.tbName;
D
stmt  
dapan1121 已提交
72 73 74 75 76 77 78

  return TSDB_CODE_SUCCESS;
}

int32_t stmtSetBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags) {
  STscStmt* pStmt = (STscStmt*)stmt;

D
stmt  
dapan1121 已提交
79 80 81 82
  pStmt->bInfo.tbUid = pTableMeta->uid;
  pStmt->bInfo.tbSuid = pTableMeta->suid;
  pStmt->bInfo.tbType = pTableMeta->tableType;
  pStmt->bInfo.boundTags = tags;
D
stmt  
dapan1121 已提交
83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100

  return TSDB_CODE_SUCCESS;
}

int32_t stmtSetExecInfo(TAOS_STMT* stmt, SHashObj* pVgHash, SHashObj* pBlockHash) {
  STscStmt* pStmt = (STscStmt*)stmt;

  pStmt->exec.pVgHash = pVgHash;
  pStmt->exec.pBlockHash = pBlockHash;

  return TSDB_CODE_SUCCESS;
}

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

  *pVgHash = pStmt->exec.pVgHash;
  *pBlockHash = pStmt->exec.pBlockHash;
D
stmt  
dapan1121 已提交
101 102 103 104

  return TSDB_CODE_SUCCESS;
}

D
stmt  
dapan1121 已提交
105 106 107 108
int32_t stmtCacheBlock(STscStmt *pStmt) {
  if (pStmt->sql.type != STMT_TYPE_MULTI_INSERT) {
    return TSDB_CODE_SUCCESS;
  }
D
stmt  
dapan1121 已提交
109

D
dapan1121 已提交
110 111
  uint64_t uid = pStmt->bInfo.tbUid; 
  uint64_t tuid = (TSDB_CHILD_TABLE == pStmt->bInfo.tbType) ? pStmt->bInfo.tbSuid : uid;
D
stmt  
dapan1121 已提交
112

D
dapan1121 已提交
113
  if (taosHashGet(pStmt->sql.pTableCache, &tuid, sizeof(tuid))) {
D
stmt  
dapan1121 已提交
114 115 116
    return TSDB_CODE_SUCCESS;
  }

D
stmt  
dapan1121 已提交
117
  STableDataBlocks** pSrc = taosHashGet(pStmt->exec.pBlockHash, &uid, sizeof(uid));
D
stmt  
dapan1121 已提交
118 119
  STableDataBlocks* pDst = NULL;
  
D
stmt  
dapan1121 已提交
120
  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc));
D
stmt  
dapan1121 已提交
121 122 123

  SStmtTableCache cache = {
    .pDataBlock = pDst,
D
stmt  
dapan1121 已提交
124
    .boundTags = pStmt->bInfo.boundTags,
D
stmt  
dapan1121 已提交
125 126
  };

D
dapan1121 已提交
127
  if (taosHashPut(pStmt->sql.pTableCache, &tuid, sizeof(tuid), &cache, sizeof(cache))) {
D
stmt  
dapan1121 已提交
128 129
    return TSDB_CODE_OUT_OF_MEMORY;
  }
D
stmt  
dapan1121 已提交
130

D
stmt  
dapan1121 已提交
131
  pStmt->bInfo.boundTags = NULL;
D
stmt  
dapan1121 已提交
132 133 134 135

  return TSDB_CODE_SUCCESS;
}

D
stmt  
dapan1121 已提交
136 137 138 139 140 141 142 143
int32_t stmtParseSql(STscStmt* pStmt) {
  SStmtCallback stmtCb = {
    .pStmt = pStmt, 
    .getTbNameFn = stmtGetTbName, 
    .setBindInfoFn = stmtSetBindInfo,
    .setExecInfoFn = stmtSetExecInfo,
    .getExecInfoFn = stmtGetExecInfo,
  };
D
stmt  
dapan1121 已提交
144 145 146 147

  if (NULL == pStmt->exec.pRequest) {
    STMT_ERR_RET(buildRequest(pStmt->taos, pStmt->sql.sqlStr, pStmt->sql.sqlLen, &pStmt->exec.pRequest));
  }
D
stmt  
dapan1121 已提交
148 149 150
  
  STMT_ERR_RET(parseSql(pStmt->exec.pRequest, false, &pStmt->sql.pQuery, &stmtCb));

D
stmt  
dapan1121 已提交
151
  pStmt->bInfo.needParse = false;
D
stmt  
dapan1121 已提交
152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170
  
  switch (nodeType(pStmt->sql.pQuery->pRoot)) {
    case QUERY_NODE_VNODE_MODIF_STMT:
      if (0 == pStmt->sql.type) {
        pStmt->sql.type = STMT_TYPE_INSERT;
      }
      break;
    case QUERY_NODE_SELECT_STMT:
      pStmt->sql.type = STMT_TYPE_QUERY;
      break;
    default:
      tscError("not supported stmt type %d", nodeType(pStmt->sql.pQuery->pRoot));
      STMT_ERR_RET(TSDB_CODE_TSC_STMT_CLAUSE_ERROR);
  }

  STMT_ERR_RET(stmtCacheBlock(pStmt));

  return TSDB_CODE_SUCCESS;
}
D
stmt  
dapan1121 已提交
171

D
stmt  
dapan1121 已提交
172
int32_t stmtCleanBindInfo(STscStmt* pStmt) {
D
stmt  
dapan1121 已提交
173 174 175 176
  pStmt->bInfo.tbUid = 0;
  pStmt->bInfo.tbSuid = 0;
  pStmt->bInfo.tbType = 0;
  pStmt->bInfo.needParse = true;
D
stmt  
dapan1121 已提交
177

D
stmt  
dapan1121 已提交
178 179 180
  taosMemoryFreeClear(pStmt->bInfo.tbName);
  destroyBoundColumnInfo(pStmt->bInfo.boundTags);
  taosMemoryFreeClear(pStmt->bInfo.boundTags);
D
stmt  
dapan1121 已提交
181 182

  return TSDB_CODE_SUCCESS;
D
stmt  
dapan1121 已提交
183 184
}

D
dapan1121 已提交
185 186 187 188 189
int32_t stmtCleanExecInfo(STscStmt* pStmt, bool keepTable, bool freeRequest) {
  if (STMT_TYPE_QUERY != pStmt->sql.type || freeRequest) {
    taos_free_result(pStmt->exec.pRequest);
    pStmt->exec.pRequest = NULL;
  }
D
stmt  
dapan1121 已提交
190 191 192 193

  void *pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
  while (pIter) {
    STableDataBlocks* pBlocks = *(STableDataBlocks**)pIter;    
D
stmt  
dapan1121 已提交
194 195
    uint64_t *key = taosHashGetKey(pIter, NULL);
    
D
stmt  
dapan1121 已提交
196
    if (keepTable && (*key == pStmt->bInfo.tbUid)) {
D
stmt  
dapan1121 已提交
197
      STMT_ERR_RET(qResetStmtDataBlock(pBlocks, true));
D
stmt  
dapan1121 已提交
198 199 200 201 202
      
      pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
      continue;
    }

D
stmt  
dapan1121 已提交
203 204
    qFreeStmtDataBlock(pBlocks);
    taosHashRemove(pStmt->exec.pBlockHash, key, sizeof(*key));
D
stmt  
dapan1121 已提交
205 206 207 208 209 210 211 212 213 214

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

  if (keepTable) {
    return TSDB_CODE_SUCCESS;
  }

  taosHashCleanup(pStmt->exec.pBlockHash);
  pStmt->exec.pBlockHash = NULL;
D
stmt  
dapan1121 已提交
215 216 217 218 219 220 221 222 223

  STMT_ERR_RET(stmtCleanBindInfo(pStmt));

  return TSDB_CODE_SUCCESS;
}

int32_t stmtCleanSQLInfo(STscStmt* pStmt) {
  taosMemoryFree(pStmt->sql.sqlStr);
  qDestroyQuery(pStmt->sql.pQuery);
D
dapan1121 已提交
224 225 226
  qDestroyQueryPlan(pStmt->sql.pQueryPlan);
  taosArrayDestroy(pStmt->sql.nodeList);
  
D
stmt  
dapan1121 已提交
227 228
  void *pIter = taosHashIterate(pStmt->sql.pTableCache, NULL);
  while (pIter) {
D
stmt  
dapan1121 已提交
229
    SStmtTableCache* pCache = (SStmtTableCache*)pIter;    
D
stmt  
dapan1121 已提交
230

D
stmt  
dapan1121 已提交
231
    qDestroyStmtDataBlock(pCache->pDataBlock);
D
stmt  
dapan1121 已提交
232
    destroyBoundColumnInfo(pCache->boundTags);
D
stmt  
dapan1121 已提交
233
    
D
stmt  
dapan1121 已提交
234 235 236 237 238 239 240
    pIter = taosHashIterate(pStmt->sql.pTableCache, pIter);
  }
  taosHashCleanup(pStmt->sql.pTableCache);
  pStmt->sql.pTableCache = NULL;

  memset(&pStmt->sql, 0, sizeof(pStmt->sql));

D
dapan1121 已提交
241
  STMT_ERR_RET(stmtCleanExecInfo(pStmt, false, true));
D
stmt  
dapan1121 已提交
242
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));
D
stmt  
dapan1121 已提交
243 244 245 246 247

  return TSDB_CODE_SUCCESS;
}

int32_t stmtGetFromCache(STscStmt* pStmt) {
D
stmt  
dapan1121 已提交
248 249
  pStmt->bInfo.needParse = true;

D
stmt  
dapan1121 已提交
250
  if (NULL == pStmt->sql.pTableCache || taosHashGetSize(pStmt->sql.pTableCache) <= 0) {
D
stmt  
dapan1121 已提交
251 252 253 254 255 256 257 258 259
    return TSDB_CODE_SUCCESS;
  }

  if (NULL == pStmt->pCatalog) {
    STMT_ERR_RET(catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog));
  }

  STableMeta *pTableMeta = NULL;
  SEpSet ep = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
D
stmt  
dapan1121 已提交
260
  STMT_ERR_RET(catalogGetTableMeta(pStmt->pCatalog, pStmt->taos->pAppInfo->pTransporter, &ep, &pStmt->bInfo.sname, &pTableMeta));
D
stmt  
dapan1121 已提交
261

D
stmt  
dapan1121 已提交
262 263
  if (pTableMeta->uid == pStmt->bInfo.tbUid) {
    pStmt->bInfo.needParse = false;
D
stmt  
dapan1121 已提交
264
    
D
stmt  
dapan1121 已提交
265 266 267
    return TSDB_CODE_SUCCESS;
  }

D
stmt  
dapan1121 已提交
268 269 270 271 272 273 274
  if (taosHashGet(pStmt->exec.pBlockHash, &pTableMeta->uid, sizeof(pTableMeta->uid))) {
    SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &pTableMeta->uid, sizeof(pTableMeta->uid));
    if (NULL == pCache) {
      tscError("table uid %" PRIx64 "found in exec blockHash, but not in sql blockHash", pTableMeta->uid);
      STMT_ERR_RET(TSDB_CODE_TSC_APP_ERROR);
    }
    
D
stmt  
dapan1121 已提交
275
    pStmt->bInfo.needParse = false;
D
stmt  
dapan1121 已提交
276
    
D
stmt  
dapan1121 已提交
277 278 279 280
    pStmt->bInfo.tbUid = pTableMeta->uid;
    pStmt->bInfo.tbSuid = pTableMeta->suid;
    pStmt->bInfo.tbType = pTableMeta->tableType;
    pStmt->bInfo.boundTags = pCache->boundTags;
D
stmt  
dapan1121 已提交
281 282 283 284
    
    return TSDB_CODE_SUCCESS;
  }

D
stmt  
dapan1121 已提交
285 286
  SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &pTableMeta->uid, sizeof(pTableMeta->uid));
  if (pCache) {
D
stmt  
dapan1121 已提交
287
    pStmt->bInfo.needParse = false;
D
stmt  
dapan1121 已提交
288

D
stmt  
dapan1121 已提交
289 290 291 292
    pStmt->bInfo.tbUid = pTableMeta->uid;
    pStmt->bInfo.tbSuid = pTableMeta->suid;
    pStmt->bInfo.tbType = pTableMeta->tableType;
    pStmt->bInfo.boundTags = pCache->boundTags;
D
stmt  
dapan1121 已提交
293

D
stmt  
dapan1121 已提交
294
    STableDataBlocks* pNewBlock = NULL;
D
stmt  
dapan1121 已提交
295
    STMT_ERR_RET(qRebuildStmtDataBlock(&pNewBlock, pCache->pDataBlock));
D
stmt  
dapan1121 已提交
296

D
stmt  
dapan1121 已提交
297
    if (taosHashPut(pStmt->exec.pBlockHash, &pStmt->bInfo.tbUid, sizeof(pStmt->bInfo.tbUid), &pNewBlock, POINTER_BYTES)) {
D
stmt  
dapan1121 已提交
298 299
      STMT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
    }
D
stmt  
dapan1121 已提交
300 301 302 303
    
    return TSDB_CODE_SUCCESS;
  }

D
stmt  
dapan1121 已提交
304 305
  STMT_ERR_RET(stmtCleanBindInfo(pStmt));

D
stmt  
dapan1121 已提交
306 307 308
  return TSDB_CODE_SUCCESS;
}

D
stmt  
dapan1121 已提交
309 310 311 312 313 314 315 316 317 318 319 320 321 322
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;
}

D
stmt  
dapan1121 已提交
323

D
dapan1121 已提交
324 325 326 327 328
TAOS_STMT *stmtInit(TAOS *taos) {
  STscObj* pObj = (STscObj*)taos;
  STscStmt* pStmt = NULL;

  pStmt = taosMemoryCalloc(1, sizeof(STscStmt));
D
stmt  
dapan1121 已提交
329
  if (NULL == pStmt) {
D
dapan1121 已提交
330 331 332
    terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
    return NULL;
  }
D
stmt  
dapan1121 已提交
333

D
stmt  
dapan1121 已提交
334 335
  pStmt->sql.pTableCache = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
  if (NULL == pStmt->sql.pTableCache) {
D
stmt  
dapan1121 已提交
336 337 338 339
    terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
    taosMemoryFree(pStmt);
    return NULL;
  }
D
stmt  
dapan1121 已提交
340

D
dapan1121 已提交
341
  pStmt->taos = pObj;
D
stmt  
dapan1121 已提交
342
  pStmt->bInfo.needParse = true;
D
stmt  
dapan1121 已提交
343
  pStmt->sql.status = STMT_INIT;
D
stmt  
dapan1121 已提交
344
  
D
stmt  
dapan1121 已提交
345 346
  return pStmt;
}
D
dapan1121 已提交
347

D
stmt  
dapan1121 已提交
348 349 350
int stmtPrepare(TAOS_STMT *stmt, const char *sql, unsigned long length) {
  STscStmt* pStmt = (STscStmt*)stmt;

D
stmt  
dapan1121 已提交
351
  if (pStmt->sql.status >= STMT_PREPARE) {
D
stmt  
dapan1121 已提交
352
    STMT_ERR_RET(stmtResetStmt(pStmt));
D
stmt  
dapan1121 已提交
353 354
  }

D
stmt  
dapan1121 已提交
355 356 357 358 359
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_PREPARE));

  if (length <= 0) {
    length = strlen(sql);
  }
D
stmt  
dapan1121 已提交
360
  
D
stmt  
dapan1121 已提交
361 362
  pStmt->sql.sqlStr = strndup(sql, length);
  pStmt->sql.sqlLen = length;
D
stmt  
dapan1121 已提交
363 364 365 366 367

  return TSDB_CODE_SUCCESS;
}


D
stmt  
dapan1121 已提交
368
int stmtSetTbName(TAOS_STMT *stmt, const char *tbName) {
D
stmt  
dapan1121 已提交
369 370
  STscStmt* pStmt = (STscStmt*)stmt;

D
stmt  
dapan1121 已提交
371
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTBNAME));
D
stmt  
dapan1121 已提交
372

D
dapan1121 已提交
373 374 375 376 377 378 379
  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);
  }

D
stmt  
dapan1121 已提交
380 381
  if (NULL == pStmt->exec.pRequest) {
    STMT_ERR_RET(buildRequest(pStmt->taos, pStmt->sql.sqlStr, pStmt->sql.sqlLen, &pStmt->exec.pRequest));
D
dapan1121 已提交
382
  }
D
stmt  
dapan1121 已提交
383
  
D
stmt  
dapan1121 已提交
384
  STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
D
stmt  
dapan1121 已提交
385
    
D
stmt  
dapan1121 已提交
386
  STMT_ERR_RET(stmtGetFromCache(pStmt));
D
stmt  
dapan1121 已提交
387

D
stmt  
dapan1121 已提交
388 389 390 391 392
  if (pStmt->bInfo.needParse) {
    taosMemoryFree(pStmt->bInfo.tbName);
    pStmt->bInfo.tbName = strdup(tbName);
  }

D
stmt  
dapan1121 已提交
393 394 395
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
396
int stmtSetTbTags(TAOS_STMT *stmt, TAOS_MULTI_BIND *tags) {
D
stmt  
dapan1121 已提交
397
  STscStmt* pStmt = (STscStmt*)stmt;
D
dapan1121 已提交
398

D
stmt  
dapan1121 已提交
399
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
D
stmt  
dapan1121 已提交
400

D
stmt  
dapan1121 已提交
401
  if (pStmt->bInfo.needParse) {
D
stmt  
dapan1121 已提交
402 403 404
    STMT_ERR_RET(stmtParseSql(pStmt));
  }

D
stmt  
dapan1121 已提交
405
  STableDataBlocks **pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, (const char*)&pStmt->bInfo.tbUid, sizeof(pStmt->bInfo.tbUid));
D
stmt  
dapan1121 已提交
406
  if (NULL == pDataBlock) {
D
stmt  
dapan1121 已提交
407
    tscError("table uid %" PRIx64 "not found in exec blockHash", pStmt->bInfo.tbUid);
D
stmt  
dapan1121 已提交
408 409 410
    STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
  }
  
D
stmt  
dapan1121 已提交
411
  STMT_ERR_RET(qBindStmtTagsValue(*pDataBlock, pStmt->bInfo.boundTags, pStmt->bInfo.tbSuid, &pStmt->bInfo.sname, tags, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
D
stmt  
dapan1121 已提交
412

D
stmt  
dapan1121 已提交
413 414 415 416
  return TSDB_CODE_SUCCESS;
}


D
dapan1121 已提交
417
int32_t stmtFetchTagFields(STscStmt* pStmt, int32_t *fieldNum, TAOS_FIELD** fields) {
D
stmt  
dapan1121 已提交
418 419 420 421 422
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
    tscError("invalid operation to get query tag fileds");
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
  }

D
stmt  
dapan1121 已提交
423
  STableDataBlocks **pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, (const char*)&pStmt->bInfo.tbUid, sizeof(pStmt->bInfo.tbUid));
D
stmt  
dapan1121 已提交
424
  if (NULL == pDataBlock) {
D
stmt  
dapan1121 已提交
425
    tscError("table uid %" PRIx64 "not found in exec blockHash", pStmt->bInfo.tbUid);
D
stmt  
dapan1121 已提交
426 427 428
    STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
  }

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

D
stmt  
dapan1121 已提交
431 432
  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
433

D
dapan1121 已提交
434
int32_t stmtFetchColFields(STscStmt* pStmt, int32_t *fieldNum, TAOS_FIELD** fields) {
D
stmt  
dapan1121 已提交
435 436 437 438 439
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
    tscError("invalid operation to get query column fileds");
    STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
  }

D
stmt  
dapan1121 已提交
440
  STableDataBlocks **pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, (const char*)&pStmt->bInfo.tbUid, sizeof(pStmt->bInfo.tbUid));
D
stmt  
dapan1121 已提交
441
  if (NULL == pDataBlock) {
D
stmt  
dapan1121 已提交
442
    tscError("table uid %" PRIx64 "not found in exec blockHash", pStmt->bInfo.tbUid);
D
stmt  
dapan1121 已提交
443 444 445
    STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
  }

D
stmt  
dapan1121 已提交
446
  STMT_ERR_RET(qBuildStmtColFields(*pDataBlock, fieldNum, fields));
D
stmt  
dapan1121 已提交
447 448 449 450

  return TSDB_CODE_SUCCESS;  
}

D
dapan1121 已提交
451
int stmtBindBatch(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind, int32_t colIdx) {
D
stmt  
dapan1121 已提交
452 453
  STscStmt* pStmt = (STscStmt*)stmt;

D
stmt  
dapan1121 已提交
454
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_BIND));
D
stmt  
dapan1121 已提交
455

D
stmt  
dapan1121 已提交
456 457
  if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 && STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
    pStmt->bInfo.needParse = false;
D
stmt  
dapan1121 已提交
458
  }
D
stmt  
dapan1121 已提交
459

D
dapan1121 已提交
460 461 462 463 464
  if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
    taos_free_result(pStmt->exec.pRequest);
    pStmt->exec.pRequest = NULL;
  }
  
D
stmt  
dapan1121 已提交
465 466
  if (NULL == pStmt->exec.pRequest) {
    STMT_ERR_RET(buildRequest(pStmt->taos, pStmt->sql.sqlStr, pStmt->sql.sqlLen, &pStmt->exec.pRequest));
D
stmt  
dapan1121 已提交
467 468
  }

D
stmt  
dapan1121 已提交
469
  if (pStmt->bInfo.needParse) {
D
stmt  
dapan1121 已提交
470 471
    STMT_ERR_RET(stmtParseSql(pStmt));
  }
D
stmt  
dapan1121 已提交
472

D
dapan1121 已提交
473 474 475 476 477 478 479 480 481 482
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
    if (NULL == pStmt->sql.pQueryPlan) {
      STMT_ERR_RET(getQueryPlan(pStmt->exec.pRequest, pStmt->sql.pQuery, &pStmt->sql.nodeList));
      pStmt->sql.pQueryPlan = pStmt->exec.pRequest->body.pDag;
      pStmt->exec.pRequest->body.pDag = NULL;
    }
    
    STMT_RET(qStmtBindParam(pStmt->sql.pQueryPlan, bind, colIdx));
  }
  
D
stmt  
dapan1121 已提交
483
  STableDataBlocks **pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, (const char*)&pStmt->bInfo.tbUid, sizeof(pStmt->bInfo.tbUid));
D
stmt  
dapan1121 已提交
484
  if (NULL == pDataBlock) {
D
stmt  
dapan1121 已提交
485
    tscError("table uid %" PRIx64 "not found in exec blockHash", pStmt->bInfo.tbUid);
D
stmt  
dapan1121 已提交
486 487
    STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
  }
D
stmt  
dapan1121 已提交
488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504

  if (colIdx < 0) {
    qBindStmtColsValue(*pDataBlock, bind, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen);
  } 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;
    
    if (0 == colIdx) {
      pStmt->bInfo.sBindRowNum = bind->num;
    }
    
    qBindStmtSingleColValue(*pDataBlock, bind, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen, colIdx, pStmt->bInfo.sBindRowNum);
  }
D
stmt  
dapan1121 已提交
505 506
  
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
507 508
}

D
stmt  
dapan1121 已提交
509 510 511 512

int stmtAddBatch(TAOS_STMT *stmt) {
  STscStmt* pStmt = (STscStmt*)stmt;

D
stmt  
dapan1121 已提交
513
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_ADD_BATCH));
D
stmt  
dapan1121 已提交
514

D
stmt  
dapan1121 已提交
515
  STMT_ERR_RET(stmtCacheBlock(pStmt));
D
stmt  
dapan1121 已提交
516
  
D
stmt  
dapan1121 已提交
517 518 519 520
  return TSDB_CODE_SUCCESS;
}

int stmtExec(TAOS_STMT *stmt) {
D
stmt  
dapan1121 已提交
521 522 523
  STscStmt* pStmt = (STscStmt*)stmt;
  int32_t code = 0;

D
stmt  
dapan1121 已提交
524
  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
D
stmt  
dapan1121 已提交
525

D
dapan1121 已提交
526 527 528 529 530 531 532
  if (STMT_TYPE_QUERY == pStmt->sql.type) {
    scheduleQuery(pStmt->exec.pRequest, pStmt->sql.pQueryPlan, pStmt->sql.nodeList);
  } else {
    STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->exec.pVgHash, pStmt->exec.pBlockHash));
    launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, TSDB_CODE_SUCCESS, true);
  }
  
D
stmt  
dapan1121 已提交
533
  STMT_ERR_JRET(pStmt->exec.pRequest->code);
D
stmt  
dapan1121 已提交
534

D
dapan1121 已提交
535 536
  pStmt->exec.affectedRows = taos_affected_rows(pStmt->exec.pRequest);
  pStmt->affectedRows += pStmt->exec.affectedRows;
D
stmt  
dapan1121 已提交
537

D
stmt  
dapan1121 已提交
538 539
_return:

D
dapan1121 已提交
540
  stmtCleanExecInfo(pStmt, (code ? false : true), false);
D
stmt  
dapan1121 已提交
541 542
  
  ++pStmt->sql.runTimes;
D
stmt  
dapan1121 已提交
543 544
  
  STMT_RET(code);
D
stmt  
dapan1121 已提交
545 546 547
}


D
stmt  
dapan1121 已提交
548
int stmtClose(TAOS_STMT *stmt) {
D
stmt  
dapan1121 已提交
549 550 551
  STscStmt* pStmt = (STscStmt*)stmt;

  STMT_RET(stmtCleanSQLInfo(pStmt));
D
stmt  
dapan1121 已提交
552 553
}

D
stmt  
dapan1121 已提交
554
const char *stmtErrstr(TAOS_STMT *stmt) {
D
stmt  
dapan1121 已提交
555
  STscStmt* pStmt = (STscStmt*)stmt;
D
stmt  
dapan1121 已提交
556

D
stmt  
dapan1121 已提交
557 558 559 560
  if (stmt == NULL) {
    return (char*) tstrerror(terrno);
  }

D
stmt  
dapan1121 已提交
561 562 563 564
  if (pStmt->exec.pRequest) {
    pStmt->exec.pRequest->code = terrno;
  }

D
stmt  
dapan1121 已提交
565
  return taos_errstr(pStmt->exec.pRequest);
D
stmt  
dapan1121 已提交
566 567
}

D
stmt  
dapan1121 已提交
568
int stmtAffectedRows(TAOS_STMT *stmt) {
D
stmt  
dapan1121 已提交
569
  return ((STscStmt*)stmt)->affectedRows;
D
stmt  
dapan1121 已提交
570 571
}

D
dapan1121 已提交
572 573 574 575
int stmtAffectedRowsOnce(TAOS_STMT *stmt) {
  return ((STscStmt*)stmt)->exec.affectedRows;
}

D
stmt  
dapan1121 已提交
576
int stmtIsInsert(TAOS_STMT *stmt, int *insert) {
D
stmt  
dapan1121 已提交
577 578
  STscStmt* pStmt = (STscStmt*)stmt;

D
stmt  
dapan1121 已提交
579 580 581 582 583
  if (pStmt->sql.type) {
    *insert = (STMT_TYPE_INSERT == pStmt->sql.type || STMT_TYPE_MULTI_INSERT == pStmt->sql.type);
  } else {
    *insert = isInsertSql(pStmt->sql.sqlStr, 0);
  }
D
stmt  
dapan1121 已提交
584
  
D
stmt  
dapan1121 已提交
585 586 587 588
  return TSDB_CODE_SUCCESS;
}

int stmtGetParamNum(TAOS_STMT *stmt, int *nums) {
D
dapan1121 已提交
589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607
  STscStmt* pStmt = (STscStmt*)stmt;

  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));

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

  if (STMT_TYPE_QUERY == pStmt->sql.type) {
    if (NULL == pStmt->sql.pQueryPlan) {
      STMT_ERR_RET(getQueryPlan(pStmt->exec.pRequest, pStmt->sql.pQuery, &pStmt->sql.nodeList));
      pStmt->sql.pQueryPlan = pStmt->exec.pRequest->body.pDag;
      pStmt->exec.pRequest->body.pDag = NULL;
    }

    *nums = (pStmt->sql.pQueryPlan->pPlaceholderValues) ? pStmt->sql.pQueryPlan->pPlaceholderValues->length : 0;
  } else {
    STMT_ERR_RET(stmtFetchColFields(stmt, nums, NULL));
  }
D
stmt  
dapan1121 已提交
608
  
D
stmt  
dapan1121 已提交
609 610 611 612
  return TSDB_CODE_SUCCESS;
}

TAOS_RES *stmtUseResult(TAOS_STMT *stmt) {
D
dapan1121 已提交
613 614 615 616 617 618 619 620
  STscStmt* pStmt = (STscStmt*)stmt;

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

  return pStmt->exec.pRequest;
D
stmt  
dapan1121 已提交
621 622 623 624
}