clientImpl.c 38.1 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

dengyihao's avatar
dengyihao 已提交
61
TAOS* taos_connect_internal(const char* ip, const char* user, const char* pass, const char* auth, const char* db,
L
Liu Jicong 已提交
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));
dengyihao's avatar
dengyihao 已提交
120
    p->mgmtEp = epSet;
H
Haojun Liao 已提交
121
    p->pTransporter = openTransporter(user, secretEncrypt, tsNumOfCores);
D
dapan 已提交
122
    p->pAppHbMgr = appHbMgrInit(p, key);
H
Haojun Liao 已提交
123
    taosHashPut(appInfo.pInstMap, key, strlen(key), &p, POINTER_BYTES);
124

H
Haojun Liao 已提交
125
    pInst = &p;
126 127
  }

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

wafwerar's avatar
wafwerar 已提交
130
  taosMemoryFreeClear(key);
L
Liu Jicong 已提交
131
  return taosConnectImpl(user, &secretEncrypt[0], localDb, NULL, NULL, *pInst, connType);
132 133
}

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

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

X
Xiaoyu Wang 已提交
148 149 150
  strntolower((*pRequest)->sqlstr, sql, (int32_t)sqlLen);
  (*pRequest)->sqlstr[sqlLen] = 0;
  (*pRequest)->sqlLen = sqlLen;
151

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

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

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

167 168 169 170 171 172 173 174 175 176
  SParseContext cxt = {.requestId = pRequest->requestId,
                       .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,
177 178
                       .pUser = pTscObj->user,
                       .isSuperUser = (0 == strcmp(pTscObj->user, TSDB_DEFAULT_USER))};
H
Haojun Liao 已提交
179

H
Haojun Liao 已提交
180 181
  cxt.mgmtEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
  int32_t code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &cxt.pCatalog);
H
Haojun Liao 已提交
182 183 184 185
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

X
Xiaoyu Wang 已提交
186
  code = qParseSql(&cxt, pQuery);
X
Xiaoyu Wang 已提交
187 188 189
  if (TSDB_CODE_SUCCESS == code) {
    if ((*pQuery)->haveResultSet) {
      setResSchemaInfo(&pRequest->body.resInfo, (*pQuery)->pResSchema, (*pQuery)->numOfResCols);
190
      setResPrecision(&pRequest->body.resInfo, (*pQuery)->precision);
X
Xiaoyu Wang 已提交
191
    }
192 193
  }
  if (TSDB_CODE_SUCCESS == code || NEED_CLIENT_HANDLE_ERROR(code)) {
wafwerar's avatar
wafwerar 已提交
194 195
    TSWAP(pRequest->dbList, (*pQuery)->pDbList);
    TSWAP(pRequest->tableList, (*pQuery)->pTableList);
X
Xiaoyu Wang 已提交
196
  }
197

X
Xiaoyu Wang 已提交
198 199
  return code;
}
H
Haojun Liao 已提交
200

201 202
int32_t execLocalCmd(SRequestObj* pRequest, SQuery* pQuery) {
  SRetrieveTableRsp* pRsp = NULL;
L
Liu Jicong 已提交
203
  int32_t            code = qExecCommand(pQuery->pRoot, &pRsp);
204
  if (TSDB_CODE_SUCCESS == code && NULL != pRsp) {
205
    code = setQueryResultFromRsp(&pRequest->body.resInfo, pRsp, false);
206 207 208 209
  }
  return code;
}

X
Xiaoyu Wang 已提交
210
int32_t execDdlQuery(SRequestObj* pRequest, SQuery* pQuery) {
X
Xiaoyu Wang 已提交
211 212 213 214 215
  // drop table if exists not_exists_table
  if (NULL == pQuery->pCmdMsg) {
    return TSDB_CODE_SUCCESS;
  }

216 217 218
  SCmdMsgInfo* pMsgInfo = pQuery->pCmdMsg;
  pRequest->type = pMsgInfo->msgType;
  pRequest->body.requestMsg = (SDataBuf){.pData = pMsgInfo->pMsg, .len = pMsgInfo->msgLen, .handle = NULL};
L
Liu Jicong 已提交
219
  pMsgInfo->pMsg = NULL;  // pMsg transferred to SMsgSendInfo management
220 221 222

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

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

227
  tsem_wait(&pRequest->body.rspSem);
X
Xiaoyu Wang 已提交
228 229
  return TSDB_CODE_SUCCESS;
}
X
Xiaoyu Wang 已提交
230

231 232
int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArray* pNodeList) {
  pRequest->type = pQuery->msgType;
L
Liu Jicong 已提交
233 234 235 236 237 238
  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,
239
                      .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE};
240 241 242 243 244 245
  SEpSet       mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp);
  SCatalog*    pCatalog = NULL;
  int32_t      code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog);
  if (TSDB_CODE_SUCCESS == code) {
    code = catalogGetQnodeList(pCatalog, pRequest->pTscObj->pAppInfo->pTransporter, &mgmtEpSet, pNodeList);
  }
X
Xiaoyu Wang 已提交
246 247
  if (TSDB_CODE_SUCCESS == code) {
    code = qCreateQueryPlan(&cxt, pPlan, pNodeList);
X
Xiaoyu Wang 已提交
248 249
  }
  return code;
X
Xiaoyu Wang 已提交
250 251
}

252
void setResSchemaInfo(SReqResultInfo* pResInfo, const SSchema* pSchema, int32_t numOfCols) {
253
  ASSERT(pSchema != NULL && numOfCols > 0);
254

255
  pResInfo->numOfCols = numOfCols;
L
Liu Jicong 已提交
256 257 258 259 260 261
  if (pResInfo->fields != NULL) {
    taosMemoryFree(pResInfo->fields);
  }
  if (pResInfo->userFields != NULL) {
    taosMemoryFree(pResInfo->userFields);
  }
262 263
  pResInfo->fields = taosMemoryCalloc(numOfCols, sizeof(TAOS_FIELD));
  pResInfo->userFields = taosMemoryCalloc(numOfCols, sizeof(TAOS_FIELD));
264 265

  for (int32_t i = 0; i < pResInfo->numOfCols; ++i) {
266
    pResInfo->fields[i].bytes = pSchema[i].bytes;
L
Liu Jicong 已提交
267
    pResInfo->fields[i].type = pSchema[i].type;
268 269

    pResInfo->userFields[i].bytes = pSchema[i].bytes;
L
Liu Jicong 已提交
270
    pResInfo->userFields[i].type = pSchema[i].type;
271 272 273

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

278
    tstrncpy(pResInfo->fields[i].name, pSchema[i].name, tListLen(pResInfo->fields[i].name));
279
    tstrncpy(pResInfo->userFields[i].name, pSchema[i].name, tListLen(pResInfo->userFields[i].name));
280
  }
X
Xiaoyu Wang 已提交
281 282
}

283
void setResPrecision(SReqResultInfo* pResInfo, int32_t precision) {
dengyihao's avatar
dengyihao 已提交
284 285
  if (precision != TSDB_TIME_PRECISION_MILLI && precision != TSDB_TIME_PRECISION_MICRO &&
      precision != TSDB_TIME_PRECISION_NANO) {
286 287 288 289 290 291
    return;
  }

  pResInfo->precision = precision;
}

D
dapan 已提交
292
int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList, void** pRes) {
293
  void* pTransporter = pRequest->pTscObj->pAppInfo->pTransporter;
294

D
dapan1121 已提交
295
  SQueryResult res = {.code = 0, .numOfRows = 0};
L
Liu Jicong 已提交
296
  int32_t      code = schedulerExecJob(pTransporter, pNodeList, pDag, &pRequest->body.queryJob, pRequest->sqlstr,
297
                                       pRequest->metric.start, &res);
D
dapan1121 已提交
298
  if (code != TSDB_CODE_SUCCESS) {
D
dapan1121 已提交
299 300
    if (pRequest->body.queryJob != 0) {
      schedulerFreeJob(pRequest->body.queryJob);
301 302
    }

D
dapan1121 已提交
303 304
    *pRes = res.res;

D
dapan1121 已提交
305
    pRequest->code = code;
D
dapan1121 已提交
306
    terrno = code;
H
Haojun Liao 已提交
307
    return pRequest->code;
X
Xiaoyu Wang 已提交
308
  }
309

310
  if (TDMT_VND_SUBMIT == pRequest->type || TDMT_VND_CREATE_TABLE == pRequest->type) {
D
dapan1121 已提交
311
    pRequest->body.resInfo.numOfRows = res.numOfRows;
dengyihao's avatar
test  
dengyihao 已提交
312

D
dapan1121 已提交
313 314
    if (pRequest->body.queryJob != 0) {
      schedulerFreeJob(pRequest->body.queryJob);
D
dapan1121 已提交
315 316
    }
  }
D
dapan1121 已提交
317

D
dapan1121 已提交
318
  *pRes = res.res;
D
dapan 已提交
319

D
dapan1121 已提交
320
  pRequest->code = res.code;
L
Liu Jicong 已提交
321
  terrno = res.code;
D
dapan1121 已提交
322
  return pRequest->code;
X
Xiaoyu Wang 已提交
323
}
X
Xiaoyu Wang 已提交
324

D
dapan1121 已提交
325 326 327 328 329
int32_t getQueryPlan(SRequestObj* pRequest, SQuery* pQuery, SArray** pNodeList) {
  *pNodeList = taosArrayInit(4, sizeof(struct SQueryNodeAddr));
  return getPlan(pRequest, pQuery, &pRequest->body.pDag, *pNodeList);
}

D
dapan1121 已提交
330 331 332
int32_t validateSversion(SRequestObj* pRequest, void* res) {
  SArray* pArray = NULL;
  int32_t code = 0;
dengyihao's avatar
dengyihao 已提交
333

D
dapan1121 已提交
334 335 336 337 338 339 340 341 342 343 344
  if (TDMT_VND_SUBMIT == pRequest->type) {
    SSubmitRsp* pRsp = (SSubmitRsp*)res;
    if (pRsp->nBlocks <= 0) {
      return TSDB_CODE_SUCCESS;
    }

    pArray = taosArrayInit(pRsp->nBlocks, sizeof(STbSVersion));
    if (NULL == pArray) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      return TSDB_CODE_OUT_OF_MEMORY;
    }
dengyihao's avatar
dengyihao 已提交
345

D
dapan1121 已提交
346
    for (int32_t i = 0; i < pRsp->nBlocks; ++i) {
dengyihao's avatar
dengyihao 已提交
347
      SSubmitBlkRsp* blk = pRsp->pBlocks + i;
D
dapan1121 已提交
348 349 350 351
      if (NULL == blk->tblFName || 0 == blk->tblFName[0]) {
        continue;
      }
      
dengyihao's avatar
dengyihao 已提交
352
      STbSVersion    tbSver = {.tbFName = blk->tblFName, .sver = blk->sver};
D
dapan1121 已提交
353 354 355
      taosArrayPush(pArray, &tbSver);
    }
  } else if (TDMT_VND_QUERY == pRequest->type) {
356 357 358 359 360 361 362 363 364 365 366
    SArray* pTbArray = (SArray*)res;
    int32_t tbNum = taosArrayGetSize(pTbArray);
    if (tbNum <= 0) {
      return TSDB_CODE_SUCCESS;
    }

    pArray = taosArrayInit(tbNum, sizeof(STbSVersion));
    if (NULL == pArray) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      return TSDB_CODE_OUT_OF_MEMORY;
    }
D
dapan1121 已提交
367

368 369 370 371 372
    for (int32_t i = 0; i < tbNum; ++i) {
      STbVerInfo* tbInfo = taosArrayGet(pTbArray, i);
      STbSVersion tbSver = {.tbFName = tbInfo->tbFName, .sver = tbInfo->sversion};
      taosArrayPush(pArray, &tbSver);
    }
D
dapan1121 已提交
373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388
  }

  SCatalog* pCatalog = NULL;
  CHECK_CODE_GOTO(catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog), _return);

  SEpSet epset = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp);

  code = catalogChkTbMetaVersion(pCatalog, pRequest->pTscObj->pAppInfo->pTransporter, &epset, pArray);

_return:

  taosArrayDestroy(pArray);

  return code;
}

D
dapan1121 已提交
389
void freeRequestRes(SRequestObj* pRequest, void* res) {
D
dapan1121 已提交
390
  if (NULL == pRequest || NULL == res) {
D
dapan1121 已提交
391 392
    return;
  }
dengyihao's avatar
dengyihao 已提交
393

D
dapan1121 已提交
394 395 396
  if (TDMT_VND_SUBMIT == pRequest->type) {
    tFreeSSubmitRsp((SSubmitRsp*)res);
  } else if (TDMT_VND_QUERY == pRequest->type) {
L
Liu Jicong 已提交
397
    taosArrayDestroy((SArray*)res);
D
dapan1121 已提交
398 399
  }
}
D
dapan1121 已提交
400

D
dapan 已提交
401
SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, int32_t code, bool keepQuery, void** res) {
D
dapan1121 已提交
402 403
  void* pRes = NULL;

X
Xiaoyu Wang 已提交
404 405 406 407 408 409 410 411
  if (TSDB_CODE_SUCCESS == code) {
    switch (pQuery->execMode) {
      case QUERY_EXEC_MODE_LOCAL:
        code = execLocalCmd(pRequest, pQuery);
        break;
      case QUERY_EXEC_MODE_RPC:
        code = execDdlQuery(pRequest, pQuery);
        break;
X
Xiaoyu Wang 已提交
412 413
      case QUERY_EXEC_MODE_SCHEDULE: {
        SArray* pNodeList = taosArrayInit(4, sizeof(struct SQueryNodeAddr));
X
Xiaoyu Wang 已提交
414 415
        code = getPlan(pRequest, pQuery, &pRequest->body.pDag, pNodeList);
        if (TSDB_CODE_SUCCESS == code) {
D
dapan1121 已提交
416 417 418 419
          code = scheduleQuery(pRequest, pRequest->body.pDag, pNodeList, &pRes);
          if (NULL != pRes) {
            code = validateSversion(pRequest, pRes);
          }
X
Xiaoyu Wang 已提交
420
        }
X
Xiaoyu Wang 已提交
421
        taosArrayDestroy(pNodeList);
X
Xiaoyu Wang 已提交
422
        break;
X
Xiaoyu Wang 已提交
423
      }
X
Xiaoyu Wang 已提交
424 425 426 427 428 429
      case QUERY_EXEC_MODE_EMPTY_RESULT:
        pRequest->type = TSDB_SQL_RETRIEVE_EMPTY_RESULT;
        break;
      default:
        break;
    }
X
Xiaoyu Wang 已提交
430 431
  }

D
stmt  
dapan1121 已提交
432 433 434
  if (!keepQuery) {
    qDestroyQuery(pQuery);
  }
dengyihao's avatar
dengyihao 已提交
435

D
dapan1121 已提交
436
  if (NULL != pRequest && TSDB_CODE_SUCCESS != code) {
X
Xiaoyu Wang 已提交
437
    pRequest->code = terrno;
D
dapan1121 已提交
438 439 440 441
  }

  if (res) {
    *res = pRes;
D
dapan1121 已提交
442 443 444
  } else {
    freeRequestRes(pRequest, pRes);
    pRes = NULL;
X
Xiaoyu Wang 已提交
445
  }
446

447 448 449
  return pRequest;
}

D
stmt  
dapan1121 已提交
450 451 452 453 454
SRequestObj* launchQuery(STscObj* pTscObj, const char* sql, int sqlLen) {
  SRequestObj* pRequest = NULL;
  SQuery*      pQuery = NULL;

  int32_t code = buildRequest(pTscObj, sql, sqlLen, &pRequest);
455 456 457 458 459 460 461
  if (code != TSDB_CODE_SUCCESS) {
    terrno = code;
    return NULL;
  }

  code = parseSql(pRequest, false, &pQuery, NULL);
  if (code != TSDB_CODE_SUCCESS) {
462 463
    pRequest->code = code;
    return pRequest;
D
stmt  
dapan1121 已提交
464 465
  }

D
dapan 已提交
466
  return launchQueryImpl(pRequest, pQuery, code, false, NULL);
D
stmt  
dapan1121 已提交
467 468
}

D
dapan1121 已提交
469
int32_t refreshMeta(STscObj* pTscObj, SRequestObj* pRequest) {
L
Liu Jicong 已提交
470 471 472 473
  SCatalog* pCatalog = NULL;
  int32_t   code = 0;
  int32_t   dbNum = taosArrayGetSize(pRequest->dbList);
  int32_t   tblNum = taosArrayGetSize(pRequest->tableList);
D
dapan1121 已提交
474 475 476 477

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

D
dapan1121 已提交
479 480 481 482 483 484 485 486
  code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCatalog);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

  SEpSet epset = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);

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

D
dapan1121 已提交
489 490 491
    code = catalogRefreshDBVgInfo(pCatalog, pTscObj->pAppInfo->pTransporter, &epset, dbFName);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
D
dapan1121 已提交
492 493 494
    }
  }

D
dapan1121 已提交
495
  for (int32_t i = 0; i < tblNum; ++i) {
L
Liu Jicong 已提交
496
    SName* tableName = taosArrayGet(pRequest->tableList, i);
D
dapan1121 已提交
497 498 499 500 501

    code = catalogRefreshTableMeta(pCatalog, pTscObj->pAppInfo->pTransporter, &epset, tableName, -1);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }
D
dapan1121 已提交
502 503
  }

D
dapan1121 已提交
504
  return code;
D
dapan1121 已提交
505 506
}

D
dapan1121 已提交
507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523
int32_t removeMeta(STscObj* pTscObj, SArray* tbList) {
  SCatalog* pCatalog = NULL;
  int32_t tbNum = taosArrayGetSize(tbList);
  int32_t code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCatalog);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }
  
  for (int32_t i = 0; i < tbNum; ++i) {
    SName* pTbName = taosArrayGet(tbList, i);
    catalogRemoveTableMeta(pCatalog, pTbName);
  }

  return TSDB_CODE_SUCCESS;
}


D
dapan1121 已提交
524 525
SRequestObj* execQuery(STscObj* pTscObj, const char* sql, int sqlLen) {
  SRequestObj* pRequest = NULL;
L
Liu Jicong 已提交
526 527
  int32_t      retryNum = 0;
  int32_t      code = 0;
D
dapan1121 已提交
528

529 530
  do {
    destroyRequest(pRequest);
D
stmt  
dapan1121 已提交
531
    pRequest = launchQuery(pTscObj, sql, sqlLen);
532
    if (pRequest == NULL || TSDB_CODE_SUCCESS == pRequest->code || !NEED_CLIENT_HANDLE_ERROR(pRequest->code)) {
D
dapan1121 已提交
533 534 535
      break;
    }

D
dapan1121 已提交
536 537 538
    code = refreshMeta(pTscObj, pRequest);
    if (code) {
      pRequest->code = code;
D
dapan1121 已提交
539 540
      break;
    }
541
  } while (retryNum++ < REQUEST_MAX_TRY_TIMES);
L
Liu Jicong 已提交
542

D
dapan1121 已提交
543 544 545 546
  if (NEED_CLIENT_RM_TBLMETA_REQ(pRequest->type)) {
    removeMeta(pTscObj, pRequest->tableList);
  }
  
D
dapan1121 已提交
547 548 549 550 551 552 553 554 555 556 557 558 559 560
  return pRequest;
}

TAOS_RES* taos_query_l(TAOS* taos, const char* sql, int sqlLen) {
  STscObj* pTscObj = (STscObj*)taos;
  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;
    return NULL;
  }

  return execQuery(pTscObj, sql, sqlLen);
}

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

H
Haojun Liao 已提交
564
  // init mnode ip set
dengyihao's avatar
dengyihao 已提交
565
  SEpSet* mgmtEpSet = &(pEpSet->epSet);
566
  mgmtEpSet->numOfEps = 0;
dengyihao's avatar
dengyihao 已提交
567
  mgmtEpSet->inUse = 0;
568

S
Shengliang Guan 已提交
569 570 571 572
  if (firstEp && firstEp[0] != 0) {
    if (strlen(firstEp) >= TSDB_EP_LEN) {
      terrno = TSDB_CODE_TSC_INVALID_FQDN;
      return -1;
S
Shengliang Guan 已提交
573
    }
S
Shengliang Guan 已提交
574

575 576 577 578 579 580
    int32_t code = taosGetFqdnPortFromEp(firstEp, &mgmtEpSet->eps[0]);
    if (code != TSDB_CODE_SUCCESS) {
      terrno = TSDB_CODE_TSC_INVALID_FQDN;
      return terrno;
    }

S
Shengliang Guan 已提交
581 582 583 584 585 586 587
    mgmtEpSet->numOfEps++;
  }

  if (secondEp && secondEp[0] != 0) {
    if (strlen(secondEp) >= TSDB_EP_LEN) {
      terrno = TSDB_CODE_TSC_INVALID_FQDN;
      return -1;
588
    }
S
Shengliang Guan 已提交
589 590 591

    taosGetFqdnPortFromEp(secondEp, &mgmtEpSet->eps[mgmtEpSet->numOfEps]);
    mgmtEpSet->numOfEps++;
592 593 594 595 596 597 598 599 600 601
  }

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

  return 0;
}

dengyihao's avatar
dengyihao 已提交
602
STscObj* taosConnectImpl(const char* user, const char* auth, const char* db, __taos_async_fn_t fp, void* param,
L
Liu Jicong 已提交
603
                         SAppInstInfo* pAppInfo, int connType) {
604
  STscObj* pTscObj = createTscObj(user, auth, db, connType, pAppInfo);
605 606 607 608 609
  if (NULL == pTscObj) {
    terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
    return pTscObj;
  }

dengyihao's avatar
dengyihao 已提交
610
  SRequestObj* pRequest = createRequest(pTscObj, fp, param, TDMT_MND_CONNECT);
611 612 613
  if (pRequest == NULL) {
    destroyTscObj(pTscObj);
    terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
614 615 616
    return NULL;
  }

617
  SMsgSendInfo* body = buildConnectMsg(pRequest);
618 619

  int64_t transporterId = 0;
H
Haojun Liao 已提交
620
  asyncSendMsgToServer(pTscObj->pAppInfo->pTransporter, &pTscObj->pAppInfo->mgmtEp.epSet, &transporterId, body);
621 622 623

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

628
    terrno = pRequest->code;
629 630 631 632
    destroyRequest(pRequest);
    taos_close(pTscObj);
    pTscObj = NULL;
  } else {
D
dapan1121 已提交
633
    tscDebug("0x%" PRIx64 " connection is opening, connId:%u, dnodeConn:%p, reqId:0x%" PRIx64, pTscObj->id,
dengyihao's avatar
dengyihao 已提交
634
             pTscObj->connId, pTscObj->pAppInfo->pTransporter, pRequest->requestId);
635 636 637 638 639 640
    destroyRequest(pRequest);
  }

  return pTscObj;
}

641
static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest) {
wafwerar's avatar
wafwerar 已提交
642
  SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
643 644 645 646 647
  if (pMsgSendInfo == NULL) {
    terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
    return NULL;
  }

dengyihao's avatar
dengyihao 已提交
648
  pMsgSendInfo->msgType = TDMT_MND_CONNECT;
649

650
  pMsgSendInfo->requestObjRefId = pRequest->self;
dengyihao's avatar
dengyihao 已提交
651 652 653
  pMsgSendInfo->requestId = pRequest->requestId;
  pMsgSendInfo->fp = handleRequestRspFp[TMSG_INDEX(pMsgSendInfo->msgType)];
  pMsgSendInfo->param = pRequest;
654

S
Shengliang Guan 已提交
655 656
  SConnectReq connectReq = {0};
  STscObj*    pObj = pRequest->pTscObj;
657

H
Haojun Liao 已提交
658
  char* db = getDbOfConnection(pObj);
659
  if (db != NULL) {
S
Shengliang Guan 已提交
660
    tstrncpy(connectReq.db, db, sizeof(connectReq.db));
661
  }
wafwerar's avatar
wafwerar 已提交
662
  taosMemoryFreeClear(db);
663

X
Xiaoyu Wang 已提交
664 665
  connectReq.connType = pObj->connType;
  connectReq.pid = htonl(appInfo.pid);
S
Shengliang Guan 已提交
666
  connectReq.startTime = htobe64(appInfo.startTime);
667

S
Shengliang Guan 已提交
668
  tstrncpy(connectReq.app, appInfo.appName, sizeof(connectReq.app));
dengyihao's avatar
dengyihao 已提交
669 670
  tstrncpy(connectReq.user, pObj->user, sizeof(connectReq.user));
  tstrncpy(connectReq.passwd, pObj->pass, sizeof(connectReq.passwd));
S
Shengliang Guan 已提交
671 672

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

S
Shengliang Guan 已提交
676 677
  pMsgSendInfo->msgInfo.len = contLen;
  pMsgSendInfo->msgInfo.pData = pReq;
678
  return pMsgSendInfo;
679 680
}

681
static void destroySendMsgInfo(SMsgSendInfo* pMsgBody) {
H
Haojun Liao 已提交
682
  assert(pMsgBody != NULL);
wafwerar's avatar
wafwerar 已提交
683 684
  taosMemoryFreeClear(pMsgBody->msgInfo.pData);
  taosMemoryFreeClear(pMsgBody);
H
Haojun Liao 已提交
685
}
686

dengyihao's avatar
dengyihao 已提交
687
bool persistConnForSpecificMsg(void* parenct, tmsg_t msgType) {
L
Liu Jicong 已提交
688 689
  return msgType == TDMT_VND_QUERY_RSP || msgType == TDMT_VND_FETCH_RSP || msgType == TDMT_VND_RES_READY_RSP ||
         msgType == TDMT_VND_QUERY_HEARTBEAT_RSP;
dengyihao's avatar
dengyihao 已提交
690
}
691

692
void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) {
S
Shengliang Guan 已提交
693 694
  SMsgSendInfo* pSendInfo = (SMsgSendInfo*)pMsg->info.ahandle;
  assert(pMsg->info.ahandle != NULL);
695

696
  if (pSendInfo->requestObjRefId != 0) {
dengyihao's avatar
dengyihao 已提交
697
    SRequestObj* pRequest = (SRequestObj*)taosAcquireRef(clientReqRefPool, pSendInfo->requestObjRefId);
698
    assert(pRequest->self == pSendInfo->requestObjRefId);
699

D
dapan1121 已提交
700
    pRequest->metric.rsp = taosGetTimestampUs();
701

dengyihao's avatar
dengyihao 已提交
702
    STscObj* pTscObj = pRequest->pTscObj;
703 704 705 706
    if (pEpSet) {
      if (!isEpsetEqual(&pTscObj->pAppInfo->mgmtEp.epSet, pEpSet)) {
        updateEpSet_s(&pTscObj->pAppInfo->mgmtEp, pEpSet);
      }
707 708
    }

709
    /*
dengyihao's avatar
dengyihao 已提交
710 711
     * There is not response callback function for submit response.
     * The actual inserted number of points is the first number.
712
     */
713
    int32_t elapsed = pRequest->metric.rsp - pRequest->metric.start;
714
    if (pMsg->code == TSDB_CODE_SUCCESS) {
dengyihao's avatar
dengyihao 已提交
715
      tscDebug("0x%" PRIx64 " message:%s, code:%s rspLen:%d, elapsed:%d ms, reqId:0x%" PRIx64, pRequest->self,
L
Liu Jicong 已提交
716
               TMSG_INFO(pMsg->msgType), tstrerror(pMsg->code), pMsg->contLen, elapsed / 1000, pRequest->requestId);
717
    } else {
dengyihao's avatar
dengyihao 已提交
718
      tscError("0x%" PRIx64 " SQL cmd:%s, code:%s rspLen:%d, elapsed time:%d ms, reqId:0x%" PRIx64, pRequest->self,
L
Liu Jicong 已提交
719
               TMSG_INFO(pMsg->msgType), tstrerror(pMsg->code), pMsg->contLen, elapsed / 1000, pRequest->requestId);
720
    }
721

H
Haojun Liao 已提交
722
    taosReleaseRef(clientReqRefPool, pSendInfo->requestObjRefId);
723 724
  }

S
Shengliang Guan 已提交
725
  SDataBuf buf = {.len = pMsg->contLen, .pData = NULL, .handle = pMsg->info.handle};
726 727

  if (pMsg->contLen > 0) {
wafwerar's avatar
wafwerar 已提交
728
    buf.pData = taosMemoryCalloc(1, pMsg->contLen);
729 730 731 732 733 734
    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);
    }
735 736
  }

737
  pSendInfo->fp(pSendInfo->param, &buf, pMsg->code);
738
  rpcFreeCont(pMsg->pCont);
739
  destroySendMsgInfo(pSendInfo);
740
}
741

dengyihao's avatar
dengyihao 已提交
742
TAOS* taos_connect_auth(const char* ip, const char* user, const char* auth, const char* db, uint16_t port) {
743 744 745 746 747 748 749 750 751 752
  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;
  }

L
Liu Jicong 已提交
753
  return taos_connect_internal(ip, user, NULL, auth, db, port, CONN_TYPE__QUERY);
754 755
}

dengyihao's avatar
dengyihao 已提交
756 757 758
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};
759
  char dbStr[TSDB_DB_NAME_LEN] = {0};
dengyihao's avatar
dengyihao 已提交
760 761
  char userStr[TSDB_USER_LEN] = {0};
  char passStr[TSDB_PASSWORD_LEN] = {0};
762

dengyihao's avatar
dengyihao 已提交
763
  strncpy(ipStr, ip, TMIN(TSDB_EP_LEN - 1, ipLen));
dengyihao's avatar
dengyihao 已提交
764 765
  strncpy(userStr, user, TMIN(TSDB_USER_LEN - 1, userLen));
  strncpy(passStr, pass, TMIN(TSDB_PASSWORD_LEN - 1, passLen));
dengyihao's avatar
dengyihao 已提交
766
  strncpy(dbStr, db, TMIN(TSDB_DB_NAME_LEN - 1, dbLen));
767
  return taos_connect(ipStr, userStr, passStr, dbStr, port);
768 769
}

L
Liu Jicong 已提交
770
void doSetOneRowPtr(SReqResultInfo* pResultInfo) {
771 772 773 774 775 776 777
  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)) {
778
      if (pCol->offset[pResultInfo->current] != -1) {
779 780 781 782 783 784
        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;
785
        pResultInfo->length[i] = 0;
786 787 788 789
      }
    } else {
      if (!colDataIsNull_f(pCol->nullbitmap, pResultInfo->current)) {
        pResultInfo->row[i] = pResultInfo->pCol[i].pData + bytes * pResultInfo->current;
790
        pResultInfo->length[i] = bytes;
791 792
      } else {
        pResultInfo->row[i] = NULL;
793
        pResultInfo->length[i] = 0;
794 795 796 797 798
      }
    }
  }
}

799
void* doFetchRows(SRequestObj* pRequest, bool setupOneRowPtr, bool convertUcs4) {
800
  assert(pRequest != NULL);
H
Haojun Liao 已提交
801

H
Haojun Liao 已提交
802
  SReqResultInfo* pResultInfo = &pRequest->body.resInfo;
H
Haojun Liao 已提交
803
  if (pResultInfo->pData == NULL || pResultInfo->current >= pResultInfo->numOfRows) {
H
Haojun Liao 已提交
804
    // All data has returned to App already, no need to try again
805 806 807 808 809
    if (pResultInfo->completed) {
      pResultInfo->numOfRows = 0;
      return NULL;
    }

H
Haojun Liao 已提交
810 811 812 813 814 815
    SReqResultInfo* pResInfo = &pRequest->body.resInfo;
    pRequest->code = schedulerFetchRows(pRequest->body.queryJob, (void**)&pResInfo->pData);
    if (pRequest->code != TSDB_CODE_SUCCESS) {
      pResultInfo->numOfRows = 0;
      return NULL;
    }
H
Haojun Liao 已提交
816

H
Haojun Liao 已提交
817 818 819 820 821
    pRequest->code = setQueryResultFromRsp(&pRequest->body.resInfo, (SRetrieveTableRsp*)pResInfo->pData, convertUcs4);
    if (pRequest->code != TSDB_CODE_SUCCESS) {
      pResultInfo->numOfRows = 0;
      return NULL;
    }
H
Haojun Liao 已提交
822

H
Haojun Liao 已提交
823 824
    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);
H
Haojun Liao 已提交
825

H
Haojun Liao 已提交
826
    if (pResultInfo->numOfRows == 0) {
H
Haojun Liao 已提交
827 828 829 830
      return NULL;
    }
  }

831 832 833
  if (setupOneRowPtr) {
    doSetOneRowPtr(pResultInfo);
    pResultInfo->current += 1;
H
Haojun Liao 已提交
834 835 836 837 838
  }

  return pResultInfo->row;
}

839
static int32_t doPrepareResPtr(SReqResultInfo* pResInfo) {
H
Haojun Liao 已提交
840
  if (pResInfo->row == NULL) {
L
Liu Jicong 已提交
841 842
    pResInfo->row = taosMemoryCalloc(pResInfo->numOfCols, POINTER_BYTES);
    pResInfo->pCol = taosMemoryCalloc(pResInfo->numOfCols, sizeof(SResultColumn));
wafwerar's avatar
wafwerar 已提交
843 844
    pResInfo->length = taosMemoryCalloc(pResInfo->numOfCols, sizeof(int32_t));
    pResInfo->convertBuf = taosMemoryCalloc(pResInfo->numOfCols, POINTER_BYTES);
845

846 847 848
    if (pResInfo->row == NULL || pResInfo->pCol == NULL || pResInfo->length == NULL || pResInfo->convertBuf == NULL) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }
849
  }
850 851

  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
852 853
}

854 855 856 857
static char* parseTagDatatoJson(void* p) {
  char*  string = NULL;
  cJSON* json = cJSON_CreateObject();
  if (json == NULL) {
wmmhello's avatar
wmmhello 已提交
858 859 860 861
    goto end;
  }

  int16_t nCols = kvRowNCols(p);
862
  char    tagJsonKey[256] = {0};
wmmhello's avatar
wmmhello 已提交
863
  for (int j = 0; j < nCols; ++j) {
864 865 866 867
    SColIdx* pColIdx = kvRowColIdxAt(p, j);
    char*    val = (char*)(kvRowColVal(p, pColIdx));
    if (j == 0) {
      if (*val == TSDB_DATA_TYPE_NULL) {
wmmhello's avatar
wmmhello 已提交
868
        string = taosMemoryCalloc(1, 8);
wmmhello's avatar
wmmhello 已提交
869
        sprintf(string, "%s", TSDB_DATA_NULL_STR_L);
wmmhello's avatar
wmmhello 已提交
870 871 872 873 874 875 876 877 878 879 880
        goto end;
      }
      continue;
    }

    // json key  encode by binary
    memset(tagJsonKey, 0, sizeof(tagJsonKey));
    memcpy(tagJsonKey, varDataVal(val), varDataLen(val));
    // json value
    val += varDataTLen(val);
    char* realData = POINTER_SHIFT(val, CHAR_BYTES);
881 882
    char  type = *val;
    if (type == TSDB_DATA_TYPE_NULL) {
wmmhello's avatar
wmmhello 已提交
883
      cJSON* value = cJSON_CreateNull();
884
      if (value == NULL) {
wmmhello's avatar
wmmhello 已提交
885 886 887
        goto end;
      }
      cJSON_AddItemToObject(json, tagJsonKey, value);
888
    } else if (type == TSDB_DATA_TYPE_NCHAR) {
wmmhello's avatar
wmmhello 已提交
889
      cJSON* value = NULL;
890 891 892
      if (varDataLen(realData) > 0) {
        char*   tagJsonValue = taosMemoryCalloc(varDataLen(realData), 1);
        int32_t length = taosUcs4ToMbs((TdUcs4*)varDataVal(realData), varDataLen(realData), tagJsonValue);
wmmhello's avatar
wmmhello 已提交
893
        if (length < 0) {
wmmhello's avatar
wmmhello 已提交
894
          tscError("charset:%s to %s. val:%s convert json value failed.", DEFAULT_UNICODE_ENCODEC, tsCharset, val);
wmmhello's avatar
wmmhello 已提交
895 896 897 898 899
          taosMemoryFree(tagJsonValue);
          goto end;
        }
        value = cJSON_CreateString(tagJsonValue);
        taosMemoryFree(tagJsonValue);
900
        if (value == NULL) {
wmmhello's avatar
wmmhello 已提交
901 902
          goto end;
        }
903
      } else if (varDataLen(realData) == 0) {
wmmhello's avatar
wmmhello 已提交
904
        value = cJSON_CreateString("");
905
      } else {
wmmhello's avatar
wmmhello 已提交
906 907 908 909
        ASSERT(0);
      }

      cJSON_AddItemToObject(json, tagJsonKey, value);
910
    } else if (type == TSDB_DATA_TYPE_DOUBLE) {
wmmhello's avatar
wmmhello 已提交
911 912
      double jsonVd = *(double*)(realData);
      cJSON* value = cJSON_CreateNumber(jsonVd);
913
      if (value == NULL) {
wmmhello's avatar
wmmhello 已提交
914 915 916
        goto end;
      }
      cJSON_AddItemToObject(json, tagJsonKey, value);
917 918 919 920 921 922 923 924 925 926
      //    }else if(type == TSDB_DATA_TYPE_BIGINT){
      //      int64_t jsonVd = *(int64_t*)(realData);
      //      cJSON* value = cJSON_CreateNumber((double)jsonVd);
      //      if (value == NULL)
      //      {
      //        goto end;
      //      }
      //      cJSON_AddItemToObject(json, tagJsonKey, value);
    } else if (type == TSDB_DATA_TYPE_BOOL) {
      char   jsonVd = *(char*)(realData);
wmmhello's avatar
wmmhello 已提交
927
      cJSON* value = cJSON_CreateBool(jsonVd);
928
      if (value == NULL) {
wmmhello's avatar
wmmhello 已提交
929 930 931
        goto end;
      }
      cJSON_AddItemToObject(json, tagJsonKey, value);
932
    } else {
wmmhello's avatar
wmmhello 已提交
933
      ASSERT(0);
wmmhello's avatar
wmmhello 已提交
934 935 936 937 938 939 940 941
    }
  }
  string = cJSON_PrintUnformatted(json);
end:
  cJSON_Delete(json);
  return string;
}

942 943 944 945 946
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;

947
    if (type == TSDB_DATA_TYPE_NCHAR && colLength[i] > 0) {
948 949 950 951 952 953 954 955 956 957 958 959 960 961
      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 已提交
962
          ASSERT((p + len) < (pResultInfo->convertBuf[i] + colLength[i]));
dengyihao's avatar
dengyihao 已提交
963

964 965 966 967 968 969 970 971
          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;
972
    } else if (type == TSDB_DATA_TYPE_JSON && colLength[i] > 0) {
973 974 975 976 977 978
      char* p = taosMemoryRealloc(pResultInfo->convertBuf[i], colLength[i]);
      if (p == NULL) {
        return TSDB_CODE_OUT_OF_MEMORY;
      }

      pResultInfo->convertBuf[i] = p;
dengyihao's avatar
dengyihao 已提交
979
      int32_t        len = 0;
980 981 982 983 984 985
      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 jsonInnerType = *pStart;
dengyihao's avatar
dengyihao 已提交
986 987 988
          char*   jsonInnerData = pStart + CHAR_BYTES;
          char    dst[TSDB_MAX_JSON_TAG_LEN] = {0};
          if (jsonInnerType == TSDB_DATA_TYPE_NULL) {
989 990
            sprintf(varDataVal(dst), "%s", TSDB_DATA_NULL_STR_L);
            varDataSetLen(dst, strlen(varDataVal(dst)));
dengyihao's avatar
dengyihao 已提交
991
          } else if (jsonInnerType == TSDB_DATA_TYPE_JSON) {
992
            char* jsonString = parseTagDatatoJson(jsonInnerData);
wmmhello's avatar
wmmhello 已提交
993 994
            STR_TO_VARSTR(dst, jsonString);
            taosMemoryFree(jsonString);
dengyihao's avatar
dengyihao 已提交
995
          } else if (jsonInnerType == TSDB_DATA_TYPE_NCHAR) {  // value -> "value"
996
            *(char*)varDataVal(dst) = '\"';
dengyihao's avatar
dengyihao 已提交
997 998
            int32_t length = taosUcs4ToMbs((TdUcs4*)varDataVal(jsonInnerData), varDataLen(jsonInnerData),
                                           varDataVal(dst) + CHAR_BYTES);
999
            if (length <= 0) {
dengyihao's avatar
dengyihao 已提交
1000 1001
              tscError("charset:%s to %s. val:%s convert failed.", DEFAULT_UNICODE_ENCODEC, tsCharset,
                       varDataVal(jsonInnerData));
1002 1003
              length = 0;
            }
dengyihao's avatar
dengyihao 已提交
1004
            varDataSetLen(dst, length + CHAR_BYTES * 2);
wmmhello's avatar
wmmhello 已提交
1005
            *(char*)POINTER_SHIFT(varDataVal(dst), length + CHAR_BYTES) = '\"';
dengyihao's avatar
dengyihao 已提交
1006
          } else if (jsonInnerType == TSDB_DATA_TYPE_DOUBLE) {
1007 1008 1009
            double jsonVd = *(double*)(jsonInnerData);
            sprintf(varDataVal(dst), "%.9lf", jsonVd);
            varDataSetLen(dst, strlen(varDataVal(dst)));
dengyihao's avatar
dengyihao 已提交
1010
          } else if (jsonInnerType == TSDB_DATA_TYPE_BIGINT) {
1011 1012 1013
            int64_t jsonVd = *(int64_t*)(jsonInnerData);
            sprintf(varDataVal(dst), "%" PRId64, jsonVd);
            varDataSetLen(dst, strlen(varDataVal(dst)));
dengyihao's avatar
dengyihao 已提交
1014 1015
          } else if (jsonInnerType == TSDB_DATA_TYPE_BOOL) {
            sprintf(varDataVal(dst), "%s", (*((char*)jsonInnerData) == 1) ? "true" : "false");
1016
            varDataSetLen(dst, strlen(varDataVal(dst)));
dengyihao's avatar
dengyihao 已提交
1017
          } else {
1018 1019 1020
            ASSERT(0);
          }

dengyihao's avatar
dengyihao 已提交
1021
          if (len + varDataTLen(dst) > colLength[i]) {
1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038
            p = taosMemoryRealloc(pResultInfo->convertBuf[i], len + varDataTLen(dst));
            if (p == NULL) {
              return TSDB_CODE_OUT_OF_MEMORY;
            }

            pResultInfo->convertBuf[i] = p;
          }
          p = pResultInfo->convertBuf[i] + len;
          memcpy(p, dst, varDataTLen(dst));
          pCol->offset[j] = len;
          len += varDataTLen(dst);
        }
      }

      pResultInfo->pCol[i].pData = pResultInfo->convertBuf[i];
      pResultInfo->row[i] = pResultInfo->pCol[i].pData;
    }
1039 1040 1041 1042 1043
  }

  return TSDB_CODE_SUCCESS;
}

L
Liu Jicong 已提交
1044 1045
int32_t setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32_t numOfCols, int32_t numOfRows,
                         bool convertUcs4) {
H
Haojun Liao 已提交
1046 1047
  assert(numOfCols > 0 && pFields != NULL && pResultInfo != NULL);
  if (numOfRows == 0) {
1048
    return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
1049 1050
  }

1051 1052 1053 1054
  int32_t code = doPrepareResPtr(pResultInfo);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }
H
Haojun Liao 已提交
1055

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

L
Liu Jicong 已提交
1058
  int32_t dataLen = *(int32_t*)p;
1059 1060
  p += sizeof(int32_t);

L
Liu Jicong 已提交
1061
  uint64_t groupId = *(uint64_t*)p;
1062 1063 1064 1065 1066 1067
  p += sizeof(uint64_t);

  int32_t* colLength = (int32_t*)p;
  p += sizeof(int32_t) * numOfCols;

  char* pStart = p;
H
Haojun Liao 已提交
1068
  for (int32_t i = 0; i < numOfCols; ++i) {
1069
    colLength[i] = htonl(colLength[i]);
1070
    ASSERT(colLength[i] < dataLen);
1071 1072 1073 1074 1075 1076 1077 1078 1079 1080

    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 已提交
1081
    pResultInfo->length[i] = pResultInfo->fields[i].bytes;
1082 1083 1084
    pResultInfo->row[i] = pResultInfo->pCol[i].pData;

    pStart += colLength[i];
1085
  }
1086

H
Haojun Liao 已提交
1087
  // convert UCS4-LE encoded character to native multi-bytes character in current data block.
1088 1089
  if (convertUcs4) {
    code = doConvertUCS4(pResultInfo, numOfRows, numOfCols, colLength);
1090 1091
  }

1092
  return code;
S
Shengliang Guan 已提交
1093 1094
}

H
Haojun Liao 已提交
1095
char* getDbOfConnection(STscObj* pObj) {
dengyihao's avatar
dengyihao 已提交
1096
  char* p = NULL;
wafwerar's avatar
wafwerar 已提交
1097
  taosThreadMutexLock(&pObj->mutex);
1098 1099 1100 1101
  size_t len = strlen(pObj->db);
  if (len > 0) {
    p = strndup(pObj->db, tListLen(pObj->db));
  }
S
Shengliang Guan 已提交
1102

wafwerar's avatar
wafwerar 已提交
1103
  taosThreadMutexUnlock(&pObj->mutex);
1104 1105 1106 1107 1108
  return p;
}

void setConnectionDB(STscObj* pTscObj, const char* db) {
  assert(db != NULL && pTscObj != NULL);
wafwerar's avatar
wafwerar 已提交
1109
  taosThreadMutexLock(&pTscObj->mutex);
1110
  tstrncpy(pTscObj->db, db, tListLen(pTscObj->db));
wafwerar's avatar
wafwerar 已提交
1111
  taosThreadMutexUnlock(&pTscObj->mutex);
1112
}
S
Shengliang Guan 已提交
1113

H
Haojun Liao 已提交
1114 1115 1116 1117 1118 1119 1120 1121 1122 1123
void resetConnectDB(STscObj* pTscObj) {
  if (pTscObj == NULL) {
    return;
  }

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

1124
int32_t setQueryResultFromRsp(SReqResultInfo* pResultInfo, const SRetrieveTableRsp* pRsp, bool convertUcs4) {
1125 1126
  assert(pResultInfo != NULL && pRsp != NULL);

L
Liu Jicong 已提交
1127 1128 1129 1130 1131
  pResultInfo->pRspMsg = (const char*)pRsp;
  pResultInfo->pData = (void*)pRsp->data;
  pResultInfo->numOfRows = htonl(pRsp->numOfRows);
  pResultInfo->current = 0;
  pResultInfo->completed = (pRsp->completed == 1);
1132
  pResultInfo->payloadLen = htonl(pRsp->compLen);
L
Liu Jicong 已提交
1133
  pResultInfo->precision = pRsp->precision;
H
Haojun Liao 已提交
1134

1135
  // TODO handle the compressed case
1136
  pResultInfo->totalRows += pResultInfo->numOfRows;
1137 1138
  return setResultDataPtr(pResultInfo, pResultInfo->fields, pResultInfo->numOfCols, pResultInfo->numOfRows,
                          convertUcs4);
L
fix  
Liu Jicong 已提交
1139
}
S
Shengliang Guan 已提交
1140 1141 1142 1143 1144 1145

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 已提交
1146
  SRpcMsg            rpcMsg = {.info.ahandle = (void*)0x9526, .msgType = TDMT_DND_SERVER_STATUS};
S
Shengliang Guan 已提交
1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164
  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 已提交
1165 1166 1167 1168 1169 1170 1171 1172
  if (fqdn == NULL) {
    fqdn = tsLocalFqdn;
  }

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

S
Shengliang Guan 已提交
1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187
  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;
S
Shengliang Guan 已提交
1188
  if (details != NULL && statusRsp.details != NULL) {
S
Shengliang Guan 已提交
1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200
    tstrncpy(details, statusRsp.details, maxlen);
  }

_OVER:
  if (clientRpc != NULL) {
    rpcClose(clientRpc);
  }
  if (rpcRsp.pCont != NULL) {
    rpcFreeCont(rpcRsp.pCont);
  }
  return code;
}