clientMain.c 39.4 KB
Newer Older
L
Liu Jicong 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#include "catalog.h"
17
#include "clientInt.h"
18
#include "clientLog.h"
L
Liu Jicong 已提交
19
#include "clientStmt.h"
L
Liu Jicong 已提交
20
#include "functionMgt.h"
L
Liu Jicong 已提交
21
#include "os.h"
H
Haojun Liao 已提交
22
#include "query.h"
X
Xiaoyu Wang 已提交
23
#include "qworker.h"
L
Liu Jicong 已提交
24
#include "scheduler.h"
25
#include "tglobal.h"
L
Liu Jicong 已提交
26 27 28
#include "tmsg.h"
#include "tref.h"
#include "trpc.h"
S
version  
Shengliang Guan 已提交
29
#include "version.h"
30 31 32 33 34

#define TSC_VAR_NOT_RELEASE 1
#define TSC_VAR_RELEASED    0

static int32_t sentinel = TSC_VAR_NOT_RELEASE;
35
static int32_t createParseContext(const SRequestObj *pRequest, SParseContext **pCxt);
36 37 38 39 40 41 42 43 44 45 46

int taos_options(TSDB_OPTION option, const void *arg, ...) {
  static int32_t lock = 0;

  for (int i = 1; atomic_val_compare_exchange_32(&lock, 0, 1) != 0; ++i) {
    if (i % 1000 == 0) {
      tscInfo("haven't acquire lock after spin %d times.", i);
      sched_yield();
    }
  }

L
Liu Jicong 已提交
47
  int ret = taos_options_imp(option, (const char *)arg);
48 49 50 51 52
  atomic_store_32(&lock, 0);
  return ret;
}
// this function may be called by user or system, or by both simultaneously.
void taos_cleanup(void) {
D
dapan1121 已提交
53
  tscDebug("start to cleanup client environment");
54 55 56 57
  if (atomic_val_compare_exchange_32(&sentinel, TSC_VAR_NOT_RELEASE, TSC_VAR_RELEASED) != TSC_VAR_NOT_RELEASE) {
    return;
  }

H
Haojun Liao 已提交
58 59
  int32_t id = clientReqRefPool;
  clientReqRefPool = -1;
60 61
  taosCloseRef(id);

D
dapan1121 已提交
62 63 64 65
  hbMgrCleanUp();

  catalogDestroy();
  schedulerDestroy();
66

X
Xiaoyu Wang 已提交
67 68
  fmFuncMgtDestroy();
  qCleanupKeywordsTable();
69
  nodesDestroyAllocatorSet();
X
Xiaoyu Wang 已提交
70

71 72
  id = clientConnRefPool;
  clientConnRefPool = -1;
73 74
  taosCloseRef(id);

D
dapan1121 已提交
75 76
  rpcCleanup();
  tscDebug("rpc cleanup");
77

D
dapan1121 已提交
78
  cleanupTaskQueue();
H
Haojun Liao 已提交
79

D
dapan1121 已提交
80
  taosConvDestroy();
dengyihao's avatar
dengyihao 已提交
81

H
Haojun Liao 已提交
82
  tscInfo("all local resources released");
83
  taosCleanupCfg();
84
  taosCloseLog();
85 86
}

L
Liu Jicong 已提交
87
static setConfRet taos_set_config_imp(const char *config) {
88 89 90 91 92
  setConfRet ret = {SET_CONF_RET_SUCC, {0}};
  // TODO: need re-implementation
  return ret;
}

L
Liu Jicong 已提交
93 94
setConfRet taos_set_config(const char *config) {
  // TODO  pthread_mutex_lock(&setConfMutex);
95
  setConfRet ret = taos_set_config_imp(config);
L
Liu Jicong 已提交
96
  //  pthread_mutex_unlock(&setConfMutex);
97 98 99
  return ret;
}

100
TAOS *taos_connect(const char *ip, const char *user, const char *pass, const char *db, uint16_t port) {
S
Shengliang Guan 已提交
101
  tscDebug("try to connect to %s:%u, user:%s db:%s", ip, port, user, db);
H
Haojun Liao 已提交
102 103 104
  if (user == NULL) {
    user = TSDB_DEFAULT_USER;
  }
105

H
Haojun Liao 已提交
106 107 108
  if (pass == NULL) {
    pass = TSDB_DEFAULT_PASS;
  }
109

L
Liu Jicong 已提交
110
  STscObj *pObj = taos_connect_internal(ip, user, pass, NULL, db, port, CONN_TYPE__QUERY);
111
  if (pObj) {
D
dapan1121 已提交
112 113
    int64_t *rid = taosMemoryCalloc(1, sizeof(int64_t));
    *rid = pObj->id;
dengyihao's avatar
dengyihao 已提交
114
    return (TAOS *)rid;
115
  }
L
Liu Jicong 已提交
116

D
dapan1121 已提交
117
  return NULL;
118 119
}

120
void taos_close_internal(void *taos) {
121 122 123 124
  if (taos == NULL) {
    return;
  }

H
Haojun Liao 已提交
125
  STscObj *pTscObj = (STscObj *)taos;
D
dapan1121 已提交
126
  tscDebug("0x%" PRIx64 " try to close connection, numOfReq:%d", pTscObj->id, pTscObj->numOfReqs);
127

D
dapan1121 已提交
128
  taosRemoveRef(clientConnRefPool, pTscObj->id);
129 130
}

131
void taos_close(TAOS *taos) {
D
dapan1121 已提交
132 133 134
  if (taos == NULL) {
    return;
  }
L
Liu Jicong 已提交
135 136

  STscObj *pObj = acquireTscObj(*(int64_t *)taos);
137
  if (NULL == pObj) {
D
dapan1121 已提交
138
    taosMemoryFree(taos);
139 140
    return;
  }
L
Liu Jicong 已提交
141

142
  taos_close_internal(pObj);
L
Liu Jicong 已提交
143
  releaseTscObj(*(int64_t *)taos);
D
dapan1121 已提交
144
  taosMemoryFree(taos);
145 146
}

L
Liu Jicong 已提交
147 148
int taos_errno(TAOS_RES *res) {
  if (res == NULL || TD_RES_TMQ_META(res)) {
149
    if (terrno == TSDB_CODE_RPC_REDIRECT) terrno = TSDB_CODE_QRY_NOT_READY;
150 151 152
    return terrno;
  }

153
  if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) {
154 155 156
    return 0;
  }

157
  return ((SRequestObj *)res)->code == TSDB_CODE_RPC_REDIRECT ? TSDB_CODE_QRY_NOT_READY : ((SRequestObj *)res)->code;
158 159
}

160
const char *taos_errstr(TAOS_RES *res) {
L
Liu Jicong 已提交
161
  if (res == NULL || TD_RES_TMQ_META(res)) {
162
    if (terrno == TSDB_CODE_RPC_REDIRECT) terrno = TSDB_CODE_QRY_NOT_READY;
L
Liu Jicong 已提交
163
    return (const char *)tstrerror(terrno);
164 165
  }

166
  if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) {
167 168 169 170
    return "success";
  }

  SRequestObj *pRequest = (SRequestObj *)res;
171
  if (NULL != pRequest->msgBuf && (strlen(pRequest->msgBuf) > 0 || pRequest->code == TSDB_CODE_RPC_FQDN_ERROR)) {
172 173
    return pRequest->msgBuf;
  } else {
174
    return pRequest->code == TSDB_CODE_RPC_REDIRECT ? (const char *)tstrerror(TSDB_CODE_QRY_NOT_READY)
175
                                                    : (const char *)tstrerror(pRequest->code);
176
  }
177 178 179
}

void taos_free_result(TAOS_RES *res) {
D
stmt  
dapan1121 已提交
180 181 182
  if (NULL == res) {
    return;
  }
183

L
Liu Jicong 已提交
184 185
  if (TD_RES_QUERY(res)) {
    SRequestObj *pRequest = (SRequestObj *)res;
D
dapan1121 已提交
186
    tscDebug("0x%" PRIx64 " taos_free_result start to free query", pRequest->requestId);
L
Liu Jicong 已提交
187
    destroyRequest(pRequest);
188
  } else if (TD_RES_TMQ_METADATA(res)) {
L
Liu Jicong 已提交
189 190 191 192 193 194 195 196 197 198 199 200
    SMqTaosxRspObj *pRsp = (SMqTaosxRspObj *)res;
    if (pRsp->rsp.blockData) taosArrayDestroyP(pRsp->rsp.blockData, taosMemoryFree);
    if (pRsp->rsp.blockDataLen) taosArrayDestroy(pRsp->rsp.blockDataLen);
    if (pRsp->rsp.withTbName) taosArrayDestroyP(pRsp->rsp.blockTbName, taosMemoryFree);
    if (pRsp->rsp.withSchema) taosArrayDestroyP(pRsp->rsp.blockSchema, (FDelete)tDeleteSSchemaWrapper);
    // taosx
    taosArrayDestroy(pRsp->rsp.createTableLen);
    taosArrayDestroyP(pRsp->rsp.createTableReq, taosMemoryFree);

    pRsp->resInfo.pRspMsg = NULL;
    doFreeReqResultInfo(&pRsp->resInfo);
    taosMemoryFree(pRsp);
L
Liu Jicong 已提交
201 202 203 204
  } else if (TD_RES_TMQ(res)) {
    SMqRspObj *pRsp = (SMqRspObj *)res;
    if (pRsp->rsp.blockData) taosArrayDestroyP(pRsp->rsp.blockData, taosMemoryFree);
    if (pRsp->rsp.blockDataLen) taosArrayDestroy(pRsp->rsp.blockDataLen);
L
Liu Jicong 已提交
205 206
    if (pRsp->rsp.withTbName) taosArrayDestroyP(pRsp->rsp.blockTbName, taosMemoryFree);
    if (pRsp->rsp.withSchema) taosArrayDestroyP(pRsp->rsp.blockSchema, (FDelete)tDeleteSSchemaWrapper);
L
Liu Jicong 已提交
207 208
    pRsp->resInfo.pRspMsg = NULL;
    doFreeReqResultInfo(&pRsp->resInfo);
209
    taosMemoryFree(pRsp);
L
Liu Jicong 已提交
210 211 212 213
  } else if (TD_RES_TMQ_META(res)) {
    SMqMetaRspObj *pRspObj = (SMqMetaRspObj *)res;
    taosMemoryFree(pRspObj->metaRsp.metaRsp);
    taosMemoryFree(pRspObj);
L
Liu Jicong 已提交
214
  }
215 216
}

D
dapan1121 已提交
217 218 219 220
void taos_kill_query(TAOS *taos) {
  if (NULL == taos) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
221

dengyihao's avatar
dengyihao 已提交
222
  int64_t  rid = *(int64_t *)taos;
dengyihao's avatar
dengyihao 已提交
223
  STscObj *pTscObj = acquireTscObj(rid);
D
dapan1121 已提交
224 225 226
  if (pTscObj) {
    stopAllRequests(pTscObj->pRequests);
  }
D
dapan1121 已提交
227 228 229
  releaseTscObj(rid);
}

L
Liu Jicong 已提交
230
int taos_field_count(TAOS_RES *res) {
L
Liu Jicong 已提交
231
  if (res == NULL || TD_RES_TMQ_META(res)) {
H
Haojun Liao 已提交
232 233 234
    return 0;
  }

L
Liu Jicong 已提交
235
  SReqResultInfo *pResInfo = tscGetCurResInfo(res);
H
Haojun Liao 已提交
236 237 238
  return pResInfo->numOfCols;
}

L
Liu Jicong 已提交
239
int taos_num_fields(TAOS_RES *res) { return taos_field_count(res); }
H
Haojun Liao 已提交
240 241

TAOS_FIELD *taos_fetch_fields(TAOS_RES *res) {
L
Liu Jicong 已提交
242
  if (taos_num_fields(res) == 0 || TD_RES_TMQ_META(res)) {
H
Haojun Liao 已提交
243 244 245
    return NULL;
  }

L
Liu Jicong 已提交
246
  SReqResultInfo *pResInfo = tscGetCurResInfo(res);
247
  return pResInfo->userFields;
H
Haojun Liao 已提交
248 249
}

250
TAOS_RES *taos_query(TAOS *taos, const char *sql) { return taosQueryImpl(taos, sql, false); }
251

252 253
TAOS_ROW taos_fetch_row(TAOS_RES *res) {
  if (res == NULL) {
254 255 256
    return NULL;
  }

L
Liu Jicong 已提交
257 258 259
  if (TD_RES_QUERY(res)) {
    SRequestObj *pRequest = (SRequestObj *)res;
    if (pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT || pRequest->type == TSDB_SQL_INSERT ||
D
dapan1121 已提交
260
        pRequest->code != TSDB_CODE_SUCCESS || taos_num_fields(res) == 0 || pRequest->killed) {
L
Liu Jicong 已提交
261 262
      return NULL;
    }
D
dapan1121 已提交
263 264

    return doAsyncFetchRows(pRequest, true, true);
265
  } else if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) {
dengyihao's avatar
dengyihao 已提交
266
    SMqRspObj      *msg = ((SMqRspObj *)res);
L
Liu Jicong 已提交
267 268 269 270 271 272
    SReqResultInfo *pResultInfo;
    if (msg->resIter == -1) {
      pResultInfo = tmqGetNextResInfo(res, true);
    } else {
      pResultInfo = tmqGetCurResInfo(res);
    }
273

L
Liu Jicong 已提交
274
    if (pResultInfo->current < pResultInfo->numOfRows) {
L
Liu Jicong 已提交
275 276
      doSetOneRowPtr(pResultInfo);
      pResultInfo->current += 1;
L
Liu Jicong 已提交
277 278
      return pResultInfo->row;
    } else {
L
Liu Jicong 已提交
279 280 281 282 283
      pResultInfo = tmqGetNextResInfo(res, true);
      if (pResultInfo == NULL) return NULL;
      doSetOneRowPtr(pResultInfo);
      pResultInfo->current += 1;
      return pResultInfo->row;
L
Liu Jicong 已提交
284
    }
L
Liu Jicong 已提交
285 286
  } else if (TD_RES_TMQ_META(res)) {
    return NULL;
L
Liu Jicong 已提交
287
  } else {
288
    // assert to avoid un-initialization error
L
Liu Jicong 已提交
289 290 291
    ASSERT(0);
  }
  return NULL;
292
}
H
Haojun Liao 已提交
293

L
Liu Jicong 已提交
294
int taos_print_row(char *str, TAOS_ROW row, TAOS_FIELD *fields, int num_fields) {
H
Haojun Liao 已提交
295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352
  int32_t len = 0;
  for (int i = 0; i < num_fields; ++i) {
    if (i > 0) {
      str[len++] = ' ';
    }

    if (row[i] == NULL) {
      len += sprintf(str + len, "%s", TSDB_DATA_NULL_STR);
      continue;
    }

    switch (fields[i].type) {
      case TSDB_DATA_TYPE_TINYINT:
        len += sprintf(str + len, "%d", *((int8_t *)row[i]));
        break;

      case TSDB_DATA_TYPE_UTINYINT:
        len += sprintf(str + len, "%u", *((uint8_t *)row[i]));
        break;

      case TSDB_DATA_TYPE_SMALLINT:
        len += sprintf(str + len, "%d", *((int16_t *)row[i]));
        break;

      case TSDB_DATA_TYPE_USMALLINT:
        len += sprintf(str + len, "%u", *((uint16_t *)row[i]));
        break;

      case TSDB_DATA_TYPE_INT:
        len += sprintf(str + len, "%d", *((int32_t *)row[i]));
        break;

      case TSDB_DATA_TYPE_UINT:
        len += sprintf(str + len, "%u", *((uint32_t *)row[i]));
        break;

      case TSDB_DATA_TYPE_BIGINT:
        len += sprintf(str + len, "%" PRId64, *((int64_t *)row[i]));
        break;

      case TSDB_DATA_TYPE_UBIGINT:
        len += sprintf(str + len, "%" PRIu64, *((uint64_t *)row[i]));
        break;

      case TSDB_DATA_TYPE_FLOAT: {
        float fv = 0;
        fv = GET_FLOAT_VAL(row[i]);
        len += sprintf(str + len, "%f", fv);
      } break;

      case TSDB_DATA_TYPE_DOUBLE: {
        double dv = 0;
        dv = GET_DOUBLE_VAL(row[i]);
        len += sprintf(str + len, "%lf", dv);
      } break;

      case TSDB_DATA_TYPE_BINARY:
      case TSDB_DATA_TYPE_NCHAR: {
L
Liu Jicong 已提交
353
        int32_t charLen = varDataLen((char *)row[i] - VARSTR_HEADER_SIZE);
H
Haojun Liao 已提交
354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373
        if (fields[i].type == TSDB_DATA_TYPE_BINARY) {
          assert(charLen <= fields[i].bytes && charLen >= 0);
        } else {
          assert(charLen <= fields[i].bytes * TSDB_NCHAR_SIZE && charLen >= 0);
        }

        memcpy(str + len, row[i], charLen);
        len += charLen;
      } break;

      case TSDB_DATA_TYPE_TIMESTAMP:
        len += sprintf(str + len, "%" PRId64, *((int64_t *)row[i]));
        break;

      case TSDB_DATA_TYPE_BOOL:
        len += sprintf(str + len, "%d", *((int8_t *)row[i]));
      default:
        break;
    }
  }
L
Liu Jicong 已提交
374
  str[len] = 0;
H
Haojun Liao 已提交
375 376 377

  return len;
}
378

L
Liu Jicong 已提交
379
int *taos_fetch_lengths(TAOS_RES *res) {
L
Liu Jicong 已提交
380
  if (res == NULL || TD_RES_TMQ_META(res)) {
381 382 383
    return NULL;
  }

L
Liu Jicong 已提交
384 385
  SReqResultInfo *pResInfo = tscGetCurResInfo(res);
  return pResInfo->length;
386 387
}

388
TAOS_ROW *taos_result_block(TAOS_RES *res) {
L
Liu Jicong 已提交
389
  if (res == NULL || TD_RES_TMQ_META(res)) {
390 391 392 393 394 395 396 397
    terrno = TSDB_CODE_INVALID_PARA;
    return NULL;
  }

  if (taos_is_update_query(res)) {
    return NULL;
  }

L
Liu Jicong 已提交
398 399
  SReqResultInfo *pResInfo = tscGetCurResInfo(res);
  return &pResInfo->row;
400 401
}

402
// todo intergrate with tDataTypes
403 404
const char *taos_data_type(int type) {
  switch (type) {
L
Liu Jicong 已提交
405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431
    case TSDB_DATA_TYPE_NULL:
      return "TSDB_DATA_TYPE_NULL";
    case TSDB_DATA_TYPE_BOOL:
      return "TSDB_DATA_TYPE_BOOL";
    case TSDB_DATA_TYPE_TINYINT:
      return "TSDB_DATA_TYPE_TINYINT";
    case TSDB_DATA_TYPE_SMALLINT:
      return "TSDB_DATA_TYPE_SMALLINT";
    case TSDB_DATA_TYPE_INT:
      return "TSDB_DATA_TYPE_INT";
    case TSDB_DATA_TYPE_BIGINT:
      return "TSDB_DATA_TYPE_BIGINT";
    case TSDB_DATA_TYPE_FLOAT:
      return "TSDB_DATA_TYPE_FLOAT";
    case TSDB_DATA_TYPE_DOUBLE:
      return "TSDB_DATA_TYPE_DOUBLE";
    case TSDB_DATA_TYPE_VARCHAR:
      return "TSDB_DATA_TYPE_VARCHAR";
      //    case TSDB_DATA_TYPE_BINARY:          return "TSDB_DATA_TYPE_VARCHAR";
    case TSDB_DATA_TYPE_TIMESTAMP:
      return "TSDB_DATA_TYPE_TIMESTAMP";
    case TSDB_DATA_TYPE_NCHAR:
      return "TSDB_DATA_TYPE_NCHAR";
    case TSDB_DATA_TYPE_JSON:
      return "TSDB_DATA_TYPE_JSON";
    default:
      return "UNKNOWN";
432 433
  }
}
434 435 436

const char *taos_get_client_info() { return version; }

X
Xiaoyu Wang 已提交
437
int taos_affected_rows(TAOS_RES *res) {
438
  if (res == NULL || TD_RES_TMQ(res) || TD_RES_TMQ_META(res) || TD_RES_TMQ_METADATA(res)) {
H
Haojun Liao 已提交
439 440 441
    return 0;
  }

dengyihao's avatar
dengyihao 已提交
442
  SRequestObj    *pRequest = (SRequestObj *)res;
L
Liu Jicong 已提交
443
  SReqResultInfo *pResInfo = &pRequest->body.resInfo;
H
Haojun Liao 已提交
444
  return pResInfo->numOfRows;
X
Xiaoyu Wang 已提交
445
}
446

447
int taos_result_precision(TAOS_RES *res) {
L
Liu Jicong 已提交
448
  if (res == NULL || TD_RES_TMQ_META(res)) {
H
Haojun Liao 已提交
449 450
    return TSDB_TIME_PRECISION_MILLI;
  }
451

L
Liu Jicong 已提交
452 453 454
  if (TD_RES_QUERY(res)) {
    SRequestObj *pRequest = (SRequestObj *)res;
    return pRequest->body.resInfo.precision;
455
  } else if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) {
L
Liu Jicong 已提交
456 457 458 459
    SReqResultInfo *info = tmqGetCurResInfo(res);
    return info->precision;
  }
  return TSDB_TIME_PRECISION_MILLI;
460
}
461 462

int taos_select_db(TAOS *taos, const char *db) {
L
Liu Jicong 已提交
463
  STscObj *pObj = acquireTscObj(*(int64_t *)taos);
464
  if (pObj == NULL) {
L
Liu Jicong 已提交
465
    releaseTscObj(*(int64_t *)taos);
466 467 468 469 470
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return TSDB_CODE_TSC_DISCONNECTED;
  }

  if (db == NULL || strlen(db) == 0) {
L
Liu Jicong 已提交
471
    releaseTscObj(*(int64_t *)taos);
472 473 474 475 476 477 478
    terrno = TSDB_CODE_TSC_INVALID_INPUT;
    return terrno;
  }

  char sql[256] = {0};
  snprintf(sql, tListLen(sql), "use %s", db);

L
Liu Jicong 已提交
479 480
  TAOS_RES *pRequest = taos_query(taos, sql);
  int32_t   code = taos_errno(pRequest);
481 482

  taos_free_result(pRequest);
L
Liu Jicong 已提交
483
  releaseTscObj(*(int64_t *)taos);
484 485 486 487
  return code;
}

void taos_stop_query(TAOS_RES *res) {
488
  if (res == NULL || TD_RES_TMQ(res) || TD_RES_TMQ_META(res) || TD_RES_TMQ_METADATA(res)) {
489 490 491
    return;
  }

L
Liu Jicong 已提交
492
  SRequestObj *pRequest = (SRequestObj *)res;
D
dapan1121 已提交
493
  pRequest->killed = true;
L
Liu Jicong 已提交
494 495

  int32_t numOfFields = taos_num_fields(pRequest);
496 497
  // It is not a query, no need to stop.
  if (numOfFields == 0) {
D
dapan1121 已提交
498
    tscDebug("request 0x%" PRIx64 " no need to be killed since not query", pRequest->requestId);
499 500 501
    return;
  }

D
dapan1121 已提交
502
  schedulerFreeJob(&pRequest->body.queryJob, TSDB_CODE_TSC_QUERY_KILLED);
D
dapan1121 已提交
503
  tscDebug("request %" PRIx64 " killed", pRequest->requestId);
504 505 506
}

bool taos_is_null(TAOS_RES *res, int32_t row, int32_t col) {
L
Liu Jicong 已提交
507 508 509
  if (res == NULL || TD_RES_TMQ_META(res)) {
    return true;
  }
L
Liu Jicong 已提交
510
  SReqResultInfo *pResultInfo = tscGetCurResInfo(res);
511 512 513 514
  if (col >= pResultInfo->numOfCols || col < 0 || row >= pResultInfo->numOfRows || row < 0) {
    return true;
  }

L
Liu Jicong 已提交
515
  SResultColumn *pCol = &pResultInfo->pCol[col];
516 517 518 519 520
  if (IS_VAR_DATA_TYPE(pResultInfo->fields[col].type)) {
    return (pCol->offset[row] == -1);
  } else {
    return colDataIsNull_f(pCol->nullbitmap, row);
  }
521 522
}

L
Liu Jicong 已提交
523
bool taos_is_update_query(TAOS_RES *res) { return taos_num_fields(res) == 0; }
524

L
Liu Jicong 已提交
525
int taos_fetch_block(TAOS_RES *res, TAOS_ROW *rows) {
H
Haojun Liao 已提交
526
  int32_t numOfRows = 0;
L
Liu Jicong 已提交
527
  /*int32_t code = */ taos_fetch_block_s(res, &numOfRows, rows);
H
Haojun Liao 已提交
528 529 530
  return numOfRows;
}

L
Liu Jicong 已提交
531
int taos_fetch_block_s(TAOS_RES *res, int *numOfRows, TAOS_ROW *rows) {
L
Liu Jicong 已提交
532
  if (res == NULL || TD_RES_TMQ_META(res)) {
533 534
    return 0;
  }
535

L
Liu Jicong 已提交
536 537
  if (TD_RES_QUERY(res)) {
    SRequestObj *pRequest = (SRequestObj *)res;
538

L
Liu Jicong 已提交
539 540
    (*rows) = NULL;
    (*numOfRows) = 0;
H
Haojun Liao 已提交
541

L
Liu Jicong 已提交
542 543 544 545
    if (pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT || pRequest->type == TSDB_SQL_INSERT ||
        pRequest->code != TSDB_CODE_SUCCESS || taos_num_fields(res) == 0) {
      return 0;
    }
546

547
    doAsyncFetchRows(pRequest, false, true);
548

L
Liu Jicong 已提交
549 550 551
    // TODO refactor
    SReqResultInfo *pResultInfo = &pRequest->body.resInfo;
    pResultInfo->current = pResultInfo->numOfRows;
552

L
Liu Jicong 已提交
553 554 555
    (*rows) = pResultInfo->row;
    (*numOfRows) = pResultInfo->numOfRows;
    return pRequest->code;
556
  } else if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) {
L
Liu Jicong 已提交
557
    SReqResultInfo *pResultInfo = tmqGetNextResInfo(res, true);
L
Liu Jicong 已提交
558
    if (pResultInfo == NULL) return -1;
H
Haojun Liao 已提交
559

L
Liu Jicong 已提交
560 561 562
    pResultInfo->current = pResultInfo->numOfRows;
    (*rows) = pResultInfo->row;
    (*numOfRows) = pResultInfo->numOfRows;
H
Haojun Liao 已提交
563
    return 0;
L
Liu Jicong 已提交
564 565 566
  } else {
    ASSERT(0);
    return -1;
H
Haojun Liao 已提交
567
  }
L
Liu Jicong 已提交
568
}
H
Haojun Liao 已提交
569

L
Liu Jicong 已提交
570
int taos_fetch_raw_block(TAOS_RES *res, int *numOfRows, void **pData) {
L
Liu Jicong 已提交
571
  if (res == NULL || TD_RES_TMQ_META(res)) {
H
Haojun Liao 已提交
572 573
    return 0;
  }
574

575
  if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) {
L
Liu Jicong 已提交
576
    SReqResultInfo *pResultInfo = tmqGetNextResInfo(res, false);
L
Liu Jicong 已提交
577 578 579 580
    if (pResultInfo == NULL) {
      (*numOfRows) = 0;
      return 0;
    }
H
Haojun Liao 已提交
581

L
Liu Jicong 已提交
582 583 584 585
    pResultInfo->current = pResultInfo->numOfRows;
    (*numOfRows) = pResultInfo->numOfRows;
    (*pData) = (void *)pResultInfo->pData;
    return 0;
586
  }
L
Liu Jicong 已提交
587

588 589 590 591 592
  SRequestObj *pRequest = (SRequestObj *)res;

  if (pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT || pRequest->type == TSDB_SQL_INSERT ||
      pRequest->code != TSDB_CODE_SUCCESS || taos_num_fields(res) == 0) {
    return 0;
L
Liu Jicong 已提交
593
  }
594

595
  doAsyncFetchRows(pRequest, false, false);
596 597 598 599 600 601 602 603

  SReqResultInfo *pResultInfo = &pRequest->body.resInfo;

  pResultInfo->current = pResultInfo->numOfRows;
  (*numOfRows) = pResultInfo->numOfRows;
  (*pData) = (void *)pResultInfo->pData;

  return 0;
604 605
}

H
Haojun Liao 已提交
606
int *taos_get_column_data_offset(TAOS_RES *res, int columnIndex) {
L
Liu Jicong 已提交
607
  if (res == NULL || TD_RES_TMQ_META(res)) {
H
Haojun Liao 已提交
608 609 610
    return 0;
  }

L
Liu Jicong 已提交
611
  int32_t numOfFields = taos_num_fields(res);
H
Haojun Liao 已提交
612 613 614 615
  if (columnIndex < 0 || columnIndex >= numOfFields || numOfFields == 0) {
    return 0;
  }

L
Liu Jicong 已提交
616
  SReqResultInfo *pResInfo = tscGetCurResInfo(res);
dengyihao's avatar
dengyihao 已提交
617
  TAOS_FIELD     *pField = &pResInfo->userFields[columnIndex];
H
Haojun Liao 已提交
618 619 620 621
  if (!IS_VAR_DATA_TYPE(pField->type)) {
    return 0;
  }

L
Liu Jicong 已提交
622
  return pResInfo->pCol[columnIndex].offset;
H
Haojun Liao 已提交
623 624
}

625 626
int taos_validate_sql(TAOS *taos, const char *sql) {
  TAOS_RES *pObj = taosQueryImpl(taos, sql, true);
D
dapan1121 已提交
627 628

  int code = taos_errno(pObj);
629

D
dapan1121 已提交
630 631 632
  taos_free_result(pObj);
  return code;
}
633

634
void taos_reset_current_db(TAOS *taos) {
L
Liu Jicong 已提交
635
  STscObj *pTscObj = acquireTscObj(*(int64_t *)taos);
636 637
  if (pTscObj == NULL) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
H
Haojun Liao 已提交
638 639 640
    return;
  }

641 642
  resetConnectDB(pTscObj);

L
Liu Jicong 已提交
643
  releaseTscObj(*(int64_t *)taos);
644 645
}

646
const char *taos_get_server_info(TAOS *taos) {
L
Liu Jicong 已提交
647
  STscObj *pTscObj = acquireTscObj(*(int64_t *)taos);
648 649
  if (pTscObj == NULL) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
650 651 652
    return NULL;
  }

L
Liu Jicong 已提交
653
  releaseTscObj(*(int64_t *)taos);
654

D
dapan1121 已提交
655
  return pTscObj->sDetailVer;
656 657
}

658 659 660 661 662
static void destoryTablesReq(void *p) {
  STablesReq *pRes = (STablesReq *)p;
  taosArrayDestroy(pRes->pTables);
}

663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685
static void destoryCatalogReq(SCatalogReq *pCatalogReq) {
  if (NULL == pCatalogReq) {
    return;
  }
  taosArrayDestroy(pCatalogReq->pDbVgroup);
  taosArrayDestroy(pCatalogReq->pDbCfg);
  taosArrayDestroy(pCatalogReq->pDbInfo);
  taosArrayDestroyEx(pCatalogReq->pTableMeta, destoryTablesReq);
  taosArrayDestroyEx(pCatalogReq->pTableHash, destoryTablesReq);
  taosArrayDestroy(pCatalogReq->pUdf);
  taosArrayDestroy(pCatalogReq->pIndex);
  taosArrayDestroy(pCatalogReq->pUser);
  taosArrayDestroy(pCatalogReq->pTableIndex);
  taosMemoryFree(pCatalogReq);
}

void destorySqlCallbackWrapper(SSqlCallbackWrapper *pWrapper) {
  if (NULL == pWrapper) {
    return;
  }
  destoryCatalogReq(pWrapper->pCatalogReq);
  qDestroyParseContext(pWrapper->pParseCtx);
  catalogFreeMetaData(pWrapper->pResultMeta);
686 687 688 689
  taosMemoryFree(pWrapper);
}

void retrieveMetaCallback(SMetaData *pResultMeta, void *param, int32_t code) {
690 691 692
  SSqlCallbackWrapper *pWrapper = (SSqlCallbackWrapper *)param;
  SRequestObj         *pRequest = pWrapper->pRequest;
  SQuery              *pQuery = pRequest->pQuery;
693

D
dapan1121 已提交
694
  pRequest->metric.ctgEnd = taosGetTimestampUs();
695
  qDebug("0x%" PRIx64 " start to semantic analysis, reqId:0x%" PRIx64, pRequest->self, pRequest->requestId);
D
dapan1121 已提交
696

697
  if (code == TSDB_CODE_SUCCESS) {
698
    code = qAnalyseSqlSemantic(pWrapper->pParseCtx, pWrapper->pCatalogReq, pResultMeta, pQuery);
699
    pRequest->stableQuery = pQuery->stableQuery;
D
dapan1121 已提交
700
    if (pQuery->pRoot) {
dengyihao's avatar
dengyihao 已提交
701
      pRequest->stmtType = pQuery->pRoot->type;
D
dapan1121 已提交
702
    }
703 704
  }

D
dapan1121 已提交
705 706
  pRequest->metric.semanticEnd = taosGetTimestampUs();

707 708 709 710 711 712 713
  if (code == TSDB_CODE_SUCCESS && pWrapper->pParseCtx->needMultiParse) {
    pWrapper->pResultMeta = catalogCloneMetaData(pResultMeta);
    if (NULL == pWrapper->pResultMeta) {
      code = TSDB_CODE_OUT_OF_MEMORY;
    }
  }

714 715 716 717 718
  if (code == TSDB_CODE_SUCCESS) {
    if (pQuery->haveResultSet) {
      setResSchemaInfo(&pRequest->body.resInfo, pQuery->pResSchema, pQuery->numOfResCols);
      setResPrecision(&pRequest->body.resInfo, pQuery->precision);
    }
719

720 721
    TSWAP(pRequest->dbList, (pQuery)->pDbList);
    TSWAP(pRequest->tableList, (pQuery)->pTableList);
D
dapan1121 已提交
722
    TSWAP(pRequest->targetTableList, (pQuery)->pTargetTableList);
723

724
    double el = (pRequest->metric.semanticEnd - pRequest->metric.ctgEnd) / 1000.0;
725 726 727
    tscDebug("0x%" PRIx64 " analysis semantics completed, start async query, elapsed time:%.2f ms, reqId:0x%" PRIx64,
             pRequest->self, el, pRequest->requestId);

728
    launchAsyncQuery(pRequest, pQuery, pResultMeta, pWrapper);
729
  } else {
730
    destorySqlCallbackWrapper(pWrapper);
X
Xiaoyu Wang 已提交
731 732
    qDestroyQuery(pRequest->pQuery);
    pRequest->pQuery = NULL;
733

734
    if (NEED_CLIENT_HANDLE_ERROR(code)) {
735 736
      tscDebug("0x%" PRIx64 " client retry to handle the error, code:%d - %s, tryCount:%d, reqId:0x%" PRIx64,
               pRequest->self, code, tstrerror(code), pRequest->retry, pRequest->requestId);
737 738 739 740
      pRequest->prevCode = code;
      doAsyncQuery(pRequest, true);
      return;
    }
741

742
    // return to app directly
743 744
    tscError("0x%" PRIx64 " error occurs, code:%s, return to user app, reqId:0x%" PRIx64, pRequest->self,
             tstrerror(code), pRequest->requestId);
745 746 747
    pRequest->code = code;
    pRequest->body.queryFp(pRequest->body.param, pRequest, code);
  }
748 749
}

750 751 752 753 754 755 756 757 758 759
int32_t continueInsertFromCsv(SSqlCallbackWrapper *pWrapper, SRequestObj *pRequest) {
  qDestroyQuery(pRequest->pQuery);
  pRequest->pQuery = (SQuery *)nodesMakeNode(QUERY_NODE_QUERY);
  if (NULL == pRequest->pQuery) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }
  retrieveMetaCallback(pWrapper->pResultMeta, pWrapper, TSDB_CODE_SUCCESS);
  return TSDB_CODE_SUCCESS;
}

760
void taos_query_a(TAOS *taos, const char *sql, __taos_async_fn_t fp, void *param) {
dengyihao's avatar
dengyihao 已提交
761
  int64_t connId = *(int64_t *)taos;
762
  taosAsyncQueryImpl(connId, sql, fp, param, false);
763
}
764

765
int32_t createParseContext(const SRequestObj *pRequest, SParseContext **pCxt) {
766 767 768 769 770 771 772
  const STscObj *pTscObj = pRequest->pTscObj;

  *pCxt = taosMemoryCalloc(1, sizeof(SParseContext));
  if (*pCxt == NULL) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }

dengyihao's avatar
dengyihao 已提交
773 774 775 776 777 778 779 780 781 782 783 784 785 786
  **pCxt = (SParseContext){.requestId = pRequest->requestId,
                           .requestRid = pRequest->self,
                           .acctId = pTscObj->acctId,
                           .db = pRequest->pDb,
                           .topicQuery = false,
                           .pSql = pRequest->sqlstr,
                           .sqlLen = pRequest->sqlLen,
                           .pMsg = pRequest->msgBuf,
                           .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
                           .pTransporter = pTscObj->pAppInfo->pTransporter,
                           .pStmtCb = NULL,
                           .pUser = pTscObj->user,
                           .schemalessType = pTscObj->schemalessType,
                           .isSuperUser = (0 == strcmp(pTscObj->user, TSDB_DEFAULT_USER)),
787
                           .enableSysInfo = pTscObj->sysInfo,
dengyihao's avatar
dengyihao 已提交
788 789
                           .async = true,
                           .svrVer = pTscObj->sVer,
790 791
                           .nodeOffline = (pTscObj->pAppInfo->onlineDnodes < pTscObj->pAppInfo->totalDnodes),
                           .allocatorId = pRequest->allocatorRefId};
792 793 794
  return TSDB_CODE_SUCCESS;
}

795
void doAsyncQuery(SRequestObj *pRequest, bool updateMetaForce) {
796 797 798
  STscObj             *pTscObj = pRequest->pTscObj;
  SSqlCallbackWrapper *pWrapper = NULL;
  int32_t              code = TSDB_CODE_SUCCESS;
799 800

  if (pRequest->retry++ > REQUEST_TOTAL_EXEC_TIMES) {
801
    code = pRequest->prevCode;
802 803
  }

804 805 806 807 808 809 810
  if (TSDB_CODE_SUCCESS == code) {
    pWrapper = taosMemoryCalloc(1, sizeof(SSqlCallbackWrapper));
    if (pWrapper == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
    } else {
      pWrapper->pRequest = pRequest;
    }
811
  }
812

813 814
  if (TSDB_CODE_SUCCESS == code) {
    code = createParseContext(pRequest, &pWrapper->pParseCtx);
815 816
  }

817 818 819
  if (TSDB_CODE_SUCCESS == code) {
    pWrapper->pParseCtx->mgmtEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
    code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pWrapper->pParseCtx->pCatalog);
820 821
  }

822 823 824 825 826 827 828 829 830 831 832 833 834 835
  if (TSDB_CODE_SUCCESS == code) {
    pRequest->metric.syntaxStart = taosGetTimestampUs();

    pWrapper->pCatalogReq = taosMemoryCalloc(1, sizeof(SCatalogReq));
    if (pWrapper->pCatalogReq == NULL) {
      code = TSDB_CODE_OUT_OF_MEMORY;
    } else {
      pWrapper->pCatalogReq->forceUpdate = updateMetaForce;
      pWrapper->pCatalogReq->qNodeRequired = qnodeRequired(pRequest);
      code = qParseSqlSyntax(pWrapper->pParseCtx, &pRequest->pQuery, pWrapper->pCatalogReq);
    }

    pRequest->metric.syntaxEnd = taosGetTimestampUs();
  }
D
dapan1121 已提交
836

837
  if (TSDB_CODE_SUCCESS == code && !updateMetaForce) {
D
dapan1121 已提交
838
    SAppClusterSummary *pActivity = &pTscObj->pAppInfo->summary;
X
Xiaoyu Wang 已提交
839
    if (NULL == pRequest->pQuery->pRoot) {
dengyihao's avatar
dengyihao 已提交
840
      atomic_add_fetch_64((int64_t *)&pActivity->numOfInsertsReq, 1);
X
Xiaoyu Wang 已提交
841
    } else if (QUERY_NODE_SELECT_STMT == pRequest->pQuery->pRoot->type) {
dengyihao's avatar
dengyihao 已提交
842
      atomic_add_fetch_64((int64_t *)&pActivity->numOfQueryReq, 1);
D
dapan1121 已提交
843 844 845
    }
  }

846 847 848 849 850
  if (TSDB_CODE_SUCCESS == code) {
    SRequestConnInfo conn = {.pTrans = pWrapper->pParseCtx->pTransporter,
                             .requestId = pWrapper->pParseCtx->requestId,
                             .requestObjRefId = pWrapper->pParseCtx->requestRid,
                             .mgmtEps = pWrapper->pParseCtx->mgmtEpSet};
D
dapan1121 已提交
851

852
    pRequest->metric.ctgStart = taosGetTimestampUs();
D
dapan1121 已提交
853

854 855
    code = catalogAsyncGetAllMeta(pWrapper->pParseCtx->pCatalog, &conn, pWrapper->pCatalogReq, retrieveMetaCallback,
                                  pWrapper, &pRequest->body.queryJob);
856 857
  }

858 859 860 861 862 863 864 865
  if (TSDB_CODE_SUCCESS != code) {
    tscError("0x%" PRIx64 " error happens, code:%d - %s, reqId:0x%" PRIx64, pRequest->self, code, tstrerror(code),
             pRequest->requestId);
    destorySqlCallbackWrapper(pWrapper);
    terrno = code;
    pRequest->code = code;
    pRequest->body.queryFp(pRequest->body.param, pRequest, code);
  }
866 867
}

868 869
static void fetchCallback(void *pResult, void *param, int32_t code) {
  SRequestObj *pRequest = (SRequestObj *)param;
870

871
  SReqResultInfo *pResultInfo = &pRequest->body.resInfo;
872
  pRequest->metric.resultReady = taosGetTimestampUs();
873

L
Liu Jicong 已提交
874 875
  tscDebug("0x%" PRIx64 " enter scheduler fetch cb, code:%d - %s, reqId:0x%" PRIx64, pRequest->self, code,
           tstrerror(code), pRequest->requestId);
D
dapan1121 已提交
876

877 878 879 880 881
  pResultInfo->pData = pResult;
  pResultInfo->numOfRows = 0;

  if (code != TSDB_CODE_SUCCESS) {
    pRequest->code = code;
882
    taosMemoryFreeClear(pResultInfo->pData);
883 884 885 886 887
    pRequest->body.fetchFp(pRequest->body.param, pRequest, 0);
    return;
  }

  if (pRequest->code != TSDB_CODE_SUCCESS) {
888
    taosMemoryFreeClear(pResultInfo->pData);
889
    pRequest->body.fetchFp(pRequest->body.param, pRequest, 0);
890
    return;
891 892
  }

893
  pRequest->code =
D
dapan1121 已提交
894
      setQueryResultFromRsp(pResultInfo, (SRetrieveTableRsp *)pResultInfo->pData, pResultInfo->convertUcs4, true);
895 896 897
  if (pRequest->code != TSDB_CODE_SUCCESS) {
    pResultInfo->numOfRows = 0;
    pRequest->code = code;
898 899 900 901 902 903
    tscError("0x%" PRIx64 " fetch results failed, code:%s, reqId:0x%" PRIx64, pRequest->self, tstrerror(code),
             pRequest->requestId);
  } else {
    tscDebug("0x%" PRIx64 " fetch results, numOfRows:%d total Rows:%" PRId64 ", complete:%d, reqId:0x%" PRIx64,
             pRequest->self, pResultInfo->numOfRows, pResultInfo->totalRows, pResultInfo->completed,
             pRequest->requestId);
D
dapan1121 已提交
904 905 906

    STscObj            *pTscObj = pRequest->pTscObj;
    SAppClusterSummary *pActivity = &pTscObj->pAppInfo->summary;
dengyihao's avatar
dengyihao 已提交
907
    atomic_add_fetch_64((int64_t *)&pActivity->fetchBytes, pRequest->body.resInfo.payloadLen);
908 909 910 911 912
  }

  pRequest->body.fetchFp(pRequest->body.param, pRequest, pResultInfo->numOfRows);
}

913
void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {
914
  ASSERT(res != NULL && fp != NULL);
L
Liu Jicong 已提交
915
  ASSERT(TD_RES_QUERY(res));
916 917 918

  SRequestObj *pRequest = res;
  pRequest->body.fetchFp = fp;
919
  pRequest->body.param = param;
920 921

  SReqResultInfo *pResultInfo = &pRequest->body.resInfo;
922 923 924

  // this query has no results or error exists, return directly
  if (taos_num_fields(pRequest) == 0 || pRequest->code != TSDB_CODE_SUCCESS) {
925 926 927 928 929
    pResultInfo->numOfRows = 0;
    pRequest->body.fetchFp(param, pRequest, pResultInfo->numOfRows);
    return;
  }

930
  // all data has returned to App already, no need to try again
D
dapan1121 已提交
931 932 933 934 935 936 937 938 939
  if (pResultInfo->completed) {
    // it is a local executed query, no need to do async fetch
    if (QUERY_EXEC_MODE_LOCAL == pRequest->body.execMode) {
      if (pResultInfo->localResultFetched) {
        pResultInfo->numOfRows = 0;
        pResultInfo->current = 0;
      } else {
        pResultInfo->localResultFetched = true;
      }
940
    } else {
D
dapan1121 已提交
941
      pResultInfo->numOfRows = 0;
942
    }
943

D
dapan1121 已提交
944
    pRequest->body.fetchFp(param, pRequest, pResultInfo->numOfRows);
945
    return;
946 947
  }

D
dapan1121 已提交
948
  SSchedulerReq req = {
dengyihao's avatar
dengyihao 已提交
949 950 951
      .syncReq = false,
      .fetchFp = fetchCallback,
      .cbParam = pRequest,
D
dapan1121 已提交
952
  };
953

D
dapan1121 已提交
954
  schedulerFetchRows(pRequest->body.queryJob, &req);
L
Liu Jicong 已提交
955
}
956

957
void taos_fetch_raw_block_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {
958
  ASSERT(res != NULL && fp != NULL);
L
Liu Jicong 已提交
959
  ASSERT(TD_RES_QUERY(res));
960

dengyihao's avatar
dengyihao 已提交
961
  SRequestObj    *pRequest = res;
962 963 964
  SReqResultInfo *pResultInfo = &pRequest->body.resInfo;

  // set the current block is all consumed
965 966
  pResultInfo->convertUcs4 = false;

967 968
  // it is a local executed query, no need to do async fetch
  taos_fetch_rows_a(pRequest, fp, param);
969 970
}

971
const void *taos_get_raw_block(TAOS_RES *res) {
H
Haojun Liao 已提交
972
  ASSERT(res != NULL);
L
Liu Jicong 已提交
973
  ASSERT(TD_RES_QUERY(res));
974
  SRequestObj *pRequest = res;
H
Haojun Liao 已提交
975 976 977 978

  return pRequest->body.resInfo.pData;
}

X
Xiaoyu Wang 已提交
979
int taos_get_db_route_info(TAOS *taos, const char *db, TAOS_DB_ROUTE_INFO *dbInfo) {
980 981 982 983 984 985 986 987 988 989 990
  if (NULL == taos) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return terrno;
  }

  if (NULL == db || NULL == dbInfo) {
    tscError("invalid input param, db:%p, dbInfo:%p", db, dbInfo);
    terrno = TSDB_CODE_TSC_INVALID_INPUT;
    return terrno;
  }

X
Xiaoyu Wang 已提交
991 992 993 994
  int64_t      connId = *(int64_t *)taos;
  SRequestObj *pRequest = NULL;
  char        *sql = "taos_get_db_route_info";
  int32_t      code = buildRequest(connId, sql, strlen(sql), NULL, false, &pRequest);
995
  if (code != TSDB_CODE_SUCCESS) {
996
    terrno = code;
997 998 999
    return terrno;
  }

X
Xiaoyu Wang 已提交
1000
  STscObj  *pTscObj = pRequest->pTscObj;
1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013
  SCatalog *pCtg = NULL;
  code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCtg);
  if (code != TSDB_CODE_SUCCESS) {
    goto _return;
  }

  SRequestConnInfo conn = {
      .pTrans = pTscObj->pAppInfo->pTransporter, .requestId = pRequest->requestId, .requestObjRefId = pRequest->self};

  conn.mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);

  char dbFName[TSDB_DB_FNAME_LEN] = {0};
  snprintf(dbFName, sizeof(dbFName), "%d.%s", pTscObj->acctId, db);
X
Xiaoyu Wang 已提交
1014

1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027
  code = catalogGetDBVgInfo(pCtg, &conn, dbFName, dbInfo);
  if (code) {
    goto _return;
  }

_return:

  terrno = code;

  destroyRequest(pRequest);
  return code;
}

X
Xiaoyu Wang 已提交
1028
int taos_get_table_vgId(TAOS *taos, const char *db, const char *table, int *vgId) {
1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039
  if (NULL == taos) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return terrno;
  }

  if (NULL == db || NULL == table || NULL == vgId) {
    tscError("invalid input param, db:%p, table:%p, vgId:%p", db, table, vgId);
    terrno = TSDB_CODE_TSC_INVALID_INPUT;
    return terrno;
  }

X
Xiaoyu Wang 已提交
1040 1041 1042 1043
  int64_t      connId = *(int64_t *)taos;
  SRequestObj *pRequest = NULL;
  char        *sql = "taos_get_table_vgId";
  int32_t      code = buildRequest(connId, sql, strlen(sql), NULL, false, &pRequest);
1044 1045 1046 1047
  if (code != TSDB_CODE_SUCCESS) {
    return terrno;
  }

X
Xiaoyu Wang 已提交
1048
  STscObj  *pTscObj = pRequest->pTscObj;
1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073
  SCatalog *pCtg = NULL;
  code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCtg);
  if (code != TSDB_CODE_SUCCESS) {
    goto _return;
  }

  SRequestConnInfo conn = {
      .pTrans = pTscObj->pAppInfo->pTransporter, .requestId = pRequest->requestId, .requestObjRefId = pRequest->self};

  conn.mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);

  SName tableName;
  toName(pTscObj->acctId, db, table, &tableName);

  SVgroupInfo vgInfo;
  code = catalogGetTableHashVgroup(pCtg, &conn, &tableName, &vgInfo);
  if (code) {
    goto _return;
  }

  *vgId = vgInfo.vgId;

_return:

  terrno = code;
1074

1075 1076
  destroyRequest(pRequest);
  return code;
1077 1078
}

1079
int taos_load_table_info(TAOS *taos, const char *tableNameList) {
D
dapan1121 已提交
1080 1081 1082 1083 1084
  if (NULL == taos) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return terrno;
  }

1085
  int64_t       connId = *(int64_t *)taos;
D
dapan1121 已提交
1086
  const int32_t MAX_TABLE_NAME_LENGTH = 12 * 1024 * 1024;  // 12MB list
1087
  int32_t       code = 0;
dengyihao's avatar
dengyihao 已提交
1088
  SRequestObj  *pRequest = NULL;
1089 1090
  SCatalogReq   catalogReq = {0};

D
dapan1121 已提交
1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102
  if (NULL == tableNameList) {
    return TSDB_CODE_SUCCESS;
  }

  int32_t length = (int32_t)strlen(tableNameList);
  if (0 == length) {
    return TSDB_CODE_SUCCESS;
  } else if (length > MAX_TABLE_NAME_LENGTH) {
    tscError("tableNameList too long, length:%d, maximum allowed:%d", length, MAX_TABLE_NAME_LENGTH);
    return TSDB_CODE_TSC_INVALID_OPERATION;
  }

1103 1104 1105 1106 1107
  char *sql = "taos_load_table_info";
  code = buildRequest(connId, sql, strlen(sql), NULL, false, &pRequest);
  if (code != TSDB_CODE_SUCCESS) {
    terrno = code;
    goto _return;
D
dapan1121 已提交
1108 1109
  }

dengyihao's avatar
dengyihao 已提交
1110
  STscObj *pTscObj = pRequest->pTscObj;
D
dapan1121 已提交
1111 1112 1113 1114 1115
  code = transferTableNameList(tableNameList, pTscObj->acctId, pTscObj->db, &catalogReq.pTableMeta);
  if (code) {
    goto _return;
  }

1116
  SCatalog *pCtg = NULL;
D
dapan1121 已提交
1117 1118 1119 1120 1121
  code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCtg);
  if (code != TSDB_CODE_SUCCESS) {
    goto _return;
  }

1122 1123
  SRequestConnInfo conn = {
      .pTrans = pTscObj->pAppInfo->pTransporter, .requestId = pRequest->requestId, .requestObjRefId = pRequest->self};
D
dapan1121 已提交
1124 1125 1126

  conn.mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);

D
dapan1121 已提交
1127
  code = catalogAsyncGetAllMeta(pCtg, &conn, &catalogReq, syncCatalogFn, pRequest->body.param, NULL);
D
dapan1121 已提交
1128 1129 1130 1131
  if (code) {
    goto _return;
  }

dengyihao's avatar
dengyihao 已提交
1132
  SSyncQueryParam *pParam = pRequest->body.param;
1133
  tsem_wait(&pParam->sem);
D
dapan1121 已提交
1134 1135 1136 1137 1138

_return:
  taosArrayDestroy(catalogReq.pTableMeta);
  destroyRequest(pRequest);
  return code;
1139 1140
}

L
Liu Jicong 已提交
1141
TAOS_STMT *taos_stmt_init(TAOS *taos) {
L
Liu Jicong 已提交
1142
  STscObj *pObj = acquireTscObj(*(int64_t *)taos);
1143 1144 1145
  if (NULL == pObj) {
    tscError("invalid parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_TSC_DISCONNECTED;
D
stmt  
dapan1121 已提交
1146 1147 1148
    return NULL;
  }

L
Liu Jicong 已提交
1149 1150 1151
  TAOS_STMT *pStmt = stmtInit(pObj);

  releaseTscObj(*(int64_t *)taos);
1152 1153

  return pStmt;
1154 1155
}

D
stmt  
dapan1121 已提交
1156 1157
int taos_stmt_prepare(TAOS_STMT *stmt, const char *sql, unsigned long length) {
  if (stmt == NULL || sql == NULL) {
D
stmt  
dapan1121 已提交
1158
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1159 1160 1161 1162
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

D
stmt  
dapan1121 已提交
1163
  return stmtPrepare(stmt, sql, length);
1164 1165
}

D
dapan1121 已提交
1166
int taos_stmt_set_tbname_tags(TAOS_STMT *stmt, const char *name, TAOS_MULTI_BIND *tags) {
D
stmt  
dapan1121 已提交
1167
  if (stmt == NULL || name == NULL) {
D
stmt  
dapan1121 已提交
1168
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1169 1170 1171 1172
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

D
stmt  
dapan1121 已提交
1173 1174 1175 1176 1177 1178 1179 1180 1181 1182
  int32_t code = stmtSetTbName(stmt, name);
  if (code) {
    return code;
  }

  if (tags) {
    return stmtSetTbTags(stmt, tags);
  }

  return TSDB_CODE_SUCCESS;
1183 1184
}

D
stmt  
dapan1121 已提交
1185 1186
int taos_stmt_set_tbname(TAOS_STMT *stmt, const char *name) {
  if (stmt == NULL || name == NULL) {
D
stmt  
dapan1121 已提交
1187
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1188
    terrno = TSDB_CODE_INVALID_PARA;
D
stmt  
dapan1121 已提交
1189
    return terrno;
D
stmt  
dapan1121 已提交
1190 1191
  }

D
stmt  
dapan1121 已提交
1192
  return stmtSetTbName(stmt, name);
1193 1194
}

D
dapan1121 已提交
1195 1196 1197 1198 1199 1200 1201 1202 1203 1204
int taos_stmt_set_tags(TAOS_STMT *stmt, TAOS_MULTI_BIND *tags) {
  if (stmt == NULL || tags == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  return stmtSetTbTags(stmt, tags);
}

1205
int taos_stmt_set_sub_tbname(TAOS_STMT *stmt, const char *name) { return taos_stmt_set_tbname(stmt, name); }
D
dapan1121 已提交
1206

1207
int taos_stmt_get_tag_fields(TAOS_STMT *stmt, int *fieldNum, TAOS_FIELD_E **fields) {
D
dapan1121 已提交
1208 1209 1210 1211 1212
  if (stmt == NULL || NULL == fieldNum) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }
1213

D
dapan1121 已提交
1214 1215 1216
  return stmtGetTagFields(stmt, fieldNum, fields);
}

1217
int taos_stmt_get_col_fields(TAOS_STMT *stmt, int *fieldNum, TAOS_FIELD_E **fields) {
D
dapan1121 已提交
1218 1219 1220 1221 1222
  if (stmt == NULL || NULL == fieldNum) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }
1223

D
dapan1121 已提交
1224 1225 1226
  return stmtGetColFields(stmt, fieldNum, fields);
}

D
dapan1121 已提交
1227
int taos_stmt_bind_param(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind) {
D
stmt  
dapan1121 已提交
1228
  if (stmt == NULL || bind == NULL) {
D
stmt  
dapan1121 已提交
1229
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1230
    terrno = TSDB_CODE_INVALID_PARA;
D
stmt  
dapan1121 已提交
1231
    return terrno;
D
stmt  
dapan1121 已提交
1232
  }
1233

D
stmt  
dapan1121 已提交
1234 1235 1236 1237 1238
  if (bind->num > 1) {
    tscError("invalid bind number %d for %s", bind->num, __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }
1239

D
stmt  
dapan1121 已提交
1240
  return stmtBindBatch(stmt, bind, -1);
1241 1242
}

D
dapan1121 已提交
1243
int taos_stmt_bind_param_batch(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind) {
D
stmt  
dapan1121 已提交
1244
  if (stmt == NULL || bind == NULL) {
D
stmt  
dapan1121 已提交
1245
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1246 1247 1248 1249
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

D
stmt  
dapan1121 已提交
1250 1251
  if (bind->num <= 0 || bind->num > INT16_MAX) {
    tscError("invalid bind num %d", bind->num);
D
stmt  
dapan1121 已提交
1252 1253 1254 1255
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

D
dapan1121 已提交
1256 1257 1258 1259 1260 1261 1262 1263
  int32_t insert = 0;
  stmtIsInsert(stmt, &insert);
  if (0 == insert && bind->num > 1) {
    tscError("only one row data allowed for query");
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

D
stmt  
dapan1121 已提交
1264
  return stmtBindBatch(stmt, bind, -1);
1265 1266
}

D
dapan1121 已提交
1267
int taos_stmt_bind_single_param_batch(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind, int colIdx) {
D
stmt  
dapan1121 已提交
1268
  if (stmt == NULL || bind == NULL) {
D
stmt  
dapan1121 已提交
1269
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1270 1271 1272 1273
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

D
stmt  
dapan1121 已提交
1274
  if (colIdx < 0) {
D
stmt  
dapan1121 已提交
1275 1276 1277 1278
    tscError("invalid bind column idx %d", colIdx);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }
D
dapan1121 已提交
1279 1280 1281 1282 1283 1284 1285 1286

  int32_t insert = 0;
  stmtIsInsert(stmt, &insert);
  if (0 == insert && bind->num > 1) {
    tscError("only one row data allowed for query");
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }
1287

D
stmt  
dapan1121 已提交
1288
  return stmtBindBatch(stmt, bind, colIdx);
1289 1290
}

D
stmt  
dapan1121 已提交
1291 1292
int taos_stmt_add_batch(TAOS_STMT *stmt) {
  if (stmt == NULL) {
D
stmt  
dapan1121 已提交
1293
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1294 1295 1296 1297
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

D
stmt  
dapan1121 已提交
1298
  return stmtAddBatch(stmt);
1299 1300
}

D
stmt  
dapan1121 已提交
1301 1302
int taos_stmt_execute(TAOS_STMT *stmt) {
  if (stmt == NULL) {
D
stmt  
dapan1121 已提交
1303
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1304 1305 1306 1307
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

D
stmt  
dapan1121 已提交
1308
  return stmtExec(stmt);
1309 1310
}

1311
int taos_stmt_is_insert(TAOS_STMT *stmt, int *insert) {
D
stmt  
dapan1121 已提交
1312
  if (stmt == NULL || insert == NULL) {
D
stmt  
dapan1121 已提交
1313
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1314 1315 1316 1317 1318
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  return stmtIsInsert(stmt, insert);
1319 1320
}

1321
int taos_stmt_num_params(TAOS_STMT *stmt, int *nums) {
D
stmt  
dapan1121 已提交
1322
  if (stmt == NULL || nums == NULL) {
D
stmt  
dapan1121 已提交
1323
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1324 1325 1326 1327 1328
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  return stmtGetParamNum(stmt, nums);
1329 1330
}

D
dapan1121 已提交
1331 1332 1333 1334 1335 1336 1337 1338 1339 1340
int taos_stmt_get_param(TAOS_STMT *stmt, int idx, int *type, int *bytes) {
  if (stmt == NULL || type == NULL || NULL == bytes || idx < 0) {
    tscError("invalid parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  return stmtGetParam(stmt, idx, type, bytes);
}

D
stmt  
dapan1121 已提交
1341
TAOS_RES *taos_stmt_use_result(TAOS_STMT *stmt) {
D
stmt  
dapan1121 已提交
1342
  if (stmt == NULL) {
D
stmt  
dapan1121 已提交
1343
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1344
    terrno = TSDB_CODE_INVALID_PARA;
D
stmt  
dapan1121 已提交
1345
    return NULL;
D
stmt  
dapan1121 已提交
1346 1347
  }

D
stmt  
dapan1121 已提交
1348
  return stmtUseResult(stmt);
1349 1350
}

1351
char *taos_stmt_errstr(TAOS_STMT *stmt) { return (char *)stmtErrstr(stmt); }
1352

D
stmt  
dapan1121 已提交
1353
int taos_stmt_affected_rows(TAOS_STMT *stmt) {
D
stmt  
dapan1121 已提交
1354
  if (stmt == NULL) {
D
stmt  
dapan1121 已提交
1355
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1356
    terrno = TSDB_CODE_INVALID_PARA;
D
stmt  
dapan1121 已提交
1357
    return 0;
D
stmt  
dapan1121 已提交
1358 1359
  }

D
stmt  
dapan1121 已提交
1360
  return stmtAffectedRows(stmt);
1361 1362
}

D
dapan1121 已提交
1363 1364 1365 1366 1367 1368 1369 1370 1371 1372
int taos_stmt_affected_rows_once(TAOS_STMT *stmt) {
  if (stmt == NULL) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return 0;
  }

  return stmtAffectedRowsOnce(stmt);
}

D
stmt  
dapan1121 已提交
1373 1374
int taos_stmt_close(TAOS_STMT *stmt) {
  if (stmt == NULL) {
D
stmt  
dapan1121 已提交
1375
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1376 1377 1378 1379
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

D
stmt  
dapan1121 已提交
1380
  return stmtClose(stmt);
D
stmt  
dapan1121 已提交
1381
}