clientMain.c 39.5 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
289 290
    tscError("invalid result passed to taos_fetch_row");
    return NULL;
L
Liu Jicong 已提交
291 292
  }
  return NULL;
293
}
H
Haojun Liao 已提交
294

L
Liu Jicong 已提交
295
int taos_print_row(char *str, TAOS_ROW row, TAOS_FIELD *fields, int num_fields) {
H
Haojun Liao 已提交
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 353
  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 已提交
354
        int32_t charLen = varDataLen((char *)row[i] - VARSTR_HEADER_SIZE);
H
Haojun Liao 已提交
355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374
        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 已提交
375
  str[len] = 0;
H
Haojun Liao 已提交
376 377 378

  return len;
}
379

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

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

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

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

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

403
// todo intergrate with tDataTypes
404 405
const char *taos_data_type(int type) {
  switch (type) {
L
Liu Jicong 已提交
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 432
    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";
433 434
  }
}
435 436 437

const char *taos_get_client_info() { return version; }

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

L
Liu Jicong 已提交
543 544 545 546
    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;
    }
547

548
    doAsyncFetchRows(pRequest, false, true);
549

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

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

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

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

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

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

589 590 591 592 593
  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 已提交
594
  }
595

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

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

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

  return 0;
605 606
}

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

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

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

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

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

  int code = taos_errno(pObj);
630

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

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

642 643
  resetConnectDB(pTscObj);

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

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

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

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

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

664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686
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);
687 688 689 690
  taosMemoryFree(pWrapper);
}

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

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

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

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

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

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

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

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

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

735
    if (NEED_CLIENT_HANDLE_ERROR(code)) {
736 737
      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);
738 739 740 741
      pRequest->prevCode = code;
      doAsyncQuery(pRequest, true);
      return;
    }
742

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

751 752 753 754 755 756 757 758 759 760
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;
}

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

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

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

dengyihao's avatar
dengyihao 已提交
774 775 776 777 778 779 780 781 782 783 784 785 786 787
  **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)),
788
                           .enableSysInfo = pTscObj->sysInfo,
dengyihao's avatar
dengyihao 已提交
789 790
                           .async = true,
                           .svrVer = pTscObj->sVer,
791 792
                           .nodeOffline = (pTscObj->pAppInfo->onlineDnodes < pTscObj->pAppInfo->totalDnodes),
                           .allocatorId = pRequest->allocatorRefId};
793 794 795
  return TSDB_CODE_SUCCESS;
}

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

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

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

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

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

823 824 825 826 827 828 829 830 831 832 833 834 835 836
  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 已提交
837

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

847 848 849 850 851
  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 已提交
852

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

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

859 860 861 862 863 864 865 866
  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);
  }
867 868
}

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

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

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

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

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

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

894
  pRequest->code =
D
dapan1121 已提交
895
      setQueryResultFromRsp(pResultInfo, (SRetrieveTableRsp *)pResultInfo->pData, pResultInfo->convertUcs4, true);
896 897 898
  if (pRequest->code != TSDB_CODE_SUCCESS) {
    pResultInfo->numOfRows = 0;
    pRequest->code = code;
899 900 901 902 903 904
    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 已提交
905 906 907

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

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

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

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

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

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

931
  // all data has returned to App already, no need to try again
D
dapan1121 已提交
932 933 934 935 936 937 938 939 940
  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;
      }
941
    } else {
D
dapan1121 已提交
942
      pResultInfo->numOfRows = 0;
943
    }
944

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

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

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

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

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

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

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

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

  return pRequest->body.resInfo.pData;
}

X
Xiaoyu Wang 已提交
980
int taos_get_db_route_info(TAOS *taos, const char *db, TAOS_DB_ROUTE_INFO *dbInfo) {
981 982 983 984 985 986 987 988 989 990 991
  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 已提交
992 993 994 995
  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);
996
  if (code != TSDB_CODE_SUCCESS) {
997
    terrno = code;
998 999 1000
    return terrno;
  }

X
Xiaoyu Wang 已提交
1001
  STscObj  *pTscObj = pRequest->pTscObj;
1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014
  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 已提交
1015

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

_return:

  terrno = code;

  destroyRequest(pRequest);
  return code;
}

X
Xiaoyu Wang 已提交
1029
int taos_get_table_vgId(TAOS *taos, const char *db, const char *table, int *vgId) {
1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040
  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 已提交
1041 1042 1043 1044
  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);
1045 1046 1047 1048
  if (code != TSDB_CODE_SUCCESS) {
    return terrno;
  }

X
Xiaoyu Wang 已提交
1049
  STscObj  *pTscObj = pRequest->pTscObj;
1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074
  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;
1075

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

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

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

D
dapan1121 已提交
1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103
  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;
  }

1104 1105 1106 1107 1108
  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 已提交
1109 1110
  }

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

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

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

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

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

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

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

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

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

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

  return pStmt;
1155 1156
}

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

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

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

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

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

  return TSDB_CODE_SUCCESS;
1184 1185
}

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

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

D
dapan1121 已提交
1196 1197 1198 1199 1200 1201 1202 1203 1204 1205
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);
}

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

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

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

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

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

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

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

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

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

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

D
dapan1121 已提交
1257 1258 1259 1260 1261 1262 1263 1264
  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 已提交
1265
  return stmtBindBatch(stmt, bind, -1);
1266 1267
}

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

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

  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;
  }
1288

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

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

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

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

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

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

  return stmtIsInsert(stmt, insert);
1320 1321
}

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

  return stmtGetParamNum(stmt, nums);
1330 1331
}

D
dapan1121 已提交
1332 1333 1334 1335 1336 1337 1338 1339 1340 1341
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 已提交
1342
TAOS_RES *taos_stmt_use_result(TAOS_STMT *stmt) {
D
stmt  
dapan1121 已提交
1343
  if (stmt == NULL) {
D
stmt  
dapan1121 已提交
1344
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1345
    terrno = TSDB_CODE_INVALID_PARA;
D
stmt  
dapan1121 已提交
1346
    return NULL;
D
stmt  
dapan1121 已提交
1347 1348
  }

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

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

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

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

D
dapan1121 已提交
1364 1365 1366 1367 1368 1369 1370 1371 1372 1373
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 已提交
1374 1375
int taos_stmt_close(TAOS_STMT *stmt) {
  if (stmt == NULL) {
D
stmt  
dapan1121 已提交
1376
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1377 1378 1379 1380
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

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