clientMain.c 36.0 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"
L
Liu Jicong 已提交
23
#include "scheduler.h"
24
#include "tglobal.h"
L
Liu Jicong 已提交
25 26 27
#include "tmsg.h"
#include "tref.h"
#include "trpc.h"
S
version  
Shengliang Guan 已提交
28
#include "version.h"
29
#include "qworker.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_RPC_NETWORK_UNAVAIL;
150 151 152
    return terrno;
  }

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

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

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

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

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

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

L
Liu Jicong 已提交
185 186
  if (TD_RES_QUERY(res)) {
    SRequestObj *pRequest = (SRequestObj *)res;
D
dapan1121 已提交
187
    tscDebug("0x%" PRIx64 " taos_free_result start to free query", pRequest->requestId);
L
Liu Jicong 已提交
188
    destroyRequest(pRequest);
189
  } else if (TD_RES_TMQ_METADATA(res)) {
L
Liu Jicong 已提交
190 191 192 193 194 195 196 197 198 199 200 201
    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 已提交
202 203 204 205
  } 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 已提交
206 207
    if (pRsp->rsp.withTbName) taosArrayDestroyP(pRsp->rsp.blockTbName, taosMemoryFree);
    if (pRsp->rsp.withSchema) taosArrayDestroyP(pRsp->rsp.blockSchema, (FDelete)tDeleteSSchemaWrapper);
L
Liu Jicong 已提交
208 209
    pRsp->resInfo.pRspMsg = NULL;
    doFreeReqResultInfo(&pRsp->resInfo);
210
    taosMemoryFree(pRsp);
L
Liu Jicong 已提交
211 212 213 214
  } else if (TD_RES_TMQ_META(res)) {
    SMqMetaRspObj *pRspObj = (SMqMetaRspObj *)res;
    taosMemoryFree(pRspObj->metaRsp.metaRsp);
    taosMemoryFree(pRspObj);
L
Liu Jicong 已提交
215
  }
216 217
}

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

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

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

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

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

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

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

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

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

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

#if SYNC_ON_TOP_OF_ASYNC
    return doAsyncFetchRows(pRequest, true, true);
#else
266
    return doFetchRows(pRequest, true, true);
267
#endif
L
Liu Jicong 已提交
268

269
  } else if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) {
dengyihao's avatar
dengyihao 已提交
270
    SMqRspObj      *msg = ((SMqRspObj *)res);
L
Liu Jicong 已提交
271 272 273 274 275 276
    SReqResultInfo *pResultInfo;
    if (msg->resIter == -1) {
      pResultInfo = tmqGetNextResInfo(res, true);
    } else {
      pResultInfo = tmqGetCurResInfo(res);
    }
277

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

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

  return len;
}
382

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

L
Liu Jicong 已提交
388 389
  SReqResultInfo *pResInfo = tscGetCurResInfo(res);
  return pResInfo->length;
390 391
}

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

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

L
Liu Jicong 已提交
402 403
  SReqResultInfo *pResInfo = tscGetCurResInfo(res);
  return &pResInfo->row;
404 405
}

406
// todo intergrate with tDataTypes
407 408
const char *taos_data_type(int type) {
  switch (type) {
L
Liu Jicong 已提交
409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435
    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";
436 437
  }
}
438 439 440

const char *taos_get_client_info() { return version; }

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

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

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

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

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

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

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

L
Liu Jicong 已提交
483 484
  TAOS_RES *pRequest = taos_query(taos, sql);
  int32_t   code = taos_errno(pRequest);
485 486

  taos_free_result(pRequest);
L
Liu Jicong 已提交
487
  releaseTscObj(*(int64_t *)taos);
488 489 490 491
  return code;
}

void taos_stop_query(TAOS_RES *res) {
492
  if (res == NULL || TD_RES_TMQ(res) || TD_RES_TMQ_META(res) || TD_RES_TMQ_METADATA(res)) {
493 494 495
    return;
  }

L
Liu Jicong 已提交
496
  SRequestObj *pRequest = (SRequestObj *)res;
D
dapan1121 已提交
497
  pRequest->killed = true;
L
Liu Jicong 已提交
498 499

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

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

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

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

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

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

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

L
Liu Jicong 已提交
540 541
  if (TD_RES_QUERY(res)) {
    SRequestObj *pRequest = (SRequestObj *)res;
542

L
Liu Jicong 已提交
543 544
    (*rows) = NULL;
    (*numOfRows) = 0;
H
Haojun Liao 已提交
545

L
Liu Jicong 已提交
546 547 548 549
    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;
    }
550

551
#if SYNC_ON_TOP_OF_ASYNC
552
    doAsyncFetchRows(pRequest, false, true);
553 554 555
#else
    doFetchRows(pRequest, true, true);
#endif
556

L
Liu Jicong 已提交
557 558 559
    // TODO refactor
    SReqResultInfo *pResultInfo = &pRequest->body.resInfo;
    pResultInfo->current = pResultInfo->numOfRows;
560

L
Liu Jicong 已提交
561 562 563
    (*rows) = pResultInfo->row;
    (*numOfRows) = pResultInfo->numOfRows;
    return pRequest->code;
564
  } else if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) {
L
Liu Jicong 已提交
565
    SReqResultInfo *pResultInfo = tmqGetNextResInfo(res, true);
L
Liu Jicong 已提交
566
    if (pResultInfo == NULL) return -1;
H
Haojun Liao 已提交
567

L
Liu Jicong 已提交
568 569 570
    pResultInfo->current = pResultInfo->numOfRows;
    (*rows) = pResultInfo->row;
    (*numOfRows) = pResultInfo->numOfRows;
H
Haojun Liao 已提交
571
    return 0;
L
Liu Jicong 已提交
572 573 574
  } else {
    ASSERT(0);
    return -1;
H
Haojun Liao 已提交
575
  }
L
Liu Jicong 已提交
576
}
H
Haojun Liao 已提交
577

L
Liu Jicong 已提交
578
int taos_fetch_raw_block(TAOS_RES *res, int *numOfRows, void **pData) {
L
Liu Jicong 已提交
579
  if (res == NULL || TD_RES_TMQ_META(res)) {
H
Haojun Liao 已提交
580 581
    return 0;
  }
582

583
  if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) {
L
Liu Jicong 已提交
584
    SReqResultInfo *pResultInfo = tmqGetNextResInfo(res, false);
L
Liu Jicong 已提交
585 586 587 588
    if (pResultInfo == NULL) {
      (*numOfRows) = 0;
      return 0;
    }
H
Haojun Liao 已提交
589

L
Liu Jicong 已提交
590 591 592 593
    pResultInfo->current = pResultInfo->numOfRows;
    (*numOfRows) = pResultInfo->numOfRows;
    (*pData) = (void *)pResultInfo->pData;
    return 0;
594
  }
L
Liu Jicong 已提交
595

596 597 598 599 600
  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 已提交
601
  }
602

603 604 605
#if SYNC_ON_TOP_OF_ASYNC
  doAsyncFetchRows(pRequest, false, false);
#else
606
  doFetchRows(pRequest, false, false);
607
#endif
608 609 610 611 612 613 614 615

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

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

  return 0;
616 617
}

H
Haojun Liao 已提交
618
int *taos_get_column_data_offset(TAOS_RES *res, int columnIndex) {
L
Liu Jicong 已提交
619
  if (res == NULL || TD_RES_TMQ_META(res)) {
H
Haojun Liao 已提交
620 621 622
    return 0;
  }

L
Liu Jicong 已提交
623
  int32_t numOfFields = taos_num_fields(res);
H
Haojun Liao 已提交
624 625 626 627
  if (columnIndex < 0 || columnIndex >= numOfFields || numOfFields == 0) {
    return 0;
  }

L
Liu Jicong 已提交
628
  SReqResultInfo *pResInfo = tscGetCurResInfo(res);
dengyihao's avatar
dengyihao 已提交
629
  TAOS_FIELD     *pField = &pResInfo->userFields[columnIndex];
H
Haojun Liao 已提交
630 631 632 633
  if (!IS_VAR_DATA_TYPE(pField->type)) {
    return 0;
  }

L
Liu Jicong 已提交
634
  return pResInfo->pCol[columnIndex].offset;
H
Haojun Liao 已提交
635 636
}

637 638
int taos_validate_sql(TAOS *taos, const char *sql) {
  TAOS_RES *pObj = taosQueryImpl(taos, sql, true);
D
dapan1121 已提交
639 640

  int code = taos_errno(pObj);
641

D
dapan1121 已提交
642 643 644
  taos_free_result(pObj);
  return code;
}
645

646
void taos_reset_current_db(TAOS *taos) {
L
Liu Jicong 已提交
647
  STscObj *pTscObj = acquireTscObj(*(int64_t *)taos);
648 649
  if (pTscObj == NULL) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
H
Haojun Liao 已提交
650 651 652
    return;
  }

653 654
  resetConnectDB(pTscObj);

L
Liu Jicong 已提交
655
  releaseTscObj(*(int64_t *)taos);
656 657
}

658
const char *taos_get_server_info(TAOS *taos) {
L
Liu Jicong 已提交
659
  STscObj *pTscObj = acquireTscObj(*(int64_t *)taos);
660 661
  if (pTscObj == NULL) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
662 663 664
    return NULL;
  }

L
Liu Jicong 已提交
665
  releaseTscObj(*(int64_t *)taos);
666

D
dapan1121 已提交
667
  return pTscObj->sDetailVer;
668 669
}

670
typedef struct SqlParseWrapper {
671
  SParseContext *pCtx;
672
  SCatalogReq    catalogReq;
dengyihao's avatar
dengyihao 已提交
673
  SRequestObj   *pRequest;
674 675
} SqlParseWrapper;

676 677 678 679 680
static void destoryTablesReq(void *p) {
  STablesReq *pRes = (STablesReq *)p;
  taosArrayDestroy(pRes->pTables);
}

681 682 683 684
static void destorySqlParseWrapper(SqlParseWrapper *pWrapper) {
  taosArrayDestroy(pWrapper->catalogReq.pDbVgroup);
  taosArrayDestroy(pWrapper->catalogReq.pDbCfg);
  taosArrayDestroy(pWrapper->catalogReq.pDbInfo);
685 686
  taosArrayDestroyEx(pWrapper->catalogReq.pTableMeta, destoryTablesReq);
  taosArrayDestroyEx(pWrapper->catalogReq.pTableHash, destoryTablesReq);
687 688 689 690
  taosArrayDestroy(pWrapper->catalogReq.pUdf);
  taosArrayDestroy(pWrapper->catalogReq.pIndex);
  taosArrayDestroy(pWrapper->catalogReq.pUser);
  taosArrayDestroy(pWrapper->catalogReq.pTableIndex);
D
dapan1121 已提交
691 692
  taosArrayDestroy(pWrapper->pCtx->pTableMetaPos);
  taosArrayDestroy(pWrapper->pCtx->pTableVgroupPos);
693 694 695 696 697 698
  taosMemoryFree(pWrapper->pCtx);
  taosMemoryFree(pWrapper);
}

void retrieveMetaCallback(SMetaData *pResultMeta, void *param, int32_t code) {
  SqlParseWrapper *pWrapper = (SqlParseWrapper *)param;
dengyihao's avatar
dengyihao 已提交
699
  SRequestObj     *pRequest = pWrapper->pRequest;
X
Xiaoyu Wang 已提交
700
  SQuery          *pQuery = pRequest->pQuery;
701

D
dapan1121 已提交
702 703
  pRequest->metric.ctgEnd = taosGetTimestampUs();

704 705
  if (code == TSDB_CODE_SUCCESS) {
    code = qAnalyseSqlSemantic(pWrapper->pCtx, &pWrapper->catalogReq, pResultMeta, pQuery);
706
    pRequest->stableQuery = pQuery->stableQuery;
D
dapan1121 已提交
707
    if (pQuery->pRoot) {
dengyihao's avatar
dengyihao 已提交
708
      pRequest->stmtType = pQuery->pRoot->type;
D
dapan1121 已提交
709
    }
710 711
  }

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

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
    destorySqlParseWrapper(pWrapper);
725

dengyihao's avatar
dengyihao 已提交
726 727
    tscDebug("0x%" PRIx64 " analysis semantics completed, start async query, reqId:0x%" PRIx64, pRequest->self,
             pRequest->requestId);
D
dapan1121 已提交
728
    launchAsyncQuery(pRequest, pQuery, pResultMeta);
729
  } else {
730
    destorySqlParseWrapper(pWrapper);
X
Xiaoyu Wang 已提交
731 732
    qDestroyQuery(pRequest->pQuery);
    pRequest->pQuery = NULL;
733
    if (NEED_CLIENT_HANDLE_ERROR(code)) {
734 735
      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);
736 737 738 739
      pRequest->prevCode = code;
      doAsyncQuery(pRequest, true);
      return;
    }
740

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

749
void taos_query_a(TAOS *taos, const char *sql, __taos_async_fn_t fp, void *param) {
dengyihao's avatar
dengyihao 已提交
750
  int64_t connId = *(int64_t *)taos;
751
  taosAsyncQueryImpl(connId, sql, fp, param, false);
752
}
753

754
int32_t createParseContext(const SRequestObj *pRequest, SParseContext **pCxt) {
755 756 757 758 759 760 761
  const STscObj *pTscObj = pRequest->pTscObj;

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

dengyihao's avatar
dengyihao 已提交
762 763 764 765 766 767 768 769 770 771 772 773 774 775
  **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)),
776
                           .enableSysInfo = pTscObj->sysInfo,
dengyihao's avatar
dengyihao 已提交
777 778
                           .async = true,
                           .svrVer = pTscObj->sVer,
779 780
                           .nodeOffline = (pTscObj->pAppInfo->onlineDnodes < pTscObj->pAppInfo->totalDnodes),
                           .allocatorId = pRequest->allocatorRefId};
781 782 783
  return TSDB_CODE_SUCCESS;
}

784 785
void doAsyncQuery(SRequestObj *pRequest, bool updateMetaForce) {
  SParseContext *pCxt = NULL;
dengyihao's avatar
dengyihao 已提交
786
  STscObj       *pTscObj = pRequest->pTscObj;
787
  int32_t        code = 0;
788 789

  if (pRequest->retry++ > REQUEST_TOTAL_EXEC_TIMES) {
790
    code = pRequest->prevCode;
791 792 793
    goto _error;
  }

794
  code = createParseContext(pRequest, &pCxt);
795 796 797
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
798 799 800 801 802

  pCxt->mgmtEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
  code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCxt->pCatalog);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
803 804
  }

D
dapan1121 已提交
805 806
  pRequest->metric.syntaxStart = taosGetTimestampUs();

D
dapan1121 已提交
807
  SCatalogReq catalogReq = {.forceUpdate = updateMetaForce, .qNodeRequired = qnodeRequired(pRequest)};
X
Xiaoyu Wang 已提交
808
  code = qParseSqlSyntax(pCxt, &pRequest->pQuery, &catalogReq);
809 810 811 812
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

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

D
dapan1121 已提交
815 816 817
  if (!updateMetaForce) {
    STscObj            *pTscObj = pRequest->pTscObj;
    SAppClusterSummary *pActivity = &pTscObj->pAppInfo->summary;
X
Xiaoyu Wang 已提交
818
    if (NULL == pRequest->pQuery->pRoot) {
dengyihao's avatar
dengyihao 已提交
819
      atomic_add_fetch_64((int64_t *)&pActivity->numOfInsertsReq, 1);
X
Xiaoyu Wang 已提交
820
    } else if (QUERY_NODE_SELECT_STMT == pRequest->pQuery->pRoot->type) {
dengyihao's avatar
dengyihao 已提交
821
      atomic_add_fetch_64((int64_t *)&pActivity->numOfQueryReq, 1);
D
dapan1121 已提交
822 823 824
    }
  }

825
  SqlParseWrapper *pWrapper = taosMemoryCalloc(1, sizeof(SqlParseWrapper));
826
  if (pWrapper == NULL) {
827
    code = TSDB_CODE_OUT_OF_MEMORY;
828 829 830
    goto _error;
  }

831 832 833 834
  pWrapper->pCtx = pCxt;
  pWrapper->pRequest = pRequest;
  pWrapper->catalogReq = catalogReq;

X
Xiaoyu Wang 已提交
835
  SRequestConnInfo conn = {.pTrans = pCxt->pTransporter,
D
dapan1121 已提交
836 837 838 839
                           .requestId = pCxt->requestId,
                           .requestObjRefId = pCxt->requestRid,
                           .mgmtEps = pCxt->mgmtEpSet};

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

D
dapan1121 已提交
842
  code = catalogAsyncGetAllMeta(pCxt->pCatalog, &conn, &catalogReq, retrieveMetaCallback, pWrapper,
X
Xiaoyu Wang 已提交
843
                                &pRequest->body.queryJob);
dengyihao's avatar
dengyihao 已提交
844
  pCxt = NULL;
845 846
  if (code == TSDB_CODE_SUCCESS) {
    return;
847 848
  }

849 850 851
_error:
  tscError("0x%" PRIx64 " error happens, code:%d - %s, reqId:0x%" PRIx64, pRequest->self, code, tstrerror(code),
           pRequest->requestId);
dengyihao's avatar
dengyihao 已提交
852 853
  taosMemoryFree(pCxt);

854 855
  terrno = code;
  pRequest->code = code;
856
  pRequest->body.queryFp(pRequest->body.param, pRequest, code);
857 858
}

859 860
static void fetchCallback(void *pResult, void *param, int32_t code) {
  SRequestObj *pRequest = (SRequestObj *)param;
861

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

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

867 868 869 870 871
  pResultInfo->pData = pResult;
  pResultInfo->numOfRows = 0;

  if (code != TSDB_CODE_SUCCESS) {
    pRequest->code = code;
872
    taosMemoryFreeClear(pResultInfo->pData);
873 874 875 876 877
    pRequest->body.fetchFp(pRequest->body.param, pRequest, 0);
    return;
  }

  if (pRequest->code != TSDB_CODE_SUCCESS) {
878
    taosMemoryFreeClear(pResultInfo->pData);
879
    pRequest->body.fetchFp(pRequest->body.param, pRequest, 0);
880
    return;
881 882
  }

883
  pRequest->code =
D
dapan1121 已提交
884
      setQueryResultFromRsp(pResultInfo, (SRetrieveTableRsp *)pResultInfo->pData, pResultInfo->convertUcs4, true);
885 886 887
  if (pRequest->code != TSDB_CODE_SUCCESS) {
    pResultInfo->numOfRows = 0;
    pRequest->code = code;
888 889 890 891 892 893
    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 已提交
894 895 896

    STscObj            *pTscObj = pRequest->pTscObj;
    SAppClusterSummary *pActivity = &pTscObj->pAppInfo->summary;
dengyihao's avatar
dengyihao 已提交
897
    atomic_add_fetch_64((int64_t *)&pActivity->fetchBytes, pRequest->body.resInfo.payloadLen);
898 899 900 901 902
  }

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

903
void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {
904
  ASSERT(res != NULL && fp != NULL);
L
Liu Jicong 已提交
905
  ASSERT(TD_RES_QUERY(res));
906 907 908

  SRequestObj *pRequest = res;
  pRequest->body.fetchFp = fp;
909
  pRequest->body.param = param;
910 911

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

  // this query has no results or error exists, return directly
  if (taos_num_fields(pRequest) == 0 || pRequest->code != TSDB_CODE_SUCCESS) {
915 916 917 918 919
    pResultInfo->numOfRows = 0;
    pRequest->body.fetchFp(param, pRequest, pResultInfo->numOfRows);
    return;
  }

920
  // all data has returned to App already, no need to try again
D
dapan1121 已提交
921 922 923 924 925 926 927 928 929 930
  if (pResultInfo->completed) {
    // it is a local executed query, no need to do async fetch
    if (QUERY_EXEC_MODE_LOCAL == pRequest->body.execMode) {
      ASSERT(pResultInfo->numOfRows >= 0);
      if (pResultInfo->localResultFetched) {
        pResultInfo->numOfRows = 0;
        pResultInfo->current = 0;
      } else {
        pResultInfo->localResultFetched = true;
      }
931
    } else {
D
dapan1121 已提交
932
      pResultInfo->numOfRows = 0;
933
    }
934

D
dapan1121 已提交
935
    pRequest->body.fetchFp(param, pRequest, pResultInfo->numOfRows);
936
    return;
937 938
  }

D
dapan1121 已提交
939
  SSchedulerReq req = {
dengyihao's avatar
dengyihao 已提交
940 941 942
      .syncReq = false,
      .fetchFp = fetchCallback,
      .cbParam = pRequest,
D
dapan1121 已提交
943
  };
944

D
dapan1121 已提交
945
  schedulerFetchRows(pRequest->body.queryJob, &req);
L
Liu Jicong 已提交
946
}
947

948
void taos_fetch_raw_block_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {
949
  ASSERT(res != NULL && fp != NULL);
L
Liu Jicong 已提交
950
  ASSERT(TD_RES_QUERY(res));
951

dengyihao's avatar
dengyihao 已提交
952
  SRequestObj    *pRequest = res;
953 954 955
  SReqResultInfo *pResultInfo = &pRequest->body.resInfo;

  // set the current block is all consumed
956 957
  pResultInfo->convertUcs4 = false;

958 959
  // it is a local executed query, no need to do async fetch
  taos_fetch_rows_a(pRequest, fp, param);
960 961
}

962
const void *taos_get_raw_block(TAOS_RES *res) {
H
Haojun Liao 已提交
963
  ASSERT(res != NULL);
L
Liu Jicong 已提交
964
  ASSERT(TD_RES_QUERY(res));
965
  SRequestObj *pRequest = res;
H
Haojun Liao 已提交
966 967 968 969

  return pRequest->body.resInfo.pData;
}

970
int taos_load_table_info(TAOS *taos, const char *tableNameList) {
D
dapan1121 已提交
971 972 973 974 975
  if (NULL == taos) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return terrno;
  }

976
  int64_t       connId = *(int64_t *)taos;
D
dapan1121 已提交
977
  const int32_t MAX_TABLE_NAME_LENGTH = 12 * 1024 * 1024;  // 12MB list
978
  int32_t       code = 0;
dengyihao's avatar
dengyihao 已提交
979
  SRequestObj  *pRequest = NULL;
980 981
  SCatalogReq   catalogReq = {0};

D
dapan1121 已提交
982 983 984 985 986 987 988 989 990 991 992 993
  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;
  }

994 995 996 997 998
  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 已提交
999 1000
  }

dengyihao's avatar
dengyihao 已提交
1001
  STscObj *pTscObj = pRequest->pTscObj;
D
dapan1121 已提交
1002 1003 1004 1005 1006
  code = transferTableNameList(tableNameList, pTscObj->acctId, pTscObj->db, &catalogReq.pTableMeta);
  if (code) {
    goto _return;
  }

1007
  SCatalog *pCtg = NULL;
D
dapan1121 已提交
1008 1009 1010 1011 1012
  code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCtg);
  if (code != TSDB_CODE_SUCCESS) {
    goto _return;
  }

1013 1014
  SRequestConnInfo conn = {
      .pTrans = pTscObj->pAppInfo->pTransporter, .requestId = pRequest->requestId, .requestObjRefId = pRequest->self};
D
dapan1121 已提交
1015 1016 1017

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

D
dapan1121 已提交
1018
  code = catalogAsyncGetAllMeta(pCtg, &conn, &catalogReq, syncCatalogFn, pRequest->body.param, NULL);
D
dapan1121 已提交
1019 1020 1021 1022
  if (code) {
    goto _return;
  }

dengyihao's avatar
dengyihao 已提交
1023
  SSyncQueryParam *pParam = pRequest->body.param;
1024
  tsem_wait(&pParam->sem);
D
dapan1121 已提交
1025 1026 1027 1028 1029

_return:
  taosArrayDestroy(catalogReq.pTableMeta);
  destroyRequest(pRequest);
  return code;
1030 1031
}

L
Liu Jicong 已提交
1032
TAOS_STMT *taos_stmt_init(TAOS *taos) {
L
Liu Jicong 已提交
1033
  STscObj *pObj = acquireTscObj(*(int64_t *)taos);
1034 1035 1036
  if (NULL == pObj) {
    tscError("invalid parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_TSC_DISCONNECTED;
D
stmt  
dapan1121 已提交
1037 1038 1039
    return NULL;
  }

L
Liu Jicong 已提交
1040 1041 1042
  TAOS_STMT *pStmt = stmtInit(pObj);

  releaseTscObj(*(int64_t *)taos);
1043 1044

  return pStmt;
1045 1046
}

D
stmt  
dapan1121 已提交
1047 1048
int taos_stmt_prepare(TAOS_STMT *stmt, const char *sql, unsigned long length) {
  if (stmt == NULL || sql == NULL) {
D
stmt  
dapan1121 已提交
1049
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1050 1051 1052 1053
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

D
stmt  
dapan1121 已提交
1054
  return stmtPrepare(stmt, sql, length);
1055 1056
}

D
dapan1121 已提交
1057
int taos_stmt_set_tbname_tags(TAOS_STMT *stmt, const char *name, TAOS_MULTI_BIND *tags) {
D
stmt  
dapan1121 已提交
1058
  if (stmt == NULL || name == NULL) {
D
stmt  
dapan1121 已提交
1059
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1060 1061 1062 1063
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

D
stmt  
dapan1121 已提交
1064 1065 1066 1067 1068 1069 1070 1071 1072 1073
  int32_t code = stmtSetTbName(stmt, name);
  if (code) {
    return code;
  }

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

  return TSDB_CODE_SUCCESS;
1074 1075
}

D
stmt  
dapan1121 已提交
1076 1077
int taos_stmt_set_tbname(TAOS_STMT *stmt, const char *name) {
  if (stmt == NULL || name == NULL) {
D
stmt  
dapan1121 已提交
1078
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1079
    terrno = TSDB_CODE_INVALID_PARA;
D
stmt  
dapan1121 已提交
1080
    return terrno;
D
stmt  
dapan1121 已提交
1081 1082
  }

D
stmt  
dapan1121 已提交
1083
  return stmtSetTbName(stmt, name);
1084 1085
}

D
dapan1121 已提交
1086 1087 1088 1089 1090 1091 1092 1093 1094 1095
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);
}

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

1098
int taos_stmt_get_tag_fields(TAOS_STMT *stmt, int *fieldNum, TAOS_FIELD_E **fields) {
D
dapan1121 已提交
1099 1100 1101 1102 1103
  if (stmt == NULL || NULL == fieldNum) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }
1104

D
dapan1121 已提交
1105 1106 1107
  return stmtGetTagFields(stmt, fieldNum, fields);
}

1108
int taos_stmt_get_col_fields(TAOS_STMT *stmt, int *fieldNum, TAOS_FIELD_E **fields) {
D
dapan1121 已提交
1109 1110 1111 1112 1113
  if (stmt == NULL || NULL == fieldNum) {
    tscError("NULL parameter for %s", __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }
1114

D
dapan1121 已提交
1115 1116 1117
  return stmtGetColFields(stmt, fieldNum, fields);
}

D
dapan1121 已提交
1118
int taos_stmt_bind_param(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind) {
D
stmt  
dapan1121 已提交
1119
  if (stmt == NULL || bind == NULL) {
D
stmt  
dapan1121 已提交
1120
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1121
    terrno = TSDB_CODE_INVALID_PARA;
D
stmt  
dapan1121 已提交
1122
    return terrno;
D
stmt  
dapan1121 已提交
1123
  }
1124

D
stmt  
dapan1121 已提交
1125 1126 1127 1128 1129
  if (bind->num > 1) {
    tscError("invalid bind number %d for %s", bind->num, __FUNCTION__);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }
1130

D
stmt  
dapan1121 已提交
1131
  return stmtBindBatch(stmt, bind, -1);
1132 1133
}

D
dapan1121 已提交
1134
int taos_stmt_bind_param_batch(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind) {
D
stmt  
dapan1121 已提交
1135
  if (stmt == NULL || bind == NULL) {
D
stmt  
dapan1121 已提交
1136
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1137 1138 1139 1140
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

D
stmt  
dapan1121 已提交
1141 1142
  if (bind->num <= 0 || bind->num > INT16_MAX) {
    tscError("invalid bind num %d", bind->num);
D
stmt  
dapan1121 已提交
1143 1144 1145 1146
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

D
dapan1121 已提交
1147 1148 1149 1150 1151 1152 1153 1154
  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 已提交
1155
  return stmtBindBatch(stmt, bind, -1);
1156 1157
}

D
dapan1121 已提交
1158
int taos_stmt_bind_single_param_batch(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind, int colIdx) {
D
stmt  
dapan1121 已提交
1159
  if (stmt == NULL || bind == NULL) {
D
stmt  
dapan1121 已提交
1160
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1161 1162 1163 1164
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

D
stmt  
dapan1121 已提交
1165
  if (colIdx < 0) {
D
stmt  
dapan1121 已提交
1166 1167 1168 1169
    tscError("invalid bind column idx %d", colIdx);
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }
D
dapan1121 已提交
1170 1171 1172 1173 1174 1175 1176 1177

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

D
stmt  
dapan1121 已提交
1179
  return stmtBindBatch(stmt, bind, colIdx);
1180 1181
}

D
stmt  
dapan1121 已提交
1182 1183
int taos_stmt_add_batch(TAOS_STMT *stmt) {
  if (stmt == NULL) {
D
stmt  
dapan1121 已提交
1184
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1185 1186 1187 1188
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

D
stmt  
dapan1121 已提交
1189
  return stmtAddBatch(stmt);
1190 1191
}

D
stmt  
dapan1121 已提交
1192 1193
int taos_stmt_execute(TAOS_STMT *stmt) {
  if (stmt == NULL) {
D
stmt  
dapan1121 已提交
1194
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1195 1196 1197 1198
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

D
stmt  
dapan1121 已提交
1199
  return stmtExec(stmt);
1200 1201
}

1202
int taos_stmt_is_insert(TAOS_STMT *stmt, int *insert) {
D
stmt  
dapan1121 已提交
1203
  if (stmt == NULL || insert == NULL) {
D
stmt  
dapan1121 已提交
1204
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1205 1206 1207 1208 1209
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  return stmtIsInsert(stmt, insert);
1210 1211
}

1212
int taos_stmt_num_params(TAOS_STMT *stmt, int *nums) {
D
stmt  
dapan1121 已提交
1213
  if (stmt == NULL || nums == NULL) {
D
stmt  
dapan1121 已提交
1214
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1215 1216 1217 1218 1219
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

  return stmtGetParamNum(stmt, nums);
1220 1221
}

D
dapan1121 已提交
1222 1223 1224 1225 1226 1227 1228 1229 1230 1231
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 已提交
1232
TAOS_RES *taos_stmt_use_result(TAOS_STMT *stmt) {
D
stmt  
dapan1121 已提交
1233
  if (stmt == NULL) {
D
stmt  
dapan1121 已提交
1234
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1235
    terrno = TSDB_CODE_INVALID_PARA;
D
stmt  
dapan1121 已提交
1236
    return NULL;
D
stmt  
dapan1121 已提交
1237 1238
  }

D
stmt  
dapan1121 已提交
1239
  return stmtUseResult(stmt);
1240 1241
}

1242
char *taos_stmt_errstr(TAOS_STMT *stmt) { return (char *)stmtErrstr(stmt); }
1243

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

D
stmt  
dapan1121 已提交
1251
  return stmtAffectedRows(stmt);
1252 1253
}

D
dapan1121 已提交
1254 1255 1256 1257 1258 1259 1260 1261 1262 1263
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 已提交
1264 1265
int taos_stmt_close(TAOS_STMT *stmt) {
  if (stmt == NULL) {
D
stmt  
dapan1121 已提交
1266
    tscError("NULL parameter for %s", __FUNCTION__);
D
stmt  
dapan1121 已提交
1267 1268 1269 1270
    terrno = TSDB_CODE_INVALID_PARA;
    return terrno;
  }

D
stmt  
dapan1121 已提交
1271
  return stmtClose(stmt);
D
stmt  
dapan1121 已提交
1272
}