clientImpl.c 64.0 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14
/*
 * 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/>.
 */
15

16
#include "cJSON.h"
17
#include "clientInt.h"
18
#include "clientLog.h"
19
#include "command.h"
20
#include "scheduler.h"
H
Haojun Liao 已提交
21
#include "tdatablock.h"
22
#include "tdataformat.h"
23
#include "tdef.h"
24
#include "tglobal.h"
25
#include "tmsgtype.h"
H
Haojun Liao 已提交
26
#include "tpagedbuf.h"
27
#include "tref.h"
X
Xiaoyu Wang 已提交
28

S
Shengliang Guan 已提交
29
static int32_t       initEpSetFromCfg(const char* firstEp, const char* secondEp, SCorEpSet* pEpSet);
30
static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest);
dengyihao's avatar
dengyihao 已提交
31
static void          destroySendMsgInfo(SMsgSendInfo* pMsgBody);
H
Haojun Liao 已提交
32

33
static bool stringLengthCheck(const char* str, size_t maxsize) {
34 35 36 37 38 39 40 41 42 43 44 45
  if (str == NULL) {
    return false;
  }

  size_t len = strlen(str);
  if (len <= 0 || len > maxsize) {
    return false;
  }

  return true;
}

dengyihao's avatar
dengyihao 已提交
46
static bool validateUserName(const char* user) { return stringLengthCheck(user, TSDB_USER_LEN - 1); }
47

dengyihao's avatar
dengyihao 已提交
48
static bool validatePassword(const char* passwd) { return stringLengthCheck(passwd, TSDB_PASSWORD_LEN - 1); }
49

dengyihao's avatar
dengyihao 已提交
50
static bool validateDbName(const char* db) { return stringLengthCheck(db, TSDB_DB_NAME_LEN - 1); }
51

52 53 54 55 56 57
static char* getClusterKey(const char* user, const char* auth, const char* ip, int32_t port) {
  char key[512] = {0};
  snprintf(key, sizeof(key), "%s:%s:%s:%d", user, auth, ip, port);
  return strdup(key);
}

dengyihao's avatar
dengyihao 已提交
58
static STscObj* taosConnectImpl(const char* user, const char* auth, const char* db, __taos_async_fn_t fp, void* param,
L
Liu Jicong 已提交
59
                                SAppInstInfo* pAppInfo, int connType);
60

61
STscObj* taos_connect_internal(const char* ip, const char* user, const char* pass, const char* auth, const char* db,
dengyihao's avatar
dengyihao 已提交
62
                               uint16_t port, int connType) {
63 64 65 66
  if (taos_init() != TSDB_CODE_SUCCESS) {
    return NULL;
  }

67 68 69 70 71
  if (!validateUserName(user)) {
    terrno = TSDB_CODE_TSC_INVALID_USER_LENGTH;
    return NULL;
  }

72
  char localDb[TSDB_DB_NAME_LEN] = {0};
H
Haojun Liao 已提交
73
  if (db != NULL && strlen(db) > 0) {
dengyihao's avatar
dengyihao 已提交
74
    if (!validateDbName(db)) {
75 76 77 78
      terrno = TSDB_CODE_TSC_INVALID_DB_LENGTH;
      return NULL;
    }

79 80
    tstrncpy(localDb, db, sizeof(localDb));
    strdequote(localDb);
81 82
  }

83
  char secretEncrypt[TSDB_PASSWORD_LEN + 1] = {0};
84 85 86 87 88 89
  if (auth == NULL) {
    if (!validatePassword(pass)) {
      terrno = TSDB_CODE_TSC_INVALID_PASS_LENGTH;
      return NULL;
    }

dengyihao's avatar
dengyihao 已提交
90
    taosEncryptPass_c((uint8_t*)pass, strlen(pass), secretEncrypt);
91 92 93 94
  } else {
    tstrncpy(secretEncrypt, auth, tListLen(secretEncrypt));
  }

95
  SCorEpSet epSet = {0};
S
Shengliang Guan 已提交
96 97 98 99 100 101 102 103 104
  if (ip) {
    if (initEpSetFromCfg(ip, NULL, &epSet) < 0) {
      return NULL;
    }
  } else {
    if (initEpSetFromCfg(tsFirst, tsSecond, &epSet) < 0) {
      return NULL;
    }
  }
105

106 107
  if (port) {
    epSet.epSet.eps[0].port = port;
S
Shengliang Guan 已提交
108
    epSet.epSet.eps[1].port = port;
109 110
  }

111
  char* key = getClusterKey(user, secretEncrypt, ip, port);
112

113
  SAppInstInfo** pInst = NULL;
wafwerar's avatar
wafwerar 已提交
114
  taosThreadMutexLock(&appInfo.mutex);
H
Haojun Liao 已提交
115 116

  pInst = taosHashGet(appInfo.pInstMap, key, strlen(key));
L
Liu Jicong 已提交
117
  SAppInstInfo* p = NULL;
118
  if (pInst == NULL) {
wafwerar's avatar
wafwerar 已提交
119
    p = taosMemoryCalloc(1, sizeof(struct SAppInstInfo));
L
Liu Jicong 已提交
120
    p->mgmtEp = epSet;
D
dapan1121 已提交
121
    taosThreadMutexInit(&p->qnodeMutex, NULL);
H
Haojun Liao 已提交
122
    p->pTransporter = openTransporter(user, secretEncrypt, tsNumOfCores);
D
dapan 已提交
123
    p->pAppHbMgr = appHbMgrInit(p, key);
H
Haojun Liao 已提交
124
    taosHashPut(appInfo.pInstMap, key, strlen(key), &p, POINTER_BYTES);
D
dapan1121 已提交
125 126 127 128
    p->instKey = key;
    key = NULL;
    tscDebug("new app inst mgr %p, user:%s, ip:%s, port:%d", p, user, ip, port);
    
H
Haojun Liao 已提交
129
    pInst = &p;
130 131
  }

wafwerar's avatar
wafwerar 已提交
132
  taosThreadMutexUnlock(&appInfo.mutex);
H
Haojun Liao 已提交
133

wafwerar's avatar
wafwerar 已提交
134
  taosMemoryFreeClear(key);
L
Liu Jicong 已提交
135
  return taosConnectImpl(user, &secretEncrypt[0], localDb, NULL, NULL, *pInst, connType);
136 137
}

dengyihao's avatar
dengyihao 已提交
138
int32_t buildRequest(STscObj* pTscObj, const char* sql, int sqlLen, SRequestObj** pRequest) {
139
  *pRequest = createRequest(pTscObj, TSDB_SQL_SELECT);
X
Xiaoyu Wang 已提交
140
  if (*pRequest == NULL) {
141
    tscError("failed to malloc sqlObj");
X
Xiaoyu Wang 已提交
142
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
143 144
  }

wafwerar's avatar
wafwerar 已提交
145
  (*pRequest)->sqlstr = taosMemoryMalloc(sqlLen + 1);
X
Xiaoyu Wang 已提交
146
  if ((*pRequest)->sqlstr == NULL) {
dengyihao's avatar
dengyihao 已提交
147
    tscError("0x%" PRIx64 " failed to prepare sql string buffer", (*pRequest)->self);
X
Xiaoyu Wang 已提交
148 149
    (*pRequest)->msgBuf = strdup("failed to prepare sql string buffer");
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
150 151
  }

X
Xiaoyu Wang 已提交
152 153 154
  strntolower((*pRequest)->sqlstr, sql, (int32_t)sqlLen);
  (*pRequest)->sqlstr[sqlLen] = 0;
  (*pRequest)->sqlLen = sqlLen;
155

dengyihao's avatar
dengyihao 已提交
156 157
  if (taosHashPut(pTscObj->pRequests, &(*pRequest)->self, sizeof((*pRequest)->self), &(*pRequest)->self,
                  sizeof((*pRequest)->self))) {
D
dapan1121 已提交
158 159 160 161 162 163
    destroyRequest(*pRequest);
    *pRequest = NULL;
    tscError("put request to request hash failed");
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
  }

dengyihao's avatar
dengyihao 已提交
164
  tscDebugL("0x%" PRIx64 " SQL: %s, reqId:0x%" PRIx64, (*pRequest)->self, (*pRequest)->sqlstr, (*pRequest)->requestId);
X
Xiaoyu Wang 已提交
165 166
  return TSDB_CODE_SUCCESS;
}
167

D
stmt  
dapan1121 已提交
168
int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery, SStmtCallback* pStmtCb) {
H
Haojun Liao 已提交
169 170
  STscObj* pTscObj = pRequest->pTscObj;

171
  SParseContext cxt = {.requestId = pRequest->requestId,
D
dapan1121 已提交
172
                       .requestRid = pRequest->self,
173 174 175 176 177 178 179 180 181
                       .acctId = pTscObj->acctId,
                       .db = pRequest->pDb,
                       .topicQuery = topicQuery,
                       .pSql = pRequest->sqlstr,
                       .sqlLen = pRequest->sqlLen,
                       .pMsg = pRequest->msgBuf,
                       .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
                       .pTransporter = pTscObj->pAppInfo->pTransporter,
                       .pStmtCb = pStmtCb,
182
                       .pUser = pTscObj->user,
wmmhello's avatar
wmmhello 已提交
183
                       .schemalessType = pTscObj->schemalessType,
D
dapan1121 已提交
184 185 186
                       .isSuperUser = (0 == strcmp(pTscObj->user, TSDB_DEFAULT_USER)),
                       .svrVer = pTscObj->sVer,
                       .nodeOffline = (pTscObj->pAppInfo->onlineDnodes < pTscObj->pAppInfo->totalDnodes)};
H
Haojun Liao 已提交
187

H
Haojun Liao 已提交
188 189
  cxt.mgmtEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
  int32_t code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &cxt.pCatalog);
H
Haojun Liao 已提交
190 191 192 193
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

X
Xiaoyu Wang 已提交
194
  code = qParseSql(&cxt, pQuery);
X
Xiaoyu Wang 已提交
195 196 197
  if (TSDB_CODE_SUCCESS == code) {
    if ((*pQuery)->haveResultSet) {
      setResSchemaInfo(&pRequest->body.resInfo, (*pQuery)->pResSchema, (*pQuery)->numOfResCols);
198
      setResPrecision(&pRequest->body.resInfo, (*pQuery)->precision);
X
Xiaoyu Wang 已提交
199
    }
200
  }
201

202
  if (TSDB_CODE_SUCCESS == code || NEED_CLIENT_HANDLE_ERROR(code)) {
wafwerar's avatar
wafwerar 已提交
203 204
    TSWAP(pRequest->dbList, (*pQuery)->pDbList);
    TSWAP(pRequest->tableList, (*pQuery)->pTableList);
X
Xiaoyu Wang 已提交
205
  }
206

X
Xiaoyu Wang 已提交
207 208
  return code;
}
H
Haojun Liao 已提交
209

210 211
int32_t execLocalCmd(SRequestObj* pRequest, SQuery* pQuery) {
  SRetrieveTableRsp* pRsp = NULL;
L
Liu Jicong 已提交
212
  int32_t            code = qExecCommand(pQuery->pRoot, &pRsp);
213
  if (TSDB_CODE_SUCCESS == code && NULL != pRsp) {
214
    code = setQueryResultFromRsp(&pRequest->body.resInfo, pRsp, false, true);
215
  }
216

217 218 219
  return code;
}

X
Xiaoyu Wang 已提交
220
int32_t execDdlQuery(SRequestObj* pRequest, SQuery* pQuery) {
X
Xiaoyu Wang 已提交
221 222 223 224 225
  // drop table if exists not_exists_table
  if (NULL == pQuery->pCmdMsg) {
    return TSDB_CODE_SUCCESS;
  }

226 227 228
  SCmdMsgInfo* pMsgInfo = pQuery->pCmdMsg;
  pRequest->type = pMsgInfo->msgType;
  pRequest->body.requestMsg = (SDataBuf){.pData = pMsgInfo->pMsg, .len = pMsgInfo->msgLen, .handle = NULL};
L
Liu Jicong 已提交
229
  pMsgInfo->pMsg = NULL;  // pMsg transferred to SMsgSendInfo management
230 231 232

  STscObj*      pTscObj = pRequest->pTscObj;
  SMsgSendInfo* pSendMsg = buildMsgInfoImpl(pRequest);
233

234 235
  int64_t transporterId = 0;
  asyncSendMsgToServer(pTscObj->pAppInfo->pTransporter, &pMsgInfo->epSet, &transporterId, pSendMsg);
X
Xiaoyu Wang 已提交
236

237
  tsem_wait(&pRequest->body.rspSem);
X
Xiaoyu Wang 已提交
238 239
  return TSDB_CODE_SUCCESS;
}
X
Xiaoyu Wang 已提交
240

241
static SAppInstInfo* getAppInfo(SRequestObj* pRequest) { return pRequest->pTscObj->pAppInfo; }
242

243 244
void asyncExecLocalCmd(SRequestObj* pRequest, SQuery* pQuery) {
  SRetrieveTableRsp* pRsp = NULL;
D
dapan1121 已提交
245 246 247 248
  if (pRequest->validateOnly) {
    pRequest->body.queryFp(pRequest->body.param, pRequest, 0);
    return;
  }
249 250 251 252 253 254 255

  int32_t code = qExecCommand(pQuery->pRoot, &pRsp);
  if (TSDB_CODE_SUCCESS == code && NULL != pRsp) {
    code = setQueryResultFromRsp(&pRequest->body.resInfo, pRsp, false, false);
  }

  SReqResultInfo* pResultInfo = &pRequest->body.resInfo;
D
dapan1121 已提交
256
  pRequest->code = code;
257 258 259 260 261 262 263 264 265 266 267

  if (pRequest->code != TSDB_CODE_SUCCESS) {
    pResultInfo->numOfRows = 0;
    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 已提交
268
  pRequest->body.queryFp(pRequest->body.param, pRequest, code);
269
  //  pRequest->body.fetchFp(pRequest->body.param, pRequest, pResultInfo->numOfRows);
270 271
}

272
int32_t asyncExecDdlQuery(SRequestObj* pRequest, SQuery* pQuery) {
D
dapan1121 已提交
273 274 275 276 277
  if (pRequest->validateOnly) {
    pRequest->body.queryFp(pRequest->body.param, pRequest, 0);
    return TSDB_CODE_SUCCESS;
  }

278 279
  // drop table if exists not_exists_table
  if (NULL == pQuery->pCmdMsg) {
D
dapan1121 已提交
280
    pRequest->body.queryFp(pRequest->body.param, pRequest, 0);
281 282 283 284 285 286 287 288 289 290 291 292
    return TSDB_CODE_SUCCESS;
  }

  SCmdMsgInfo* pMsgInfo = pQuery->pCmdMsg;
  pRequest->type = pMsgInfo->msgType;
  pRequest->body.requestMsg = (SDataBuf){.pData = pMsgInfo->pMsg, .len = pMsgInfo->msgLen, .handle = NULL};
  pMsgInfo->pMsg = NULL;  // pMsg transferred to SMsgSendInfo management

  SAppInstInfo* pAppInfo = getAppInfo(pRequest);
  SMsgSendInfo* pSendMsg = buildMsgInfoImpl(pRequest);

  int64_t transporterId = 0;
D
dapan1121 已提交
293 294 295 296 297
  int32_t code = asyncSendMsgToServer(pAppInfo->pTransporter, &pMsgInfo->epSet, &transporterId, pSendMsg);
  if (code) {
    pRequest->body.queryFp(pRequest->body.param, pRequest, code);
  }
  return code;
298 299
}

D
dapan1121 已提交
300
int compareQueryNodeLoad(const void* elem1, const void* elem2) {
L
Liu Jicong 已提交
301 302
  SQueryNodeLoad* node1 = (SQueryNodeLoad*)elem1;
  SQueryNodeLoad* node2 = (SQueryNodeLoad*)elem2;
D
dapan1121 已提交
303 304 305 306

  if (node1->load < node2->load) {
    return -1;
  }
L
Liu Jicong 已提交
307

D
dapan1121 已提交
308 309 310
  return node1->load > node2->load;
}

L
Liu Jicong 已提交
311
int32_t updateQnodeList(SAppInstInfo* pInfo, SArray* pNodeList) {
D
dapan1121 已提交
312 313 314 315 316
  taosThreadMutexLock(&pInfo->qnodeMutex);
  if (pInfo->pQnodeList) {
    taosArrayDestroy(pInfo->pQnodeList);
    pInfo->pQnodeList = NULL;
  }
L
Liu Jicong 已提交
317

D
dapan1121 已提交
318 319 320 321 322 323 324 325 326
  if (pNodeList) {
    pInfo->pQnodeList = taosArrayDup(pNodeList);
    taosArraySort(pInfo->pQnodeList, compareQueryNodeLoad);
  }
  taosThreadMutexUnlock(&pInfo->qnodeMutex);

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
327 328 329 330 331 332
bool qnodeRequired(SRequestObj* pRequest) {
  if (QUERY_POLICY_VNODE == tsQueryPolicy) {
    return false;
  }

  SAppInstInfo* pInfo = pRequest->pTscObj->pAppInfo;
dengyihao's avatar
dengyihao 已提交
333 334
  bool          required = false;

D
dapan1121 已提交
335 336 337 338 339 340 341
  taosThreadMutexLock(&pInfo->qnodeMutex);
  required = (NULL == pInfo->pQnodeList);
  taosThreadMutexUnlock(&pInfo->qnodeMutex);

  return required;
}

D
dapan1121 已提交
342
int32_t getQnodeList(SRequestObj* pRequest, SArray** pNodeList) {
L
Liu Jicong 已提交
343 344 345
  SAppInstInfo* pInfo = pRequest->pTscObj->pAppInfo;
  int32_t       code = 0;

D
dapan1121 已提交
346 347 348 349 350 351 352
  taosThreadMutexLock(&pInfo->qnodeMutex);
  if (pInfo->pQnodeList) {
    *pNodeList = taosArrayDup(pInfo->pQnodeList);
  }
  taosThreadMutexUnlock(&pInfo->qnodeMutex);

  if (NULL == *pNodeList) {
L
Liu Jicong 已提交
353
    SCatalog* pCatalog = NULL;
D
dapan1121 已提交
354 355 356
    code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog);
    if (TSDB_CODE_SUCCESS == code) {
      *pNodeList = taosArrayInit(5, sizeof(SQueryNodeLoad));
X
Xiaoyu Wang 已提交
357
      SRequestConnInfo conn = {.pTrans = pRequest->pTscObj->pAppInfo->pTransporter,
D
dapan1121 已提交
358 359 360 361
                               .requestId = pRequest->requestId,
                               .requestObjRefId = pRequest->self,
                               .mgmtEps = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp)};
      code = catalogGetQnodeList(pCatalog, &conn, *pNodeList);
D
dapan1121 已提交
362
    }
L
Liu Jicong 已提交
363

D
dapan1121 已提交
364 365 366 367 368 369 370 371
    if (TSDB_CODE_SUCCESS == code && *pNodeList) {
      code = updateQnodeList(pInfo, *pNodeList);
    }
  }

  return code;
}

D
dapan1121 已提交
372
int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArray* pNodeList) {
373
  pRequest->type = pQuery->msgType;
374 375
  SAppInstInfo* pAppInfo = getAppInfo(pRequest);

L
Liu Jicong 已提交
376 377
  SPlanContext cxt = {.queryId = pRequest->requestId,
                      .acctId = pRequest->pTscObj->acctId,
378
                      .mgmtEpSet = getEpSet_s(&pAppInfo->mgmtEp),
L
Liu Jicong 已提交
379 380 381
                      .pAstRoot = pQuery->pRoot,
                      .showRewrite = pQuery->showRewrite,
                      .pMsg = pRequest->msgBuf,
X
Xiaoyu Wang 已提交
382 383
                      .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
                      .pUser = pRequest->pTscObj->user};
D
dapan1121 已提交
384

D
dapan1121 已提交
385
  return qCreateQueryPlan(&cxt, pPlan, pNodeList);
X
Xiaoyu Wang 已提交
386 387
}

388
void setResSchemaInfo(SReqResultInfo* pResInfo, const SSchema* pSchema, int32_t numOfCols) {
389
  ASSERT(pSchema != NULL && numOfCols > 0);
390

391
  pResInfo->numOfCols = numOfCols;
L
Liu Jicong 已提交
392 393 394 395 396 397
  if (pResInfo->fields != NULL) {
    taosMemoryFree(pResInfo->fields);
  }
  if (pResInfo->userFields != NULL) {
    taosMemoryFree(pResInfo->userFields);
  }
398 399
  pResInfo->fields = taosMemoryCalloc(numOfCols, sizeof(TAOS_FIELD));
  pResInfo->userFields = taosMemoryCalloc(numOfCols, sizeof(TAOS_FIELD));
400 401

  for (int32_t i = 0; i < pResInfo->numOfCols; ++i) {
402
    pResInfo->fields[i].bytes = pSchema[i].bytes;
L
Liu Jicong 已提交
403
    pResInfo->fields[i].type = pSchema[i].type;
404 405

    pResInfo->userFields[i].bytes = pSchema[i].bytes;
L
Liu Jicong 已提交
406
    pResInfo->userFields[i].type = pSchema[i].type;
407 408 409

    if (pSchema[i].type == TSDB_DATA_TYPE_VARCHAR) {
      pResInfo->userFields[i].bytes -= VARSTR_HEADER_SIZE;
wmmhello's avatar
wmmhello 已提交
410
    } else if (pSchema[i].type == TSDB_DATA_TYPE_NCHAR || pSchema[i].type == TSDB_DATA_TYPE_JSON) {
411 412 413
      pResInfo->userFields[i].bytes = (pResInfo->userFields[i].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;
    }

414
    tstrncpy(pResInfo->fields[i].name, pSchema[i].name, tListLen(pResInfo->fields[i].name));
415
    tstrncpy(pResInfo->userFields[i].name, pSchema[i].name, tListLen(pResInfo->userFields[i].name));
416
  }
X
Xiaoyu Wang 已提交
417 418
}

419
void setResPrecision(SReqResultInfo* pResInfo, int32_t precision) {
dengyihao's avatar
dengyihao 已提交
420 421
  if (precision != TSDB_TIME_PRECISION_MILLI && precision != TSDB_TIME_PRECISION_MICRO &&
      precision != TSDB_TIME_PRECISION_NANO) {
422 423 424 425 426 427
    return;
  }

  pResInfo->precision = precision;
}

D
dapan1121 已提交
428 429 430 431 432 433 434 435 436 437 438 439
int32_t buildVnodePolicyNodeList(SRequestObj* pRequest, SArray** pNodeList, SArray* pMnodeList, SArray* pDbVgList) {
  SArray* nodeList = taosArrayInit(4, sizeof(SQueryNodeLoad));

  int32_t dbNum = taosArrayGetSize(pDbVgList);
  for (int32_t i = 0; i < dbNum; ++i) {
    SArray* pVg = taosArrayGetP(pDbVgList, i);
    int32_t vgNum = taosArrayGetSize(pVg);
    if (vgNum <= 0) {
      continue;
    }

    for (int32_t j = 0; j < vgNum; ++j) {
dengyihao's avatar
dengyihao 已提交
440
      SVgroupInfo*   pInfo = taosArrayGet(pVg, j);
D
dapan1121 已提交
441 442 443
      SQueryNodeLoad load = {0};
      load.addr.nodeId = pInfo->vgId;
      load.addr.epSet = pInfo->epSet;
dengyihao's avatar
dengyihao 已提交
444

D
dapan1121 已提交
445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501
      taosArrayPush(nodeList, &load);
    }
  }

  int32_t vnodeNum = taosArrayGetSize(nodeList);
  if (vnodeNum > 0) {
    tscDebug("0x%" PRIx64 " vnode policy, use vnode list, num:%d", pRequest->requestId, vnodeNum);
    goto _return;
  }

  int32_t mnodeNum = taosArrayGetSize(pMnodeList);
  if (mnodeNum <= 0) {
    tscDebug("0x%" PRIx64 " vnode policy, empty node list", pRequest->requestId);
    goto _return;
  }

  void* pData = taosArrayGet(pMnodeList, 0);
  taosArrayAddBatch(nodeList, pData, mnodeNum);

  tscDebug("0x%" PRIx64 " vnode policy, use mnode list, num:%d", pRequest->requestId, mnodeNum);

_return:

  *pNodeList = nodeList;

  return TSDB_CODE_SUCCESS;
}

int32_t buildQnodePolicyNodeList(SRequestObj* pRequest, SArray** pNodeList, SArray* pMnodeList, SArray* pQnodeList) {
  SArray* nodeList = taosArrayInit(4, sizeof(SQueryNodeLoad));

  int32_t qNodeNum = taosArrayGetSize(pQnodeList);
  if (qNodeNum > 0) {
    void* pData = taosArrayGet(pQnodeList, 0);
    taosArrayAddBatch(nodeList, pData, qNodeNum);
    tscDebug("0x%" PRIx64 " qnode policy, use qnode list, num:%d", pRequest->requestId, qNodeNum);
    goto _return;
  }

  int32_t mnodeNum = taosArrayGetSize(pMnodeList);
  if (mnodeNum <= 0) {
    tscDebug("0x%" PRIx64 " qnode policy, empty node list", pRequest->requestId);
    goto _return;
  }

  void* pData = taosArrayGet(pMnodeList, 0);
  taosArrayAddBatch(nodeList, pData, mnodeNum);

  tscDebug("0x%" PRIx64 " qnode policy, use mnode list, num:%d", pRequest->requestId, mnodeNum);

_return:

  *pNodeList = nodeList;

  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
502
int32_t buildAsyncExecNodeList(SRequestObj* pRequest, SArray** pNodeList, SArray* pMnodeList, SMetaData* pResultMeta) {
D
dapan1121 已提交
503 504 505
  SArray* pDbVgList = NULL;
  SArray* pQnodeList = NULL;
  int32_t code = 0;
dengyihao's avatar
dengyihao 已提交
506

D
dapan1121 已提交
507 508 509 510
  switch (tsQueryPolicy) {
    case QUERY_POLICY_VNODE: {
      if (pResultMeta) {
        pDbVgList = taosArrayInit(4, POINTER_BYTES);
dengyihao's avatar
dengyihao 已提交
511

D
dapan1121 已提交
512 513 514 515 516 517 518 519
        int32_t dbNum = taosArrayGetSize(pResultMeta->pDbVgroup);
        for (int32_t i = 0; i < dbNum; ++i) {
          SMetaRes* pRes = taosArrayGet(pResultMeta->pDbVgroup, i);
          if (pRes->code || NULL == pRes->pRes) {
            continue;
          }

          taosArrayPush(pDbVgList, &pRes->pRes);
dengyihao's avatar
dengyihao 已提交
520
        }
D
dapan1121 已提交
521
      }
dengyihao's avatar
dengyihao 已提交
522

D
dapan1121 已提交
523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542
      code = buildVnodePolicyNodeList(pRequest, pNodeList, pMnodeList, pDbVgList);
      break;
    }
    case QUERY_POLICY_HYBRID:
    case QUERY_POLICY_QNODE: {
      if (pResultMeta && taosArrayGetSize(pResultMeta->pQnodeList) > 0) {
        SMetaRes* pRes = taosArrayGet(pResultMeta->pQnodeList, 0);
        if (pRes->code) {
          pQnodeList = NULL;
        } else {
          pQnodeList = taosArrayDup((SArray*)pRes->pRes);
        }
      } else {
        SAppInstInfo* pInst = pRequest->pTscObj->pAppInfo;
        taosThreadMutexLock(&pInst->qnodeMutex);
        if (pInst->pQnodeList) {
          pQnodeList = taosArrayDup(pInst->pQnodeList);
        }
        taosThreadMutexUnlock(&pInst->qnodeMutex);
      }
dengyihao's avatar
dengyihao 已提交
543

D
dapan1121 已提交
544 545 546 547 548 549 550 551 552 553
      code = buildQnodePolicyNodeList(pRequest, pNodeList, pMnodeList, pQnodeList);
      break;
    }
    default:
      tscError("unknown query policy: %d", tsQueryPolicy);
      return TSDB_CODE_TSC_APP_ERROR;
  }

  taosArrayDestroy(pDbVgList);
  taosArrayDestroy(pQnodeList);
dengyihao's avatar
dengyihao 已提交
554

D
dapan1121 已提交
555 556 557 558 559 560 561
  return code;
}

int32_t buildSyncExecNodeList(SRequestObj* pRequest, SArray** pNodeList, SArray* pMnodeList) {
  SArray* pDbVgList = NULL;
  SArray* pQnodeList = NULL;
  int32_t code = 0;
dengyihao's avatar
dengyihao 已提交
562

D
dapan1121 已提交
563 564 565 566
  switch (tsQueryPolicy) {
    case QUERY_POLICY_VNODE: {
      int32_t dbNum = taosArrayGetSize(pRequest->dbList);
      if (dbNum > 0) {
dengyihao's avatar
dengyihao 已提交
567
        SCatalog*     pCtg = NULL;
D
dapan1121 已提交
568 569 570 571 572 573
        SAppInstInfo* pInst = pRequest->pTscObj->pAppInfo;
        code = catalogGetHandle(pInst->clusterId, &pCtg);
        if (code != TSDB_CODE_SUCCESS) {
          goto _return;
        }

dengyihao's avatar
dengyihao 已提交
574
        pDbVgList = taosArrayInit(dbNum, POINTER_BYTES);
D
dapan1121 已提交
575 576
        SArray* pVgList = NULL;
        for (int32_t i = 0; i < dbNum; ++i) {
dengyihao's avatar
dengyihao 已提交
577
          char*            dbFName = taosArrayGet(pRequest->dbList, i);
D
dapan1121 已提交
578 579 580
          SRequestConnInfo conn = {.pTrans = pInst->pTransporter,
                                   .requestId = pRequest->requestId,
                                   .requestObjRefId = pRequest->self,
dengyihao's avatar
dengyihao 已提交
581 582
                                   .mgmtEps = getEpSet_s(&pInst->mgmtEp)};

D
dapan1121 已提交
583 584 585 586
          code = catalogGetDBVgInfo(pCtg, &conn, dbFName, &pVgList);
          if (code) {
            goto _return;
          }
dengyihao's avatar
dengyihao 已提交
587

D
dapan1121 已提交
588
          taosArrayPush(pDbVgList, &pVgList);
dengyihao's avatar
dengyihao 已提交
589
        }
D
dapan1121 已提交
590
      }
dengyihao's avatar
dengyihao 已提交
591

D
dapan1121 已提交
592 593 594 595 596 597
      code = buildVnodePolicyNodeList(pRequest, pNodeList, pMnodeList, pDbVgList);
      break;
    }
    case QUERY_POLICY_HYBRID:
    case QUERY_POLICY_QNODE: {
      getQnodeList(pRequest, &pQnodeList);
dengyihao's avatar
dengyihao 已提交
598

D
dapan1121 已提交
599 600 601 602 603 604 605 606 607 608 609 610
      code = buildQnodePolicyNodeList(pRequest, pNodeList, pMnodeList, pQnodeList);
      break;
    }
    default:
      tscError("unknown query policy: %d", tsQueryPolicy);
      return TSDB_CODE_TSC_APP_ERROR;
  }

_return:

  taosArrayDestroy(pDbVgList);
  taosArrayDestroy(pQnodeList);
dengyihao's avatar
dengyihao 已提交
611

D
dapan1121 已提交
612 613 614
  return code;
}

D
dapan1121 已提交
615
int32_t scheduleAsyncQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList) {
D
dapan1121 已提交
616 617
  tsem_init(&schdRspSem, 0, 0);

X
Xiaoyu Wang 已提交
618 619
  SQueryResult     res = {.code = 0, .numOfRows = 0};
  SRequestConnInfo conn = {.pTrans = pRequest->pTscObj->pAppInfo->pTransporter,
D
dapan1121 已提交
620
                           .requestId = pRequest->requestId,
X
Xiaoyu Wang 已提交
621 622 623 624 625 626 627 628 629 630
                           .requestObjRefId = pRequest->self};
  SSchedulerReq    req = {.pConn = &conn,
                          .pNodeList = pNodeList,
                          .pDag = pDag,
                          .sql = pRequest->sqlstr,
                          .startTs = pRequest->metric.start,
                          .fp = schdExecCallback,
                          .cbParam = &res};

  int32_t code = schedulerAsyncExecJob(&req, &pRequest->body.queryJob);
D
dapan1121 已提交
631 632 633

  pRequest->body.resInfo.execRes = res.res;

634
  while (true) {
D
dapan1121 已提交
635 636
    if (code != TSDB_CODE_SUCCESS) {
      if (pRequest->body.queryJob != 0) {
D
dapan1121 已提交
637
        schedulerFreeJob(pRequest->body.queryJob, 0);
D
dapan1121 已提交
638 639 640 641 642 643 644
      }

      pRequest->code = code;
      terrno = code;
      return pRequest->code;
    } else {
      tsem_wait(&schdRspSem);
L
Liu Jicong 已提交
645

D
dapan1121 已提交
646 647 648 649 650 651 652 653 654 655 656 657
      if (res.code) {
        code = res.code;
      } else {
        break;
      }
    }
  }

  if (TDMT_VND_SUBMIT == pRequest->type || TDMT_VND_CREATE_TABLE == pRequest->type) {
    pRequest->body.resInfo.numOfRows = res.numOfRows;

    if (pRequest->body.queryJob != 0) {
D
dapan1121 已提交
658
      schedulerFreeJob(pRequest->body.queryJob, 0);
D
dapan1121 已提交
659 660 661 662 663 664 665 666
    }
  }

  pRequest->code = res.code;
  terrno = res.code;
  return pRequest->code;
}

D
dapan1121 已提交
667
int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList) {
668
  void* pTransporter = pRequest->pTscObj->pAppInfo->pTransporter;
669

X
Xiaoyu Wang 已提交
670 671
  SQueryResult     res = {0};
  SRequestConnInfo conn = {.pTrans = pRequest->pTscObj->pAppInfo->pTransporter,
D
dapan1121 已提交
672
                           .requestId = pRequest->requestId,
X
Xiaoyu Wang 已提交
673 674 675 676 677 678 679
                           .requestObjRefId = pRequest->self};
  SSchedulerReq    req = {.pConn = &conn,
                          .pNodeList = pNodeList,
                          .pDag = pDag,
                          .sql = pRequest->sqlstr,
                          .startTs = pRequest->metric.start,
                          .fp = NULL,
D
dapan1121 已提交
680 681
                          .cbParam = NULL,
                          .reqKilled = &pRequest->killed};
X
Xiaoyu Wang 已提交
682 683

  int32_t code = schedulerExecJob(&req, &pRequest->body.queryJob, &res);
D
dapan1121 已提交
684
  pRequest->body.resInfo.execRes = res.res;
D
dapan1121 已提交
685

D
dapan1121 已提交
686
  if (code != TSDB_CODE_SUCCESS) {
D
dapan1121 已提交
687
    if (pRequest->body.queryJob != 0) {
D
dapan1121 已提交
688
      schedulerFreeJob(pRequest->body.queryJob, 0);
689 690
    }

D
dapan1121 已提交
691
    pRequest->code = code;
D
dapan1121 已提交
692
    terrno = code;
H
Haojun Liao 已提交
693
    return pRequest->code;
X
Xiaoyu Wang 已提交
694
  }
695

696 697
  if (TDMT_VND_SUBMIT == pRequest->type || TDMT_VND_DELETE == pRequest->type ||
      TDMT_VND_CREATE_TABLE == pRequest->type) {
D
dapan1121 已提交
698
    pRequest->body.resInfo.numOfRows = res.numOfRows;
dengyihao's avatar
test  
dengyihao 已提交
699

D
dapan1121 已提交
700
    if (pRequest->body.queryJob != 0) {
D
dapan1121 已提交
701
      schedulerFreeJob(pRequest->body.queryJob, 0);
D
dapan1121 已提交
702 703
    }
  }
D
dapan1121 已提交
704

D
dapan1121 已提交
705
  pRequest->code = res.code;
L
Liu Jicong 已提交
706
  terrno = res.code;
D
dapan1121 已提交
707
  return pRequest->code;
X
Xiaoyu Wang 已提交
708
}
X
Xiaoyu Wang 已提交
709

710 711 712
int32_t handleSubmitExecRes(SRequestObj* pRequest, void* res, SCatalog* pCatalog, SEpSet* epset) {
  int32_t     code = 0;
  SArray*     pArray = NULL;
D
dapan1121 已提交
713 714 715 716
  SSubmitRsp* pRsp = (SSubmitRsp*)res;
  if (pRsp->nBlocks <= 0) {
    return TSDB_CODE_SUCCESS;
  }
dengyihao's avatar
dengyihao 已提交
717

D
dapan1121 已提交
718 719 720 721 722
  pArray = taosArrayInit(pRsp->nBlocks, sizeof(STbSVersion));
  if (NULL == pArray) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return TSDB_CODE_OUT_OF_MEMORY;
  }
D
dapan1121 已提交
723

D
dapan1121 已提交
724 725 726 727
  for (int32_t i = 0; i < pRsp->nBlocks; ++i) {
    SSubmitBlkRsp* blk = pRsp->pBlocks + i;
    if (NULL == blk->tblFName || 0 == blk->tblFName[0]) {
      continue;
D
dapan1121 已提交
728
    }
dengyihao's avatar
dengyihao 已提交
729

D
dapan1121 已提交
730 731 732
    STbSVersion tbSver = {.tbFName = blk->tblFName, .sver = blk->sver};
    taosArrayPush(pArray, &tbSver);
  }
dengyihao's avatar
dengyihao 已提交
733

X
Xiaoyu Wang 已提交
734
  SRequestConnInfo conn = {.pTrans = pRequest->pTscObj->pAppInfo->pTransporter,
D
dapan1121 已提交
735 736 737 738 739
                           .requestId = pRequest->requestId,
                           .requestObjRefId = pRequest->self,
                           .mgmtEps = *epset};

  code = catalogChkTbMetaVersion(pCatalog, &conn, pArray);
740

D
dapan1121 已提交
741
_return:
742

D
dapan1121 已提交
743
  taosArrayDestroy(pArray);
744
  return code;
D
dapan1121 已提交
745
}
D
dapan1121 已提交
746

747
int32_t handleQueryExecRes(SRequestObj* pRequest, void* res, SCatalog* pCatalog, SEpSet* epset) {
D
dapan1121 已提交
748 749 750 751 752 753 754
  int32_t code = 0;
  SArray* pArray = NULL;
  SArray* pTbArray = (SArray*)res;
  int32_t tbNum = taosArrayGetSize(pTbArray);
  if (tbNum <= 0) {
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
755

D
dapan1121 已提交
756 757 758 759 760
  pArray = taosArrayInit(tbNum, sizeof(STbSVersion));
  if (NULL == pArray) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return TSDB_CODE_OUT_OF_MEMORY;
  }
D
dapan1121 已提交
761

D
dapan1121 已提交
762 763 764 765
  for (int32_t i = 0; i < tbNum; ++i) {
    STbVerInfo* tbInfo = taosArrayGet(pTbArray, i);
    STbSVersion tbSver = {.tbFName = tbInfo->tbFName, .sver = tbInfo->sversion, .tver = tbInfo->tversion};
    taosArrayPush(pArray, &tbSver);
D
dapan1121 已提交
766 767
  }

X
Xiaoyu Wang 已提交
768
  SRequestConnInfo conn = {.pTrans = pRequest->pTscObj->pAppInfo->pTransporter,
D
dapan1121 已提交
769 770 771 772 773
                           .requestId = pRequest->requestId,
                           .requestObjRefId = pRequest->self,
                           .mgmtEps = *epset};

  code = catalogChkTbMetaVersion(pCatalog, &conn, pArray);
D
dapan1121 已提交
774 775 776 777 778

_return:

  taosArrayDestroy(pArray);
  return code;
D
dapan1121 已提交
779
}
D
dapan1121 已提交
780

D
dapan1121 已提交
781 782
int32_t handleAlterTbExecRes(void* res, SCatalog* pCatalog) {
  return catalogUpdateTableMeta(pCatalog, (STableMetaRsp*)res);
D
dapan1121 已提交
783 784
}

H
Haojun Liao 已提交
785
int32_t handleQueryExecRsp(SRequestObj* pRequest) {
D
dapan1121 已提交
786 787 788
  if (NULL == pRequest->body.resInfo.execRes.res) {
    return TSDB_CODE_SUCCESS;
  }
dengyihao's avatar
dengyihao 已提交
789

790 791
  SCatalog*     pCatalog = NULL;
  SAppInstInfo* pAppInfo = getAppInfo(pRequest);
H
Haojun Liao 已提交
792 793

  int32_t code = catalogGetHandle(pAppInfo->clusterId, &pCatalog);
D
dapan1121 已提交
794 795
  if (code) {
    return code;
D
dapan1121 已提交
796
  }
797

798
  SEpSet         epset = getEpSet_s(&pAppInfo->mgmtEp);
D
dapan1121 已提交
799
  SQueryExecRes* pRes = &pRequest->body.resInfo.execRes;
800

D
dapan1121 已提交
801
  switch (pRes->msgType) {
802
    case TDMT_VND_ALTER_TABLE:
D
dapan1121 已提交
803
    case TDMT_MND_ALTER_STB: {
D
dapan1121 已提交
804
      code = handleAlterTbExecRes(pRes->res, pCatalog);
D
dapan1121 已提交
805 806 807
      break;
    }
    case TDMT_VND_SUBMIT: {
D
dapan1121 已提交
808
      code = handleSubmitExecRes(pRequest, pRes->res, pCatalog, &epset);
D
dapan1121 已提交
809
      break;
810
    }
D
dapan1121 已提交
811
    case TDMT_VND_QUERY: {
D
dapan1121 已提交
812
      code = handleQueryExecRes(pRequest, pRes->res, pCatalog, &epset);
D
dapan1121 已提交
813 814 815
      break;
    }
    default:
816 817
      tscError("0x%" PRIx64 ", invalid exec result for request type %d, reqId:0x%" PRIx64, pRequest->self,
               pRequest->type, pRequest->requestId);
H
Haojun Liao 已提交
818
      code = TSDB_CODE_APP_ERROR;
D
dapan1121 已提交
819
  }
D
dapan1121 已提交
820 821

  return code;
D
dapan1121 已提交
822
}
D
dapan1121 已提交
823

824
void schedulerExecCb(SQueryResult* pResult, void* param, int32_t code) {
825
  SRequestObj* pRequest = (SRequestObj*)param;
826 827
  pRequest->code = code;

D
dapan1121 已提交
828 829 830 831 832 833 834 835 836 837
  if (TDMT_VND_SUBMIT == pRequest->type || TDMT_VND_DELETE == pRequest->type ||
      TDMT_VND_CREATE_TABLE == pRequest->type) {
    pRequest->body.resInfo.numOfRows = pResult->numOfRows;

    if (pRequest->body.queryJob != 0) {
      schedulerFreeJob(pRequest->body.queryJob, 0);
      pRequest->body.queryJob = 0;
    }
  }

dengyihao's avatar
dengyihao 已提交
838 839
  tscDebug("0x%" PRIx64 " enter scheduler exec cb, code:%d - %s, reqId:0x%" PRIx64, pRequest->self, code,
           tstrerror(code), pRequest->requestId);
D
dapan1121 已提交
840

841 842
  STscObj* pTscObj = pRequest->pTscObj;
  if (code != TSDB_CODE_SUCCESS && NEED_CLIENT_HANDLE_ERROR(code)) {
843 844
    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);
845 846 847 848 849
    pRequest->prevCode = code;
    doAsyncQuery(pRequest, true);
    return;
  }

H
Haojun Liao 已提交
850 851
  if (code == TSDB_CODE_SUCCESS) {
    code = handleQueryExecRsp(pRequest);
852
    ASSERT(pRequest->code == TSDB_CODE_SUCCESS);
H
Haojun Liao 已提交
853 854 855
    pRequest->code = code;
  }

D
dapan1121 已提交
856
  tscDebug("schedulerExecCb request type %s", TMSG_INFO(pRequest->type));
857 858 859
  if (NEED_CLIENT_RM_TBLMETA_REQ(pRequest->type)) {
    removeMeta(pTscObj, pRequest->tableList);
  }
860 861 862 863 864

  // return to client
  pRequest->body.queryFp(pRequest->body.param, pRequest, code);
}

H
Haojun Liao 已提交
865 866 867 868 869
SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, bool keepQuery, void** res) {
  int32_t code = 0;

  switch (pQuery->execMode) {
    case QUERY_EXEC_MODE_LOCAL:
D
dapan1121 已提交
870 871 872
      if (!pRequest->validateOnly) {
        code = execLocalCmd(pRequest, pQuery);
      }
H
Haojun Liao 已提交
873 874
      break;
    case QUERY_EXEC_MODE_RPC:
D
dapan1121 已提交
875 876 877
      if (!pRequest->validateOnly) {
        code = execDdlQuery(pRequest, pQuery);
      }
H
Haojun Liao 已提交
878 879
      break;
    case QUERY_EXEC_MODE_SCHEDULE: {
D
dapan1121 已提交
880
      SArray* pMnodeList = taosArrayInit(4, sizeof(SQueryNodeLoad));
D
dapan1121 已提交
881 882 883 884 885 886 887 888 889 890 891
      SQueryPlan* pDag = NULL;
      code = getPlan(pRequest, pQuery, &pDag, pMnodeList);
      if (TSDB_CODE_SUCCESS == code) {        
        pRequest->body.subplanNum = pDag->numOfSubplans;
        if (!pRequest->validateOnly) {
          SArray* pNodeList = NULL;
          buildSyncExecNodeList(pRequest, &pNodeList, pMnodeList);
          
          code = scheduleQuery(pRequest, pDag, pNodeList);
          taosArrayDestroy(pNodeList);
        }
X
Xiaoyu Wang 已提交
892
      }
D
dapan1121 已提交
893
      taosArrayDestroy(pMnodeList);
H
Haojun Liao 已提交
894
      break;
X
Xiaoyu Wang 已提交
895
    }
H
Haojun Liao 已提交
896 897 898 899 900
    case QUERY_EXEC_MODE_EMPTY_RESULT:
      pRequest->type = TSDB_SQL_RETRIEVE_EMPTY_RESULT;
      break;
    default:
      break;
X
Xiaoyu Wang 已提交
901 902
  }

D
stmt  
dapan1121 已提交
903 904 905
  if (!keepQuery) {
    qDestroyQuery(pQuery);
  }
dengyihao's avatar
dengyihao 已提交
906

H
Haojun Liao 已提交
907
  handleQueryExecRsp(pRequest);
D
dapan1121 已提交
908

D
dapan1121 已提交
909
  if (NULL != pRequest && TSDB_CODE_SUCCESS != code) {
X
Xiaoyu Wang 已提交
910
    pRequest->code = terrno;
D
dapan1121 已提交
911 912 913
  }

  if (res) {
D
dapan1121 已提交
914 915
    *res = pRequest->body.resInfo.execRes.res;
    pRequest->body.resInfo.execRes.res = NULL;
X
Xiaoyu Wang 已提交
916
  }
917

918 919 920
  return pRequest;
}

D
dapan1121 已提交
921
SRequestObj* launchQuery(STscObj* pTscObj, const char* sql, int sqlLen, bool validateOnly) {
D
stmt  
dapan1121 已提交
922 923 924 925
  SRequestObj* pRequest = NULL;
  SQuery*      pQuery = NULL;

  int32_t code = buildRequest(pTscObj, sql, sqlLen, &pRequest);
926 927 928 929 930
  if (code != TSDB_CODE_SUCCESS) {
    terrno = code;
    return NULL;
  }

D
dapan1121 已提交
931 932
  pRequest->validateOnly = validateOnly;

933 934
  code = parseSql(pRequest, false, &pQuery, NULL);
  if (code != TSDB_CODE_SUCCESS) {
935 936
    pRequest->code = code;
    return pRequest;
D
stmt  
dapan1121 已提交
937 938
  }

939 940
  pRequest->stableQuery = pQuery->stableQuery;

941 942 943
  return launchQueryImpl(pRequest, pQuery, false, NULL);
}

dengyihao's avatar
dengyihao 已提交
944
void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultMeta) {
945
  int32_t code = 0;
946

947 948
  switch (pQuery->execMode) {
    case QUERY_EXEC_MODE_LOCAL:
949 950
      asyncExecLocalCmd(pRequest, pQuery);
      return;
951 952 953 954
    case QUERY_EXEC_MODE_RPC:
      code = asyncExecDdlQuery(pRequest, pQuery);
      break;
    case QUERY_EXEC_MODE_SCHEDULE: {
D
dapan1121 已提交
955
      SArray* pMnodeList = taosArrayInit(4, sizeof(SQueryNodeLoad));
956 957 958 959 960 961 962 963 964

      pRequest->type = pQuery->msgType;

      SPlanContext cxt = {.queryId = pRequest->requestId,
                          .acctId = pRequest->pTscObj->acctId,
                          .mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp),
                          .pAstRoot = pQuery->pRoot,
                          .showRewrite = pQuery->showRewrite,
                          .pMsg = pRequest->msgBuf,
X
Xiaoyu Wang 已提交
965 966
                          .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
                          .pUser = pRequest->pTscObj->user};
967 968

      SAppInstInfo* pAppInfo = getAppInfo(pRequest);
D
dapan1121 已提交
969 970
      SQueryPlan* pDag = NULL;
      code = qCreateQueryPlan(&cxt, &pDag, pMnodeList);
D
dapan1121 已提交
971 972 973
      if (code) {
        tscError("0x%" PRIx64 " failed to create query plan, code:%s 0x%" PRIx64, pRequest->self, tstrerror(code),
                 pRequest->requestId);
D
dapan1121 已提交
974 975
      } else {
        pRequest->body.subplanNum = pDag->numOfSubplans;
976 977
      }

D
dapan1121 已提交
978
      if (TSDB_CODE_SUCCESS == code && !pRequest->validateOnly) {
D
dapan1121 已提交
979 980
        SArray* pNodeList = NULL;
        buildAsyncExecNodeList(pRequest, &pNodeList, pMnodeList, pResultMeta);
dengyihao's avatar
dengyihao 已提交
981

X
Xiaoyu Wang 已提交
982 983
        SRequestConnInfo conn = {
            .pTrans = pAppInfo->pTransporter, .requestId = pRequest->requestId, .requestObjRefId = pRequest->self};
D
dapan1121 已提交
984 985
        SSchedulerReq req = {.pConn = &conn,
                             .pNodeList = pNodeList,
D
dapan1121 已提交
986
                             .pDag = pDag,
D
dapan1121 已提交
987 988 989
                             .sql = pRequest->sqlstr,
                             .startTs = pRequest->metric.start,
                             .fp = schedulerExecCb,
D
dapan1121 已提交
990 991
                             .cbParam = pRequest,
                             .reqKilled = &pRequest->killed};
D
dapan1121 已提交
992
        code = schedulerAsyncExecJob(&req, &pRequest->body.queryJob);
D
dapan1121 已提交
993
        taosArrayDestroy(pNodeList);
D
dapan1121 已提交
994
      } else {
D
dapan1121 已提交
995
        tscDebug("0x%" PRIx64 " plan not executed, code:%s 0x%" PRIx64, pRequest->self, tstrerror(code),
996
                 pRequest->requestId);
D
dapan1121 已提交
997
        pRequest->body.queryFp(pRequest->body.param, pRequest, code);
998 999
      }

1000
      // todo not to be released here
D
dapan1121 已提交
1001
      taosArrayDestroy(pMnodeList);
1002 1003 1004
      break;
    }
    case QUERY_EXEC_MODE_EMPTY_RESULT:
1005
      pRequest->type = TSDB_SQL_RETRIEVE_EMPTY_RESULT;
D
fix bug  
dapan1121 已提交
1006
      pRequest->body.queryFp(pRequest->body.param, pRequest, 0);
1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018
      break;
    default:
      break;
  }

  //    if (!keepQuery) {
  //      qDestroyQuery(pQuery);
  //    }

  if (NULL != pRequest && TSDB_CODE_SUCCESS != code) {
    pRequest->code = terrno;
  }
D
stmt  
dapan1121 已提交
1019 1020
}

D
dapan1121 已提交
1021
int32_t refreshMeta(STscObj* pTscObj, SRequestObj* pRequest) {
L
Liu Jicong 已提交
1022 1023 1024 1025
  SCatalog* pCatalog = NULL;
  int32_t   code = 0;
  int32_t   dbNum = taosArrayGetSize(pRequest->dbList);
  int32_t   tblNum = taosArrayGetSize(pRequest->tableList);
D
dapan1121 已提交
1026 1027 1028 1029

  if (dbNum <= 0 && tblNum <= 0) {
    return TSDB_CODE_QRY_APP_ERROR;
  }
L
Liu Jicong 已提交
1030

D
dapan1121 已提交
1031 1032 1033 1034 1035
  code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCatalog);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

X
Xiaoyu Wang 已提交
1036
  SRequestConnInfo conn = {.pTrans = pTscObj->pAppInfo->pTransporter,
D
dapan1121 已提交
1037 1038 1039
                           .requestId = pRequest->requestId,
                           .requestObjRefId = pRequest->self,
                           .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)};
D
dapan1121 已提交
1040 1041

  for (int32_t i = 0; i < dbNum; ++i) {
L
Liu Jicong 已提交
1042 1043
    char* dbFName = taosArrayGet(pRequest->dbList, i);

D
dapan1121 已提交
1044
    code = catalogRefreshDBVgInfo(pCatalog, &conn, dbFName);
D
dapan1121 已提交
1045 1046
    if (code != TSDB_CODE_SUCCESS) {
      return code;
D
dapan1121 已提交
1047 1048 1049
    }
  }

D
dapan1121 已提交
1050
  for (int32_t i = 0; i < tblNum; ++i) {
L
Liu Jicong 已提交
1051
    SName* tableName = taosArrayGet(pRequest->tableList, i);
D
dapan1121 已提交
1052

D
dapan1121 已提交
1053
    code = catalogRefreshTableMeta(pCatalog, &conn, tableName, -1);
D
dapan1121 已提交
1054 1055 1056
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }
D
dapan1121 已提交
1057 1058
  }

D
dapan1121 已提交
1059
  return code;
D
dapan1121 已提交
1060 1061
}

D
dapan1121 已提交
1062 1063
int32_t removeMeta(STscObj* pTscObj, SArray* tbList) {
  SCatalog* pCatalog = NULL;
1064 1065
  int32_t   tbNum = taosArrayGetSize(tbList);
  int32_t   code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCatalog);
D
dapan1121 已提交
1066 1067 1068
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }
1069

D
dapan1121 已提交
1070 1071 1072 1073 1074 1075 1076 1077
  for (int32_t i = 0; i < tbNum; ++i) {
    SName* pTbName = taosArrayGet(tbList, i);
    catalogRemoveTableMeta(pCatalog, pTbName);
  }

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
1078
SRequestObj* execQuery(STscObj* pTscObj, const char* sql, int sqlLen, bool validateOnly) {
D
dapan1121 已提交
1079
  SRequestObj* pRequest = NULL;
L
Liu Jicong 已提交
1080 1081
  int32_t      retryNum = 0;
  int32_t      code = 0;
D
dapan1121 已提交
1082

1083 1084
  do {
    destroyRequest(pRequest);
D
dapan1121 已提交
1085
    pRequest = launchQuery(pTscObj, sql, sqlLen, validateOnly);
1086
    if (pRequest == NULL || TSDB_CODE_SUCCESS == pRequest->code || !NEED_CLIENT_HANDLE_ERROR(pRequest->code)) {
D
dapan1121 已提交
1087 1088 1089
      break;
    }

D
dapan1121 已提交
1090 1091 1092
    code = refreshMeta(pTscObj, pRequest);
    if (code) {
      pRequest->code = code;
D
dapan1121 已提交
1093 1094
      break;
    }
1095
  } while (retryNum++ < REQUEST_TOTAL_EXEC_TIMES);
L
Liu Jicong 已提交
1096

D
dapan1121 已提交
1097 1098 1099
  if (NEED_CLIENT_RM_TBLMETA_REQ(pRequest->type)) {
    removeMeta(pTscObj, pRequest->tableList);
  }
1100

D
dapan1121 已提交
1101 1102 1103
  return pRequest;
}

S
Shengliang Guan 已提交
1104 1105
int initEpSetFromCfg(const char* firstEp, const char* secondEp, SCorEpSet* pEpSet) {
  pEpSet->version = 0;
1106

H
Haojun Liao 已提交
1107
  // init mnode ip set
dengyihao's avatar
dengyihao 已提交
1108
  SEpSet* mgmtEpSet = &(pEpSet->epSet);
1109
  mgmtEpSet->numOfEps = 0;
dengyihao's avatar
dengyihao 已提交
1110
  mgmtEpSet->inUse = 0;
1111

S
Shengliang Guan 已提交
1112 1113 1114 1115
  if (firstEp && firstEp[0] != 0) {
    if (strlen(firstEp) >= TSDB_EP_LEN) {
      terrno = TSDB_CODE_TSC_INVALID_FQDN;
      return -1;
S
Shengliang Guan 已提交
1116
    }
S
Shengliang Guan 已提交
1117

1118 1119 1120 1121 1122 1123
    int32_t code = taosGetFqdnPortFromEp(firstEp, &mgmtEpSet->eps[0]);
    if (code != TSDB_CODE_SUCCESS) {
      terrno = TSDB_CODE_TSC_INVALID_FQDN;
      return terrno;
    }

S
Shengliang Guan 已提交
1124 1125 1126 1127 1128 1129 1130
    mgmtEpSet->numOfEps++;
  }

  if (secondEp && secondEp[0] != 0) {
    if (strlen(secondEp) >= TSDB_EP_LEN) {
      terrno = TSDB_CODE_TSC_INVALID_FQDN;
      return -1;
1131
    }
S
Shengliang Guan 已提交
1132 1133 1134

    taosGetFqdnPortFromEp(secondEp, &mgmtEpSet->eps[mgmtEpSet->numOfEps]);
    mgmtEpSet->numOfEps++;
1135 1136 1137 1138 1139 1140 1141 1142 1143 1144
  }

  if (mgmtEpSet->numOfEps == 0) {
    terrno = TSDB_CODE_TSC_INVALID_FQDN;
    return -1;
  }

  return 0;
}

dengyihao's avatar
dengyihao 已提交
1145
STscObj* taosConnectImpl(const char* user, const char* auth, const char* db, __taos_async_fn_t fp, void* param,
L
Liu Jicong 已提交
1146
                         SAppInstInfo* pAppInfo, int connType) {
1147
  STscObj* pTscObj = createTscObj(user, auth, db, connType, pAppInfo);
1148 1149 1150 1151 1152
  if (NULL == pTscObj) {
    terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
    return pTscObj;
  }

1153
  SRequestObj* pRequest = createRequest(pTscObj, TDMT_MND_CONNECT);
1154 1155 1156
  if (pRequest == NULL) {
    destroyTscObj(pTscObj);
    terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
1157 1158 1159
    return NULL;
  }

1160
  SMsgSendInfo* body = buildConnectMsg(pRequest);
1161 1162

  int64_t transporterId = 0;
H
Haojun Liao 已提交
1163
  asyncSendMsgToServer(pTscObj->pAppInfo->pTransporter, &pTscObj->pAppInfo->mgmtEp.epSet, &transporterId, body);
1164 1165 1166

  tsem_wait(&pRequest->body.rspSem);
  if (pRequest->code != TSDB_CODE_SUCCESS) {
dengyihao's avatar
dengyihao 已提交
1167 1168
    const char* errorMsg =
        (pRequest->code == TSDB_CODE_RPC_FQDN_ERROR) ? taos_errstr(pRequest) : tstrerror(pRequest->code);
1169
    fprintf(stderr, "failed to connect to server, reason: %s\n\n", errorMsg);
1170

1171
    terrno = pRequest->code;
1172
    destroyRequest(pRequest);
1173
    taos_close_internal(pTscObj);
1174 1175
    pTscObj = NULL;
  } else {
D
dapan1121 已提交
1176
    tscDebug("0x%" PRIx64 " connection is opening, connId:%u, dnodeConn:%p, reqId:0x%" PRIx64, pTscObj->id,
dengyihao's avatar
dengyihao 已提交
1177
             pTscObj->connId, pTscObj->pAppInfo->pTransporter, pRequest->requestId);
1178 1179 1180 1181 1182 1183
    destroyRequest(pRequest);
  }

  return pTscObj;
}

1184
static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest) {
wafwerar's avatar
wafwerar 已提交
1185
  SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
1186 1187 1188 1189 1190
  if (pMsgSendInfo == NULL) {
    terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
    return NULL;
  }

dengyihao's avatar
dengyihao 已提交
1191
  pMsgSendInfo->msgType = TDMT_MND_CONNECT;
1192

1193
  pMsgSendInfo->requestObjRefId = pRequest->self;
dengyihao's avatar
dengyihao 已提交
1194
  pMsgSendInfo->requestId = pRequest->requestId;
H
Haojun Liao 已提交
1195
  pMsgSendInfo->fp = getMsgRspHandle(pMsgSendInfo->msgType);
dengyihao's avatar
dengyihao 已提交
1196
  pMsgSendInfo->param = pRequest;
1197

S
Shengliang Guan 已提交
1198 1199
  SConnectReq connectReq = {0};
  STscObj*    pObj = pRequest->pTscObj;
1200

H
Haojun Liao 已提交
1201
  char* db = getDbOfConnection(pObj);
1202
  if (db != NULL) {
S
Shengliang Guan 已提交
1203
    tstrncpy(connectReq.db, db, sizeof(connectReq.db));
1204
  }
wafwerar's avatar
wafwerar 已提交
1205
  taosMemoryFreeClear(db);
1206

X
Xiaoyu Wang 已提交
1207
  connectReq.connType = pObj->connType;
1208 1209
  connectReq.pid = appInfo.pid;
  connectReq.startTime = appInfo.startTime;
1210

S
Shengliang Guan 已提交
1211
  tstrncpy(connectReq.app, appInfo.appName, sizeof(connectReq.app));
dengyihao's avatar
dengyihao 已提交
1212 1213
  tstrncpy(connectReq.user, pObj->user, sizeof(connectReq.user));
  tstrncpy(connectReq.passwd, pObj->pass, sizeof(connectReq.passwd));
S
Shengliang Guan 已提交
1214 1215

  int32_t contLen = tSerializeSConnectReq(NULL, 0, &connectReq);
wafwerar's avatar
wafwerar 已提交
1216
  void*   pReq = taosMemoryMalloc(contLen);
S
Shengliang Guan 已提交
1217
  tSerializeSConnectReq(pReq, contLen, &connectReq);
1218

S
Shengliang Guan 已提交
1219 1220
  pMsgSendInfo->msgInfo.len = contLen;
  pMsgSendInfo->msgInfo.pData = pReq;
1221
  return pMsgSendInfo;
1222 1223
}

1224
static void destroySendMsgInfo(SMsgSendInfo* pMsgBody) {
H
Haojun Liao 已提交
1225
  assert(pMsgBody != NULL);
D
dapan1121 已提交
1226
  taosMemoryFreeClear(pMsgBody->target.dbFName);
wafwerar's avatar
wafwerar 已提交
1227 1228
  taosMemoryFreeClear(pMsgBody->msgInfo.pData);
  taosMemoryFreeClear(pMsgBody);
H
Haojun Liao 已提交
1229
}
1230

D
dapan1121 已提交
1231 1232 1233 1234
void updateTargetEpSet(SMsgSendInfo* pSendInfo, STscObj* pTscObj, SRpcMsg* pMsg, SEpSet* pEpSet) {
  if (NULL == pEpSet) {
    return;
  }
L
Liu Jicong 已提交
1235

D
dapan1121 已提交
1236 1237 1238
  switch (pSendInfo->target.type) {
    case TARGET_TYPE_MNODE:
      if (NULL == pTscObj) {
X
Xiaoyu Wang 已提交
1239 1240
        tscError("mnode epset changed but not able to update it, msg:%s, reqObjRefId:%" PRIx64,
                 TMSG_INFO(pMsg->msgType), pSendInfo->requestObjRefId);
D
dapan1121 已提交
1241 1242 1243
        return;
      }

D
dapan1121 已提交
1244
      SEpSet* pOrig = &pTscObj->pAppInfo->mgmtEp.epSet;
X
Xiaoyu Wang 已提交
1245 1246 1247 1248
      SEp*    pOrigEp = &pOrig->eps[pOrig->inUse];
      SEp*    pNewEp = &pEpSet->eps[pEpSet->inUse];
      tscDebug("mnode epset updated from %d/%d=>%s:%d to %d/%d=>%s:%d in client", pOrig->inUse, pOrig->numOfEps,
               pOrigEp->fqdn, pOrigEp->port, pEpSet->inUse, pEpSet->numOfEps, pNewEp->fqdn, pNewEp->port);
L
Liu Jicong 已提交
1249
      updateEpSet_s(&pTscObj->pAppInfo->mgmtEp, pEpSet);
D
dapan1121 已提交
1250 1251 1252
      break;
    case TARGET_TYPE_VNODE: {
      if (NULL == pTscObj) {
X
Xiaoyu Wang 已提交
1253 1254
        tscError("vnode epset changed but not able to update it, msg:%s, reqObjRefId:%" PRIx64,
                 TMSG_INFO(pMsg->msgType), pSendInfo->requestObjRefId);
D
dapan1121 已提交
1255 1256 1257 1258
        return;
      }

      SCatalog* pCatalog = NULL;
L
Liu Jicong 已提交
1259
      int32_t   code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCatalog);
D
dapan1121 已提交
1260
      if (code != TSDB_CODE_SUCCESS) {
L
Liu Jicong 已提交
1261 1262
        tscError("fail to get catalog handle, clusterId:%" PRIx64 ", error %s", pTscObj->pAppInfo->clusterId,
                 tstrerror(code));
D
dapan1121 已提交
1263 1264
        return;
      }
L
Liu Jicong 已提交
1265

D
dapan1121 已提交
1266
      catalogUpdateVgEpSet(pCatalog, pSendInfo->target.dbFName, pSendInfo->target.vgId, pEpSet);
D
dapan1121 已提交
1267
      taosMemoryFreeClear(pSendInfo->target.dbFName);
D
dapan1121 已提交
1268 1269 1270 1271 1272 1273 1274 1275
      break;
    }
    default:
      tscDebug("epset changed, not updated, msgType %s", TMSG_INFO(pMsg->msgType));
      break;
  }
}

1276
void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) {
S
Shengliang Guan 已提交
1277 1278
  SMsgSendInfo* pSendInfo = (SMsgSendInfo*)pMsg->info.ahandle;
  assert(pMsg->info.ahandle != NULL);
D
dapan1121 已提交
1279
  STscObj* pTscObj = NULL;
1280

D
dapan1121 已提交
1281 1282
  tscDebug("processMsgFromServer message: %s, code: %s", TMSG_INFO(pMsg->msgType), tstrerror(pMsg->code));

1283
  if (pSendInfo->requestObjRefId != 0) {
dengyihao's avatar
dengyihao 已提交
1284
    SRequestObj* pRequest = (SRequestObj*)taosAcquireRef(clientReqRefPool, pSendInfo->requestObjRefId);
D
dapan1121 已提交
1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301
    if (pRequest) {
      assert(pRequest->self == pSendInfo->requestObjRefId);

      pRequest->metric.rsp = taosGetTimestampUs();
      pTscObj = pRequest->pTscObj;
      /*
       * There is not response callback function for submit response.
       * The actual inserted number of points is the first number.
       */
      int32_t elapsed = pRequest->metric.rsp - pRequest->metric.start;
      if (pMsg->code == TSDB_CODE_SUCCESS) {
        tscDebug("0x%" PRIx64 " message:%s, code:%s rspLen:%d, elapsed:%d ms, reqId:0x%" PRIx64, pRequest->self,
                 TMSG_INFO(pMsg->msgType), tstrerror(pMsg->code), pMsg->contLen, elapsed / 1000, pRequest->requestId);
      } else {
        tscError("0x%" PRIx64 " SQL cmd:%s, code:%s rspLen:%d, elapsed time:%d ms, reqId:0x%" PRIx64, pRequest->self,
                 TMSG_INFO(pMsg->msgType), tstrerror(pMsg->code), pMsg->contLen, elapsed / 1000, pRequest->requestId);
      }
1302

D
dapan1121 已提交
1303 1304
      taosReleaseRef(clientReqRefPool, pSendInfo->requestObjRefId);
    }
1305 1306
  }

D
dapan1121 已提交
1307 1308
  updateTargetEpSet(pSendInfo, pTscObj, pMsg, pEpSet);

S
Shengliang Guan 已提交
1309
  SDataBuf buf = {.len = pMsg->contLen, .pData = NULL, .handle = pMsg->info.handle};
1310 1311

  if (pMsg->contLen > 0) {
wafwerar's avatar
wafwerar 已提交
1312
    buf.pData = taosMemoryCalloc(1, pMsg->contLen);
1313 1314 1315 1316 1317 1318
    if (buf.pData == NULL) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      pMsg->code = TSDB_CODE_OUT_OF_MEMORY;
    } else {
      memcpy(buf.pData, pMsg->pCont, pMsg->contLen);
    }
1319 1320
  }

1321
  pSendInfo->fp(pSendInfo->param, &buf, pMsg->code);
1322
  rpcFreeCont(pMsg->pCont);
1323
  destroySendMsgInfo(pSendInfo);
1324
}
1325

dengyihao's avatar
dengyihao 已提交
1326
TAOS* taos_connect_auth(const char* ip, const char* user, const char* auth, const char* db, uint16_t port) {
1327 1328 1329 1330 1331 1332 1333 1334 1335 1336
  tscDebug("try to connect to %s:%u by auth, user:%s db:%s", ip, port, user, db);
  if (user == NULL) {
    user = TSDB_DEFAULT_USER;
  }

  if (auth == NULL) {
    tscError("No auth info is given, failed to connect to server");
    return NULL;
  }

1337 1338
  STscObj* pObj = taos_connect_internal(ip, user, NULL, auth, db, port, CONN_TYPE__QUERY);
  if (pObj) {
D
dapan1121 已提交
1339 1340 1341
    int64_t *rid = taosMemoryCalloc(1, sizeof(int64_t));
    *rid = pObj->id;
    return (TAOS*)rid;
1342
  }
dengyihao's avatar
dengyihao 已提交
1343

D
dapan1121 已提交
1344
  return NULL;
1345 1346
}

dengyihao's avatar
dengyihao 已提交
1347 1348 1349
TAOS* taos_connect_l(const char* ip, int ipLen, const char* user, int userLen, const char* pass, int passLen,
                     const char* db, int dbLen, uint16_t port) {
  char ipStr[TSDB_EP_LEN] = {0};
1350
  char dbStr[TSDB_DB_NAME_LEN] = {0};
dengyihao's avatar
dengyihao 已提交
1351 1352
  char userStr[TSDB_USER_LEN] = {0};
  char passStr[TSDB_PASSWORD_LEN] = {0};
1353

dengyihao's avatar
dengyihao 已提交
1354
  strncpy(ipStr, ip, TMIN(TSDB_EP_LEN - 1, ipLen));
dengyihao's avatar
dengyihao 已提交
1355 1356
  strncpy(userStr, user, TMIN(TSDB_USER_LEN - 1, userLen));
  strncpy(passStr, pass, TMIN(TSDB_PASSWORD_LEN - 1, passLen));
dengyihao's avatar
dengyihao 已提交
1357
  strncpy(dbStr, db, TMIN(TSDB_DB_NAME_LEN - 1, dbLen));
1358
  return taos_connect(ipStr, userStr, passStr, dbStr, port);
1359 1360
}

L
Liu Jicong 已提交
1361
void doSetOneRowPtr(SReqResultInfo* pResultInfo) {
1362 1363 1364 1365 1366 1367 1368
  for (int32_t i = 0; i < pResultInfo->numOfCols; ++i) {
    SResultColumn* pCol = &pResultInfo->pCol[i];

    int32_t type = pResultInfo->fields[i].type;
    int32_t bytes = pResultInfo->fields[i].bytes;

    if (IS_VAR_DATA_TYPE(type)) {
1369
      if (!IS_VAR_NULL_TYPE(type, bytes) && pCol->offset[pResultInfo->current] != -1) {
1370 1371 1372 1373 1374 1375
        char* pStart = pResultInfo->pCol[i].offset[pResultInfo->current] + pResultInfo->pCol[i].pData;

        pResultInfo->length[i] = varDataLen(pStart);
        pResultInfo->row[i] = varDataVal(pStart);
      } else {
        pResultInfo->row[i] = NULL;
1376
        pResultInfo->length[i] = 0;
1377 1378 1379 1380
      }
    } else {
      if (!colDataIsNull_f(pCol->nullbitmap, pResultInfo->current)) {
        pResultInfo->row[i] = pResultInfo->pCol[i].pData + bytes * pResultInfo->current;
1381
        pResultInfo->length[i] = bytes;
1382 1383
      } else {
        pResultInfo->row[i] = NULL;
1384
        pResultInfo->length[i] = 0;
1385 1386 1387 1388 1389
      }
    }
  }
}

1390
void* doFetchRows(SRequestObj* pRequest, bool setupOneRowPtr, bool convertUcs4) {
D
dapan1121 已提交
1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401
  assert(pRequest != NULL);

  SReqResultInfo* pResultInfo = &pRequest->body.resInfo;
  if (pResultInfo->pData == NULL || pResultInfo->current >= pResultInfo->numOfRows) {
    // All data has returned to App already, no need to try again
    if (pResultInfo->completed) {
      pResultInfo->numOfRows = 0;
      return NULL;
    }

    SReqResultInfo* pResInfo = &pRequest->body.resInfo;
1402
    pRequest->code = schedulerFetchRows(pRequest->body.queryJob, (void**)&pResInfo->pData);
D
dapan1121 已提交
1403 1404 1405 1406 1407
    if (pRequest->code != TSDB_CODE_SUCCESS) {
      pResultInfo->numOfRows = 0;
      return NULL;
    }

L
Liu Jicong 已提交
1408 1409
    pRequest->code =
        setQueryResultFromRsp(&pRequest->body.resInfo, (SRetrieveTableRsp*)pResInfo->pData, convertUcs4, true);
D
dapan1121 已提交
1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430
    if (pRequest->code != TSDB_CODE_SUCCESS) {
      pResultInfo->numOfRows = 0;
      return NULL;
    }

    tscDebug("0x%" PRIx64 " fetch results, numOfRows:%d total Rows:%" PRId64 ", complete:%d, reqId:0x%" PRIx64,
             pRequest->self, pResInfo->numOfRows, pResInfo->totalRows, pResInfo->completed, pRequest->requestId);

    if (pResultInfo->numOfRows == 0) {
      return NULL;
    }
  }

  if (setupOneRowPtr) {
    doSetOneRowPtr(pResultInfo);
    pResultInfo->current += 1;
  }

  return pResultInfo->row;
}

H
Haojun Liao 已提交
1431 1432 1433
static void syncFetchFn(void* param, TAOS_RES* res, int32_t numOfRows) {
  SSyncQueryParam* pParam = param;
  tsem_post(&pParam->sem);
1434
}
D
dapan1121 已提交
1435

1436
void* doAsyncFetchRows(SRequestObj* pRequest, bool setupOneRowPtr, bool convertUcs4) {
1437
  assert(pRequest != NULL);
H
Haojun Liao 已提交
1438

H
Haojun Liao 已提交
1439
  SReqResultInfo* pResultInfo = &pRequest->body.resInfo;
H
Haojun Liao 已提交
1440
  if (pResultInfo->pData == NULL || pResultInfo->current >= pResultInfo->numOfRows) {
H
Haojun Liao 已提交
1441
    // All data has returned to App already, no need to try again
1442 1443 1444 1445 1446
    if (pResultInfo->completed) {
      pResultInfo->numOfRows = 0;
      return NULL;
    }

1447
    SSyncQueryParam* pParam = pRequest->body.param;
1448 1449 1450 1451

    // convert ucs4 to native multi-bytes string
    pResultInfo->convertUcs4 = convertUcs4;

1452
    taos_fetch_rows_a(pRequest, syncFetchFn, pParam);
1453
    tsem_wait(&pParam->sem);
H
Haojun Liao 已提交
1454 1455
  }

H
Haojun Liao 已提交
1456
  if (pRequest->code == TSDB_CODE_SUCCESS && pResultInfo->numOfRows > 0 && setupOneRowPtr) {
1457 1458
    doSetOneRowPtr(pResultInfo);
    pResultInfo->current += 1;
H
Haojun Liao 已提交
1459 1460 1461 1462 1463
  }

  return pResultInfo->row;
}

1464
static int32_t doPrepareResPtr(SReqResultInfo* pResInfo) {
H
Haojun Liao 已提交
1465
  if (pResInfo->row == NULL) {
L
Liu Jicong 已提交
1466 1467
    pResInfo->row = taosMemoryCalloc(pResInfo->numOfCols, POINTER_BYTES);
    pResInfo->pCol = taosMemoryCalloc(pResInfo->numOfCols, sizeof(SResultColumn));
wafwerar's avatar
wafwerar 已提交
1468 1469
    pResInfo->length = taosMemoryCalloc(pResInfo->numOfCols, sizeof(int32_t));
    pResInfo->convertBuf = taosMemoryCalloc(pResInfo->numOfCols, POINTER_BYTES);
1470

1471 1472 1473
    if (pResInfo->row == NULL || pResInfo->pCol == NULL || pResInfo->length == NULL || pResInfo->convertBuf == NULL) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }
1474
  }
1475 1476

  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
1477 1478
}

1479 1480 1481 1482 1483
static int32_t doConvertUCS4(SReqResultInfo* pResultInfo, int32_t numOfRows, int32_t numOfCols, int32_t* colLength) {
  for (int32_t i = 0; i < numOfCols; ++i) {
    int32_t type = pResultInfo->fields[i].type;
    int32_t bytes = pResultInfo->fields[i].bytes;

1484
    if (type == TSDB_DATA_TYPE_NCHAR && colLength[i] > 0) {
1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498
      char* p = taosMemoryRealloc(pResultInfo->convertBuf[i], colLength[i]);
      if (p == NULL) {
        return TSDB_CODE_OUT_OF_MEMORY;
      }

      pResultInfo->convertBuf[i] = p;

      SResultColumn* pCol = &pResultInfo->pCol[i];
      for (int32_t j = 0; j < numOfRows; ++j) {
        if (pCol->offset[j] != -1) {
          char* pStart = pCol->offset[j] + pCol->pData;

          int32_t len = taosUcs4ToMbs((TdUcs4*)varDataVal(pStart), varDataLen(pStart), varDataVal(p));
          ASSERT(len <= bytes);
D
stmt  
dapan1121 已提交
1499
          ASSERT((p + len) < (pResultInfo->convertBuf[i] + colLength[i]));
dengyihao's avatar
dengyihao 已提交
1500

1501 1502 1503 1504 1505 1506 1507 1508
          varDataSetLen(p, len);
          pCol->offset[j] = (p - pResultInfo->convertBuf[i]);
          p += (len + VARSTR_HEADER_SIZE);
        }
      }

      pResultInfo->pCol[i].pData = pResultInfo->convertBuf[i];
      pResultInfo->row[i] = pResultInfo->pCol[i].pData;
wmmhello's avatar
wmmhello 已提交
1509 1510
    }
  }
1511

wmmhello's avatar
wmmhello 已提交
1512 1513
  return TSDB_CODE_SUCCESS;
}
1514

dengyihao's avatar
dengyihao 已提交
1515
static int32_t estimateJsonLen(SReqResultInfo* pResultInfo, int32_t numOfCols, int32_t numOfRows) {
wmmhello's avatar
wmmhello 已提交
1516 1517
  char* p = (char*)pResultInfo->pData;

dengyihao's avatar
dengyihao 已提交
1518
  int32_t  len = sizeof(int32_t) + sizeof(uint64_t) + numOfCols * (sizeof(int16_t) + sizeof(int32_t));
wmmhello's avatar
wmmhello 已提交
1519 1520 1521 1522 1523 1524 1525 1526 1527
  int32_t* colLength = (int32_t*)(p + len);
  len += sizeof(int32_t) * numOfCols;

  char* pStart = p + len;
  for (int32_t i = 0; i < numOfCols; ++i) {
    int32_t colLen = htonl(colLength[i]);

    if (pResultInfo->fields[i].type == TSDB_DATA_TYPE_JSON) {
      int32_t* offset = (int32_t*)pStart;
dengyihao's avatar
dengyihao 已提交
1528
      int32_t  lenTmp = numOfRows * sizeof(int32_t);
wmmhello's avatar
wmmhello 已提交
1529 1530
      len += lenTmp;
      pStart += lenTmp;
1531 1532

      for (int32_t j = 0; j < numOfRows; ++j) {
wmmhello's avatar
wmmhello 已提交
1533 1534 1535 1536 1537 1538 1539 1540 1541
        if (offset[j] == -1) {
          continue;
        }
        char* data = offset[j] + pStart;

        int32_t jsonInnerType = *data;
        char*   jsonInnerData = data + CHAR_BYTES;
        if (jsonInnerType == TSDB_DATA_TYPE_NULL) {
          len += (VARSTR_HEADER_SIZE + strlen(TSDB_DATA_NULL_STR_L));
wmmhello's avatar
wmmhello 已提交
1542
        } else if (tTagIsJson(data)) {
wmmhello's avatar
wmmhello 已提交
1543 1544 1545 1546 1547 1548 1549 1550 1551 1552 1553 1554 1555 1556 1557 1558 1559 1560 1561 1562 1563 1564 1565 1566 1567
          len += (VARSTR_HEADER_SIZE + ((const STag*)(data))->len);
        } else if (jsonInnerType == TSDB_DATA_TYPE_NCHAR) {  // value -> "value"
          len += varDataTLen(jsonInnerData) + CHAR_BYTES * 2;
        } else if (jsonInnerType == TSDB_DATA_TYPE_DOUBLE) {
          len += (VARSTR_HEADER_SIZE + 32);
        } else if (jsonInnerType == TSDB_DATA_TYPE_BOOL) {
          len += (VARSTR_HEADER_SIZE + 5);
        } else {
          ASSERT(0);
        }
      }
    } else if (IS_VAR_DATA_TYPE(pResultInfo->fields[i].type)) {
      int32_t lenTmp = numOfRows * sizeof(int32_t);
      len += (lenTmp + colLen);
      pStart += lenTmp;
    } else {
      int32_t lenTmp = BitmapLen(pResultInfo->numOfRows);
      len += (lenTmp + colLen);
      pStart += lenTmp;
    }
    pStart += colLen;
  }
  return len;
}

wmmhello's avatar
wmmhello 已提交
1568 1569 1570 1571 1572 1573 1574 1575
static int32_t doConvertJson(SReqResultInfo* pResultInfo, int32_t numOfCols, int32_t numOfRows) {
  bool needConvert = false;
  for (int32_t i = 0; i < numOfCols; ++i) {
    if (pResultInfo->fields[i].type == TSDB_DATA_TYPE_JSON) {
      needConvert = true;
      break;
    }
  }
dengyihao's avatar
dengyihao 已提交
1576
  if (!needConvert) return TSDB_CODE_SUCCESS;
1577

dengyihao's avatar
dengyihao 已提交
1578
  char*   p = (char*)pResultInfo->pData;
wmmhello's avatar
wmmhello 已提交
1579 1580
  int32_t dataLen = estimateJsonLen(pResultInfo, numOfCols, numOfRows);

wmmhello's avatar
wmmhello 已提交
1581
  pResultInfo->convertJson = taosMemoryCalloc(1, dataLen);
dengyihao's avatar
dengyihao 已提交
1582
  if (pResultInfo->convertJson == NULL) return TSDB_CODE_OUT_OF_MEMORY;
wmmhello's avatar
wmmhello 已提交
1583 1584
  char* p1 = pResultInfo->convertJson;

wmmhello's avatar
wmmhello 已提交
1585
  int32_t len = sizeof(int32_t) + sizeof(uint64_t) + numOfCols * (sizeof(int16_t) + sizeof(int32_t));
wmmhello's avatar
wmmhello 已提交
1586
  memcpy(p1, p, len);
1587

wmmhello's avatar
wmmhello 已提交
1588 1589
  p += len;
  p1 += len;
1590

wmmhello's avatar
wmmhello 已提交
1591 1592 1593 1594 1595 1596
  len = sizeof(int32_t) * numOfCols;
  int32_t* colLength = (int32_t*)p;
  int32_t* colLength1 = (int32_t*)p1;
  memcpy(p1, p, len);
  p += len;
  p1 += len;
1597

wmmhello's avatar
wmmhello 已提交
1598 1599 1600 1601 1602 1603 1604 1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625
  char* pStart = p;
  char* pStart1 = p1;
  for (int32_t i = 0; i < numOfCols; ++i) {
    int32_t colLen = htonl(colLength[i]);
    int32_t colLen1 = htonl(colLength1[i]);
    ASSERT(colLen < dataLen);

    if (pResultInfo->fields[i].type == TSDB_DATA_TYPE_JSON) {
      int32_t* offset = (int32_t*)pStart;
      int32_t* offset1 = (int32_t*)pStart1;
      len = numOfRows * sizeof(int32_t);
      memcpy(pStart1, pStart, len);
      pStart += len;
      pStart1 += len;

      len = 0;
      for (int32_t j = 0; j < numOfRows; ++j) {
        if (offset[j] == -1) {
          continue;
        }
        char* data = offset[j] + pStart;

        int32_t jsonInnerType = *data;
        char*   jsonInnerData = data + CHAR_BYTES;
        char    dst[TSDB_MAX_JSON_TAG_LEN] = {0};
        if (jsonInnerType == TSDB_DATA_TYPE_NULL) {
          sprintf(varDataVal(dst), "%s", TSDB_DATA_NULL_STR_L);
          varDataSetLen(dst, strlen(varDataVal(dst)));
wmmhello's avatar
wmmhello 已提交
1626
        } else if (tTagIsJson(data)) {
wmmhello's avatar
wmmhello 已提交
1627 1628 1629 1630 1631 1632 1633 1634 1635 1636
          char* jsonString = parseTagDatatoJson(data);
          STR_TO_VARSTR(dst, jsonString);
          taosMemoryFree(jsonString);
        } else if (jsonInnerType == TSDB_DATA_TYPE_NCHAR) {  // value -> "value"
          *(char*)varDataVal(dst) = '\"';
          int32_t length = taosUcs4ToMbs((TdUcs4*)varDataVal(jsonInnerData), varDataLen(jsonInnerData),
                                         varDataVal(dst) + CHAR_BYTES);
          if (length <= 0) {
            tscError("charset:%s to %s. convert failed.", DEFAULT_UNICODE_ENCODEC, tsCharset);
            length = 0;
1637
          }
wmmhello's avatar
wmmhello 已提交
1638 1639 1640 1641 1642 1643 1644 1645 1646 1647 1648
          varDataSetLen(dst, length + CHAR_BYTES * 2);
          *(char*)POINTER_SHIFT(varDataVal(dst), length + CHAR_BYTES) = '\"';
        } else if (jsonInnerType == TSDB_DATA_TYPE_DOUBLE) {
          double jsonVd = *(double*)(jsonInnerData);
          sprintf(varDataVal(dst), "%.9lf", jsonVd);
          varDataSetLen(dst, strlen(varDataVal(dst)));
        } else if (jsonInnerType == TSDB_DATA_TYPE_BOOL) {
          sprintf(varDataVal(dst), "%s", (*((char*)jsonInnerData) == 1) ? "true" : "false");
          varDataSetLen(dst, strlen(varDataVal(dst)));
        } else {
          ASSERT(0);
1649
        }
wmmhello's avatar
wmmhello 已提交
1650

dengyihao's avatar
dengyihao 已提交
1651
        offset1[j] = len;
wmmhello's avatar
wmmhello 已提交
1652 1653
        memcpy(pStart1 + len, dst, varDataTLen(dst));
        len += varDataTLen(dst);
1654
      }
wmmhello's avatar
wmmhello 已提交
1655 1656 1657 1658 1659 1660 1661 1662 1663 1664 1665 1666 1667 1668
      colLen1 = len;
      colLength1[i] = htonl(len);
    } else if (IS_VAR_DATA_TYPE(pResultInfo->fields[i].type)) {
      len = numOfRows * sizeof(int32_t);
      memcpy(pStart1, pStart, len);
      pStart += len;
      pStart1 += len;
      memcpy(pStart1, pStart, colLen);
    } else {
      len = BitmapLen(pResultInfo->numOfRows);
      memcpy(pStart1, pStart, len);
      pStart += len;
      pStart1 += len;
      memcpy(pStart1, pStart, colLen);
1669
    }
wmmhello's avatar
wmmhello 已提交
1670 1671
    pStart += colLen;
    pStart1 += colLen1;
1672 1673
  }

wmmhello's avatar
wmmhello 已提交
1674
  pResultInfo->pData = pResultInfo->convertJson;
1675 1676 1677
  return TSDB_CODE_SUCCESS;
}

L
Liu Jicong 已提交
1678 1679
int32_t setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32_t numOfCols, int32_t numOfRows,
                         bool convertUcs4) {
H
Haojun Liao 已提交
1680 1681
  assert(numOfCols > 0 && pFields != NULL && pResultInfo != NULL);
  if (numOfRows == 0) {
1682
    return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
1683 1684
  }

1685 1686 1687 1688
  int32_t code = doPrepareResPtr(pResultInfo);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }
wmmhello's avatar
wmmhello 已提交
1689 1690 1691 1692
  code = doConvertJson(pResultInfo, numOfCols, numOfRows);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }
H
Haojun Liao 已提交
1693

L
Liu Jicong 已提交
1694
  char* p = (char*)pResultInfo->pData;
1695

L
Liu Jicong 已提交
1696
  int32_t dataLen = *(int32_t*)p;
1697 1698
  p += sizeof(int32_t);

L
Liu Jicong 已提交
1699
  uint64_t groupId = *(uint64_t*)p;
1700 1701
  p += sizeof(uint64_t);

1702
  // check fields
1703 1704
  for (int32_t i = 0; i < numOfCols; ++i) {
    int16_t type = *(int16_t*)p;
1705 1706
    p += sizeof(int16_t);

1707
    int32_t bytes = *(int32_t*)p;
1708 1709
    p += sizeof(int32_t);

L
Liu Jicong 已提交
1710
    /*ASSERT(type == pFields[i].type && bytes == pFields[i].bytes);*/
1711 1712
  }

1713 1714 1715 1716
  int32_t* colLength = (int32_t*)p;
  p += sizeof(int32_t) * numOfCols;

  char* pStart = p;
H
Haojun Liao 已提交
1717
  for (int32_t i = 0; i < numOfCols; ++i) {
1718
    colLength[i] = htonl(colLength[i]);
1719
    ASSERT(colLength[i] < dataLen);
1720 1721 1722 1723 1724 1725 1726 1727 1728 1729

    if (IS_VAR_DATA_TYPE(pResultInfo->fields[i].type)) {
      pResultInfo->pCol[i].offset = (int32_t*)pStart;
      pStart += numOfRows * sizeof(int32_t);
    } else {
      pResultInfo->pCol[i].nullbitmap = pStart;
      pStart += BitmapLen(pResultInfo->numOfRows);
    }

    pResultInfo->pCol[i].pData = pStart;
H
Haojun Liao 已提交
1730
    pResultInfo->length[i] = pResultInfo->fields[i].bytes;
1731 1732 1733
    pResultInfo->row[i] = pResultInfo->pCol[i].pData;

    pStart += colLength[i];
1734
  }
1735

dengyihao's avatar
dengyihao 已提交
1736
  if (convertUcs4) {
1737
    code = doConvertUCS4(pResultInfo, numOfRows, numOfCols, colLength);
1738 1739
  }

1740
  return code;
S
Shengliang Guan 已提交
1741 1742
}

H
Haojun Liao 已提交
1743
char* getDbOfConnection(STscObj* pObj) {
dengyihao's avatar
dengyihao 已提交
1744
  char* p = NULL;
wafwerar's avatar
wafwerar 已提交
1745
  taosThreadMutexLock(&pObj->mutex);
1746 1747 1748 1749
  size_t len = strlen(pObj->db);
  if (len > 0) {
    p = strndup(pObj->db, tListLen(pObj->db));
  }
S
Shengliang Guan 已提交
1750

wafwerar's avatar
wafwerar 已提交
1751
  taosThreadMutexUnlock(&pObj->mutex);
1752 1753 1754 1755 1756
  return p;
}

void setConnectionDB(STscObj* pTscObj, const char* db) {
  assert(db != NULL && pTscObj != NULL);
wafwerar's avatar
wafwerar 已提交
1757
  taosThreadMutexLock(&pTscObj->mutex);
1758
  tstrncpy(pTscObj->db, db, tListLen(pTscObj->db));
wafwerar's avatar
wafwerar 已提交
1759
  taosThreadMutexUnlock(&pTscObj->mutex);
1760
}
S
Shengliang Guan 已提交
1761

H
Haojun Liao 已提交
1762 1763 1764 1765 1766 1767 1768 1769 1770 1771
void resetConnectDB(STscObj* pTscObj) {
  if (pTscObj == NULL) {
    return;
  }

  taosThreadMutexLock(&pTscObj->mutex);
  pTscObj->db[0] = 0;
  taosThreadMutexUnlock(&pTscObj->mutex);
}

L
Liu Jicong 已提交
1772 1773
int32_t setQueryResultFromRsp(SReqResultInfo* pResultInfo, const SRetrieveTableRsp* pRsp, bool convertUcs4,
                              bool freeAfterUse) {
1774 1775
  assert(pResultInfo != NULL && pRsp != NULL);

L
Liu Jicong 已提交
1776 1777
  if (freeAfterUse) taosMemoryFreeClear(pResultInfo->pRspMsg);

L
Liu Jicong 已提交
1778 1779 1780 1781 1782
  pResultInfo->pRspMsg = (const char*)pRsp;
  pResultInfo->pData = (void*)pRsp->data;
  pResultInfo->numOfRows = htonl(pRsp->numOfRows);
  pResultInfo->current = 0;
  pResultInfo->completed = (pRsp->completed == 1);
1783
  pResultInfo->payloadLen = htonl(pRsp->compLen);
L
Liu Jicong 已提交
1784
  pResultInfo->precision = pRsp->precision;
H
Haojun Liao 已提交
1785

1786
  // TODO handle the compressed case
1787
  pResultInfo->totalRows += pResultInfo->numOfRows;
1788 1789
  return setResultDataPtr(pResultInfo, pResultInfo->fields, pResultInfo->numOfCols, pResultInfo->numOfRows,
                          convertUcs4);
L
fix  
Liu Jicong 已提交
1790
}
S
Shengliang Guan 已提交
1791 1792 1793 1794 1795 1796

TSDB_SERVER_STATUS taos_check_server_status(const char* fqdn, int port, char* details, int maxlen) {
  TSDB_SERVER_STATUS code = TSDB_SRV_STATUS_UNAVAILABLE;
  void*              clientRpc = NULL;
  SServerStatusRsp   statusRsp = {0};
  SEpSet             epSet = {.inUse = 0, .numOfEps = 1};
S
Shengliang Guan 已提交
1797
  SRpcMsg            rpcMsg = {.info.ahandle = (void*)0x9526, .msgType = TDMT_DND_SERVER_STATUS};
S
Shengliang Guan 已提交
1798 1799 1800 1801 1802 1803 1804 1805 1806 1807 1808 1809 1810 1811 1812 1813 1814 1815
  SRpcMsg            rpcRsp = {0};
  SRpcInit           rpcInit = {0};
  char               pass[TSDB_PASSWORD_LEN + 1] = {0};

  rpcInit.label = "CHK";
  rpcInit.numOfThreads = 1;
  rpcInit.cfp = NULL;
  rpcInit.sessions = 16;
  rpcInit.connType = TAOS_CONN_CLIENT;
  rpcInit.idleTime = tsShellActivityTimer * 1000;
  rpcInit.user = "_dnd";

  clientRpc = rpcOpen(&rpcInit);
  if (clientRpc == NULL) {
    tscError("failed to init server status client");
    goto _OVER;
  }

S
Shengliang Guan 已提交
1816 1817 1818 1819 1820 1821 1822 1823
  if (fqdn == NULL) {
    fqdn = tsLocalFqdn;
  }

  if (port == 0) {
    port = tsServerPort;
  }

S
Shengliang Guan 已提交
1824 1825 1826 1827 1828 1829 1830 1831 1832 1833 1834 1835 1836 1837 1838
  tstrncpy(epSet.eps[0].fqdn, fqdn, TSDB_FQDN_LEN);
  epSet.eps[0].port = (uint16_t)port;
  rpcSendRecv(clientRpc, &epSet, &rpcMsg, &rpcRsp);

  if (rpcRsp.code != 0 || rpcRsp.contLen <= 0 || rpcRsp.pCont == NULL) {
    tscError("failed to send server status req since %s", terrstr());
    goto _OVER;
  }

  if (tDeserializeSServerStatusRsp(rpcRsp.pCont, rpcRsp.contLen, &statusRsp) != 0) {
    tscError("failed to parse server status rsp since %s", terrstr());
    goto _OVER;
  }

  code = statusRsp.statusCode;
1839
  if (details != NULL) {
S
Shengliang Guan 已提交
1840 1841 1842 1843 1844 1845 1846 1847 1848 1849 1850 1851
    tstrncpy(details, statusRsp.details, maxlen);
  }

_OVER:
  if (clientRpc != NULL) {
    rpcClose(clientRpc);
  }
  if (rpcRsp.pCont != NULL) {
    rpcFreeCont(rpcRsp.pCont);
  }
  return code;
}
D
dapan1121 已提交
1852

X
Xiaoyu Wang 已提交
1853 1854
int32_t appendTbToReq(SArray* pList, int32_t pos1, int32_t len1, int32_t pos2, int32_t len2, const char* str,
                      int32_t acctId, char* db) {
D
dapan1121 已提交
1855
  SName name;
X
Xiaoyu Wang 已提交
1856

D
dapan1121 已提交
1857 1858 1859 1860
  if (len1 <= 0) {
    return -1;
  }

X
Xiaoyu Wang 已提交
1861 1862 1863 1864
  const char* dbName = db;
  const char* tbName = NULL;
  int32_t     dbLen = 0;
  int32_t     tbLen = 0;
D
dapan1121 已提交
1865 1866 1867 1868 1869 1870 1871 1872 1873 1874
  if (len2 > 0) {
    dbName = str + pos1;
    dbLen = len1;
    tbName = str + pos2;
    tbLen = len2;
  } else {
    dbLen = strlen(db);
    tbName = str + pos1;
    tbLen = len1;
  }
X
Xiaoyu Wang 已提交
1875

D
dapan1121 已提交
1876 1877 1878 1879 1880 1881 1882 1883 1884 1885 1886 1887 1888 1889 1890 1891 1892 1893 1894
  if (tNameSetDbName(&name, acctId, dbName, dbLen)) {
    return -1;
  }

  if (tNameAddTbName(&name, tbName, tbLen)) {
    return -1;
  }

  taosArrayPush(pList, &name);

  return TSDB_CODE_SUCCESS;
}

int32_t transferTableNameList(const char* tbList, int32_t acctId, char* dbName, SArray** pReq) {
  *pReq = taosArrayInit(10, sizeof(SName));
  if (NULL == *pReq) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return terrno;
  }
X
Xiaoyu Wang 已提交
1895 1896

  bool    inEscape = false;
D
dapan1121 已提交
1897
  int32_t code = 0;
X
Xiaoyu Wang 已提交
1898

D
dapan1121 已提交
1899 1900 1901 1902 1903 1904
  int32_t vIdx = 0;
  int32_t vPos[2];
  int32_t vLen[2];

  memset(vPos, -1, sizeof(vPos));
  memset(vLen, 0, sizeof(vLen));
X
Xiaoyu Wang 已提交
1905 1906

  for (int32_t i = 0;; ++i) {
D
dapan1121 已提交
1907 1908 1909 1910 1911 1912 1913 1914 1915 1916 1917 1918
    if (0 == *(tbList + i)) {
      if (vPos[vIdx] >= 0 && vLen[vIdx] <= 0) {
        vLen[vIdx] = i - vPos[vIdx];
      }

      code = appendTbToReq(*pReq, vPos[0], vLen[0], vPos[1], vLen[1], tbList, acctId, dbName);
      if (code) {
        goto _return;
      }

      break;
    }
X
Xiaoyu Wang 已提交
1919

D
dapan1121 已提交
1920 1921 1922 1923 1924 1925 1926 1927 1928 1929 1930 1931 1932 1933 1934 1935 1936 1937 1938 1939 1940 1941 1942 1943 1944 1945 1946 1947 1948 1949 1950 1951 1952 1953 1954 1955 1956 1957 1958 1959 1960 1961 1962 1963 1964 1965
    if ('`' == *(tbList + i)) {
      inEscape = !inEscape;
      if (!inEscape) {
        if (vPos[vIdx] >= 0) {
          vLen[vIdx] = i - vPos[vIdx];
        } else {
          goto _return;
        }
      }

      continue;
    }

    if (inEscape) {
      if (vPos[vIdx] < 0) {
        vPos[vIdx] = i;
      }
      continue;
    }

    if ('.' == *(tbList + i)) {
      if (vPos[vIdx] < 0) {
        goto _return;
      }
      if (vLen[vIdx] <= 0) {
        vLen[vIdx] = i - vPos[vIdx];
      }
      vIdx++;
      if (vIdx >= 2) {
        goto _return;
      }
      continue;
    }

    if (',' == *(tbList + i)) {
      if (vPos[vIdx] < 0) {
        goto _return;
      }
      if (vLen[vIdx] <= 0) {
        vLen[vIdx] = i - vPos[vIdx];
      }

      code = appendTbToReq(*pReq, vPos[0], vLen[0], vPos[1], vLen[1], tbList, acctId, dbName);
      if (code) {
        goto _return;
      }
X
Xiaoyu Wang 已提交
1966

D
dapan1121 已提交
1967 1968 1969 1970 1971 1972 1973 1974 1975 1976 1977 1978 1979
      memset(vPos, -1, sizeof(vPos));
      memset(vLen, 0, sizeof(vLen));
      vIdx = 0;
      continue;
    }

    if (' ' == *(tbList + i) || '\r' == *(tbList + i) || '\t' == *(tbList + i) || '\n' == *(tbList + i)) {
      if (vPos[vIdx] >= 0 && vLen[vIdx] <= 0) {
        vLen[vIdx] = i - vPos[vIdx];
      }
      continue;
    }

X
Xiaoyu Wang 已提交
1980
    if (('a' <= *(tbList + i) && 'z' >= *(tbList + i)) || ('A' <= *(tbList + i) && 'Z' >= *(tbList + i)) ||
D
dapan1121 已提交
1981 1982 1983 1984 1985 1986 1987 1988 1989 1990 1991 1992 1993 1994 1995 1996 1997 1998 1999 2000 2001
        ('0' <= *(tbList + i) && '9' >= *(tbList + i))) {
      if (vLen[vIdx] > 0) {
        goto _return;
      }
      if (vPos[vIdx] < 0) {
        vPos[vIdx] = i;
      }
      continue;
    }

    goto _return;
  }

  return TSDB_CODE_SUCCESS;

_return:

  terrno = TSDB_CODE_TSC_INVALID_OPERATION;

  taosArrayDestroy(*pReq);
  *pReq = NULL;
X
Xiaoyu Wang 已提交
2002

D
dapan1121 已提交
2003 2004 2005 2006
  return terrno;
}

void syncCatalogFn(SMetaData* pResult, void* param, int32_t code) {
X
Xiaoyu Wang 已提交
2007
  SSyncQueryParam* pParam = param;
D
dapan1121 已提交
2008 2009 2010 2011 2012
  pParam->pRequest->code = code;

  tsem_post(&pParam->sem);
}

X
Xiaoyu Wang 已提交
2013 2014
void syncQueryFn(void* param, void* res, int32_t code) {
  SSyncQueryParam* pParam = param;
D
dapan1121 已提交
2015
  pParam->pRequest = res;
D
dapan1121 已提交
2016 2017 2018
  if (pParam->pRequest) {
    pParam->pRequest->code = code;
  }
D
dapan1121 已提交
2019 2020 2021 2022

  tsem_post(&pParam->sem);
}

X
Xiaoyu Wang 已提交
2023
void taosAsyncQueryImpl(TAOS* taos, const char* sql, __taos_async_fn_t fp, void* param, bool validateOnly) {
D
dapan1121 已提交
2024 2025 2026 2027 2028 2029 2030 2031
  if (NULL == taos) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    fp(param, NULL, terrno);
    return;
  }

  int64_t rid = *(int64_t*)taos;
  STscObj* pTscObj = acquireTscObj(rid);
D
dapan1121 已提交
2032 2033 2034
  if (pTscObj == NULL || sql == NULL || NULL == fp) {
    terrno = TSDB_CODE_INVALID_PARA;
    if (pTscObj) {
D
dapan1121 已提交
2035
      releaseTscObj(rid);
D
dapan1121 已提交
2036 2037 2038 2039 2040 2041 2042 2043 2044 2045 2046
    } else {
      terrno = TSDB_CODE_TSC_DISCONNECTED;
    }
    fp(param, NULL, terrno);
    return;
  }

  size_t sqlLen = strlen(sql);
  if (sqlLen > (size_t)TSDB_MAX_ALLOWED_SQL_LEN) {
    tscError("sql string exceeds max length:%d", TSDB_MAX_ALLOWED_SQL_LEN);
    terrno = TSDB_CODE_TSC_EXCEED_SQL_LIMIT;
D
dapan1121 已提交
2047
    releaseTscObj(rid);
D
dapan1121 已提交
2048 2049 2050 2051 2052

    fp(param, NULL, terrno);
    return;
  }

X
Xiaoyu Wang 已提交
2053
  SRequestObj* pRequest = NULL;
D
dapan1121 已提交
2054 2055 2056
  int32_t      code = buildRequest(pTscObj, sql, sqlLen, &pRequest);
  if (code != TSDB_CODE_SUCCESS) {
    terrno = code;
D
dapan1121 已提交
2057
    releaseTscObj(rid);
D
dapan1121 已提交
2058 2059 2060 2061 2062 2063 2064 2065
    fp(param, NULL, terrno);
    return;
  }

  pRequest->validateOnly = validateOnly;
  pRequest->body.queryFp = fp;
  pRequest->body.param = param;
  doAsyncQuery(pRequest, false);
D
dapan1121 已提交
2066
  releaseTscObj(rid);  
D
dapan1121 已提交
2067 2068
}

X
Xiaoyu Wang 已提交
2069
TAOS_RES* taosQueryImpl(TAOS* taos, const char* sql, bool validateOnly) {
D
dapan1121 已提交
2070 2071 2072 2073 2074
  if (NULL == taos) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return NULL;
  }

D
dapan1121 已提交
2075 2076
  int64_t rid = *(int64_t*)taos;
  STscObj* pTscObj = acquireTscObj(rid);
D
dapan1121 已提交
2077 2078 2079 2080 2081 2082
  if (pTscObj == NULL || sql == NULL) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return NULL;
  }

#if SYNC_ON_TOP_OF_ASYNC
X
Xiaoyu Wang 已提交
2083
  SSyncQueryParam* param = taosMemoryCalloc(1, sizeof(SSyncQueryParam));
D
dapan1121 已提交
2084 2085
  tsem_init(&param->sem, 0, 0);

D
dapan1121 已提交
2086
  taosAsyncQueryImpl((TAOS*)&rid, sql, syncQueryFn, param, validateOnly);
D
dapan1121 已提交
2087 2088
  tsem_wait(&param->sem);

D
dapan1121 已提交
2089
  releaseTscObj(rid);
D
dapan1121 已提交
2090 2091 2092 2093 2094

  return param->pRequest;
#else
  size_t sqlLen = strlen(sql);
  if (sqlLen > (size_t)TSDB_MAX_ALLOWED_SQL_LEN) {
D
dapan1121 已提交
2095
    releaseTscObj(rid);
D
dapan1121 已提交
2096 2097 2098 2099 2100
    tscError("sql string exceeds max length:%d", TSDB_MAX_ALLOWED_SQL_LEN);
    terrno = TSDB_CODE_TSC_EXCEED_SQL_LIMIT;
    return NULL;
  }

X
Xiaoyu Wang 已提交
2101
  TAOS_RES* pRes = execQuery(pTscObj, sql, sqlLen, validateOnly);
D
dapan1121 已提交
2102

D
dapan1121 已提交
2103
  releaseTscObj(rid);
D
dapan1121 已提交
2104 2105 2106 2107

  return pRes;
#endif
}