clientInt.h 12.1 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22
/*
 * 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/>.
 */

#ifndef TDENGINE_CLIENTINT_H
#define TDENGINE_CLIENTINT_H

#ifdef __cplusplus
extern "C" {
#endif

L
Liu Jicong 已提交
23
#include "catalog.h"
dengyihao's avatar
dengyihao 已提交
24
#include "parser.h"
25
#include "planner.h"
dengyihao's avatar
dengyihao 已提交
26 27
#include "query.h"
#include "taos.h"
L
Liu Jicong 已提交
28
#include "tcommon.h"
H
Haojun Liao 已提交
29
#include "tdatablock.h"
30
#include "tdef.h"
31 32
#include "thash.h"
#include "tlist.h"
dengyihao's avatar
dengyihao 已提交
33
#include "tmsg.h"
34
#include "tmsgtype.h"
35
#include "trpc.h"
36

S
Shengliang Guan 已提交
37
#include "tconfig.h"
S
Shengliang Guan 已提交
38

39 40
#define CHECK_CODE_GOTO(expr, label) \
  do {                               \
D
dapan1121 已提交
41
    code = expr;                     \
42 43 44 45
    if (TSDB_CODE_SUCCESS != code) { \
      goto label;                    \
    }                                \
  } while (0)
46

47
#define ERROR_MSG_BUF_DEFAULT_SIZE 512
L
Liu Jicong 已提交
48
#define HEARTBEAT_INTERVAL         1500  // ms
49
#define SYNC_ON_TOP_OF_ASYNC       1
50

51 52 53
enum {
  RES_TYPE__QUERY = 1,
  RES_TYPE__TMQ,
L
Liu Jicong 已提交
54
  RES_TYPE__TMQ_META,
55 56
};

L
Liu Jicong 已提交
57
#define SHOW_VARIABLES_RESULT_COLS       2
D
dapan1121 已提交
58 59 60
#define SHOW_VARIABLES_RESULT_FIELD1_LEN (TSDB_CONFIG_OPTION_LEN + VARSTR_HEADER_SIZE)
#define SHOW_VARIABLES_RESULT_FIELD2_LEN (TSDB_CONFIG_VALUE_LEN + VARSTR_HEADER_SIZE)

L
Liu Jicong 已提交
61 62 63
#define TD_RES_QUERY(res)    (*(int8_t*)res == RES_TYPE__QUERY)
#define TD_RES_TMQ(res)      (*(int8_t*)res == RES_TYPE__TMQ)
#define TD_RES_TMQ_META(res) (*(int8_t*)res == RES_TYPE__TMQ_META)
64

L
Liu Jicong 已提交
65 66
typedef struct SAppInstInfo SAppInstInfo;

L
Liu Jicong 已提交
67
typedef struct {
dengyihao's avatar
dengyihao 已提交
68
  char* key;
L
Liu Jicong 已提交
69 70 71 72 73 74
  // statistics
  int32_t reportCnt;
  int32_t connKeyCnt;
  int64_t reportBytes;  // not implemented
  int64_t startTime;
  // ctl
L
Liu Jicong 已提交
75
  SRWLatch      lock;  // lock is used in serialization
L
Liu Jicong 已提交
76
  SAppInstInfo* pAppInstInfo;
L
Liu Jicong 已提交
77
  SHashObj*     activeInfo;  // hash<SClientHbKey, SClientHbReq>
L
Liu Jicong 已提交
78 79
} SAppHbMgr;

L
Liu Jicong 已提交
80
typedef int32_t (*FHbRspHandle)(SAppHbMgr* pAppHbMgr, SClientHbRsp* pRsp);
D
dapan1121 已提交
81

dengyihao's avatar
dengyihao 已提交
82
typedef int32_t (*FHbReqHandle)(SClientHbKey* connKey, void* param, SClientHbReq* req);
D
dapan1121 已提交
83

L
Liu Jicong 已提交
84
typedef struct {
L
Liu Jicong 已提交
85 86
  int8_t  inited;
  int64_t appId;
L
Liu Jicong 已提交
87
  // ctl
L
Liu Jicong 已提交
88 89
  int8_t        threadStop;
  TdThread      thread;
L
Liu Jicong 已提交
90 91
  TdThreadMutex lock;  // used when app init and cleanup
  SHashObj*     appSummary;
L
Liu Jicong 已提交
92
  SArray*       appHbMgrs;  // SArray<SAppHbMgr*> one for each cluster
D
dapan1121 已提交
93 94
  FHbReqHandle  reqHandle[CONN_TYPE__MAX];
  FHbRspHandle  rspHandle[CONN_TYPE__MAX];
L
Liu Jicong 已提交
95 96
} SClientHbMgr;

97
typedef struct SQueryExecMetric {
D
dapan1121 已提交
98 99 100 101
  int64_t start;   // start timestamp, us
  int64_t parsed;  // start to parse, us
  int64_t send;    // start to send to server, us
  int64_t rsp;     // receive response from server, us
102 103 104
} SQueryExecMetric;

typedef struct SHeartBeatInfo {
dengyihao's avatar
dengyihao 已提交
105
  void* pTimer;  // timer, used to send request msg to mnode
106 107
} SHeartBeatInfo;

L
Liu Jicong 已提交
108
struct SAppInstInfo {
D
dapan1121 已提交
109 110
  int64_t            numOfConns;
  SCorEpSet          mgmtEp;
D
dapan1121 已提交
111 112
  int32_t            totalDnodes;
  int32_t            onlineDnodes;
D
dapan1121 已提交
113 114 115 116 117 118 119
  TdThreadMutex      qnodeMutex;
  SArray*            pQnodeList;
  SAppClusterSummary summary;
  SList*             pConnList;  // STscObj linked list
  uint64_t           clusterId;
  void*              pTransporter;
  SAppHbMgr*         pAppHbMgr;
D
dapan1121 已提交
120
  char*              instKey;
L
Liu Jicong 已提交
121
};
122 123

typedef struct SAppInfo {
L
Liu Jicong 已提交
124 125 126 127 128 129
  int64_t       startTime;
  char          appName[TSDB_APP_NAME_LEN];
  char*         ep;
  int32_t       pid;
  int32_t       numOfThreads;
  SHashObj*     pInstMap;
wafwerar's avatar
wafwerar 已提交
130
  TdThreadMutex mutex;
131 132 133
} SAppInfo;

typedef struct STscObj {
L
Liu Jicong 已提交
134 135 136
  char          user[TSDB_USER_LEN];
  char          pass[TSDB_PASSWORD_LEN];
  char          db[TSDB_DB_FNAME_LEN];
D
dapan1121 已提交
137 138
  char          sVer[TSDB_VERSION_LEN];
  char          sDetailVer[128];
L
Liu Jicong 已提交
139
  int8_t        connType;
L
Liu Jicong 已提交
140 141
  int32_t       acctId;
  uint32_t      connId;
D
dapan1121 已提交
142
  int64_t       id;         // ref ID returned by taosAddRef
L
Liu Jicong 已提交
143 144 145
  TdThreadMutex mutex;      // used to protect the operation on db
  int32_t       numOfReqs;  // number of sqlObj bound to this connection
  SAppInstInfo* pAppInfo;
146
  SHashObj*     pRequests;
147
  int8_t        schemalessType;  // todo remove it, this attribute should be move to request
148 149
} STscObj;

150 151
typedef struct SResultColumn {
  union {
L
Liu Jicong 已提交
152 153
    char*    nullbitmap;  // bitmap, one bit for each item in the list
    int32_t* offset;
154
  };
L
Liu Jicong 已提交
155
  char* pData;
156 157
} SResultColumn;

158
typedef struct SReqResultInfo {
D
dapan1121 已提交
159
  SExecResult    execRes;
160 161
  const char*    pRspMsg;
  const char*    pData;
L
Liu Jicong 已提交
162 163
  TAOS_FIELD*    fields;      // todo, column names are not needed.
  TAOS_FIELD*    userFields;  // the fields info that return to user
164 165
  uint32_t       numOfCols;
  int32_t*       length;
166
  char**         convertBuf;
167 168 169 170 171 172
  TAOS_ROW       row;
  SResultColumn* pCol;
  uint32_t       numOfRows;
  uint64_t       totalRows;
  uint32_t       current;
  bool           completed;
H
Haojun Liao 已提交
173
  int32_t        precision;
174
  bool           convertUcs4;
175
  int32_t        payloadLen;
wmmhello's avatar
wmmhello 已提交
176
  char*          convertJson;
177 178 179
} SReqResultInfo;

typedef struct SRequestSendRecvBody {
dengyihao's avatar
dengyihao 已提交
180 181 182 183 184 185 186 187
  tsem_t            rspSem;  // not used now
  __taos_async_fn_t queryFp;
  __taos_async_fn_t fetchFp;
  void*             param;
  SDataBuf          requestMsg;
  int64_t           queryJob;  // query job, created according to sql query DAG.
  int32_t           subplanNum;
  SReqResultInfo    resInfo;
188
} SRequestSendRecvBody;
189

L
Liu Jicong 已提交
190
typedef struct {
L
Liu Jicong 已提交
191 192
  int8_t         resType;
  char           topic[TSDB_TOPIC_FNAME_LEN];
L
Liu Jicong 已提交
193
  char           db[TSDB_DB_FNAME_LEN];
L
Liu Jicong 已提交
194 195 196
  int32_t        vgId;
  SSchemaWrapper schema;
  int32_t        resIter;
L
Liu Jicong 已提交
197
  SMqDataRsp     rsp;
L
Liu Jicong 已提交
198
  SReqResultInfo resInfo;
L
Liu Jicong 已提交
199
} SMqRspObj;
L
Liu Jicong 已提交
200

L
Liu Jicong 已提交
201 202 203 204 205 206 207 208
typedef struct {
  int8_t     resType;
  char       topic[TSDB_TOPIC_FNAME_LEN];
  char       db[TSDB_DB_FNAME_LEN];
  int32_t    vgId;
  SMqMetaRsp metaRsp;
} SMqMetaRspObj;

209
typedef struct SRequestObj {
L
Liu Jicong 已提交
210
  int8_t               resType;  // query or tmq
dengyihao's avatar
dengyihao 已提交
211 212 213
  uint64_t             requestId;
  int32_t              type;  // request type
  STscObj*             pTscObj;
214
  char*                pDb;     // current database string
dengyihao's avatar
dengyihao 已提交
215 216 217 218
  char*                sqlstr;  // sql string
  int32_t              sqlLen;
  int64_t              self;
  char*                msgBuf;
D
stmt  
dapan1121 已提交
219
  int32_t              msgBufLen;
dengyihao's avatar
dengyihao 已提交
220
  int32_t              code;
D
dapan1121 已提交
221 222
  SArray*              dbList;
  SArray*              tableList;
dengyihao's avatar
dengyihao 已提交
223
  SQueryExecMetric     metric;
H
Haojun Liao 已提交
224
  SRequestSendRecvBody body;
225
  bool                 stableQuery;
D
dapan1121 已提交
226
  bool                 validateOnly;
227

L
Liu Jicong 已提交
228 229 230
  bool     killed;
  uint32_t prevCode;  // previous error code: todo refactor, add update flag for catalog
  uint32_t retry;
231 232
} SRequestObj;

233 234 235 236 237
typedef struct SSyncQueryParam {
  tsem_t       sem;
  SRequestObj* pRequest;
} SSyncQueryParam;

L
Liu Jicong 已提交
238 239
void* doAsyncFetchRows(SRequestObj* pRequest, bool setupOneRowPtr, bool convertUcs4);
void* doFetchRows(SRequestObj* pRequest, bool setupOneRowPtr, bool convertUcs4);
240

L
Liu Jicong 已提交
241 242 243 244 245 246 247 248
void    doSetOneRowPtr(SReqResultInfo* pResultInfo);
void    setResPrecision(SReqResultInfo* pResInfo, int32_t precision);
int32_t setQueryResultFromRsp(SReqResultInfo* pResultInfo, const SRetrieveTableRsp* pRsp, bool convertUcs4,
                              bool freeAfterUse);
void    setResSchemaInfo(SReqResultInfo* pResInfo, const SSchema* pSchema, int32_t numOfCols);
void    doFreeReqResultInfo(SReqResultInfo* pResInfo);
int32_t transferTableNameList(const char* tbList, int32_t acctId, char* dbName, SArray** pReq);
void    syncCatalogFn(SMetaData* pResult, void* param, int32_t code);
D
dapan1121 已提交
249 250

SRequestObj* execQuery(STscObj* pTscObj, const char* sql, int sqlLen, bool validateOnly);
L
Liu Jicong 已提交
251 252
TAOS_RES*    taosQueryImpl(TAOS* taos, const char* sql, bool validateOnly);
void         taosAsyncQueryImpl(TAOS* taos, const char* sql, __taos_async_fn_t fp, void* param, bool validateOnly);
L
Liu Jicong 已提交
253

L
Liu Jicong 已提交
254
static FORCE_INLINE SReqResultInfo* tmqGetCurResInfo(TAOS_RES* res) {
L
Liu Jicong 已提交
255
  SMqRspObj* msg = (SMqRspObj*)res;
L
Liu Jicong 已提交
256
  return (SReqResultInfo*)&msg->resInfo;
L
Liu Jicong 已提交
257 258
}

L
Liu Jicong 已提交
259
static FORCE_INLINE SReqResultInfo* tmqGetNextResInfo(TAOS_RES* res, bool convertUcs4) {
L
Liu Jicong 已提交
260
  SMqRspObj* msg = (SMqRspObj*)res;
L
Liu Jicong 已提交
261 262 263
  msg->resIter++;
  if (msg->resIter < msg->rsp.blockNum) {
    SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)taosArrayGetP(msg->rsp.blockData, msg->resIter);
L
Liu Jicong 已提交
264 265 266
    if (msg->rsp.withSchema) {
      SSchemaWrapper* pSW = (SSchemaWrapper*)taosArrayGetP(msg->rsp.blockSchema, msg->resIter);
      setResSchemaInfo(&msg->resInfo, pSW->pSchema, pSW->nCols);
L
Liu Jicong 已提交
267 268 269 270
      taosMemoryFreeClear(msg->resInfo.row);
      taosMemoryFreeClear(msg->resInfo.pCol);
      taosMemoryFreeClear(msg->resInfo.length);
      taosMemoryFreeClear(msg->resInfo.convertBuf);
wmmhello's avatar
wmmhello 已提交
271
      taosMemoryFreeClear(msg->resInfo.convertJson);
L
Liu Jicong 已提交
272
    }
L
Liu Jicong 已提交
273
    setQueryResultFromRsp(&msg->resInfo, pRetrieve, convertUcs4, false);
L
Liu Jicong 已提交
274
    return &msg->resInfo;
L
Liu Jicong 已提交
275 276 277 278 279 280 281 282 283
  }
  return NULL;
}

static FORCE_INLINE SReqResultInfo* tscGetCurResInfo(TAOS_RES* res) {
  if (TD_RES_QUERY(res)) return &(((SRequestObj*)res)->body.resInfo);
  return tmqGetCurResInfo(res);
}

284
extern SAppInfo appInfo;
H
Haojun Liao 已提交
285
extern int32_t  clientReqRefPool;
286
extern int32_t  clientConnRefPool;
dengyihao's avatar
dengyihao 已提交
287
extern void*    tscQhandle;
288

H
Haojun Liao 已提交
289 290
__async_send_cb_fn_t getMsgRspHandle(int32_t msgType);

H
Haojun Liao 已提交
291
SMsgSendInfo* buildMsgInfoImpl(SRequestObj* pReqObj);
292

293
void*    createTscObj(const char* user, const char* auth, const char* db, int32_t connType, SAppInstInfo* pAppInfo);
L
Liu Jicong 已提交
294 295 296
void     destroyTscObj(void* pObj);
STscObj* acquireTscObj(int64_t rid);
int32_t  releaseTscObj(int64_t rid);
297

298 299
uint64_t generateRequestId();

300
void*        createRequest(STscObj* pObj, int32_t type);
L
Liu Jicong 已提交
301 302 303
void         destroyRequest(SRequestObj* pRequest);
SRequestObj* acquireRequest(int64_t rid);
int32_t      releaseRequest(int64_t rid);
D
dapan1121 已提交
304
int32_t      removeRequest(int64_t rid);
dengyihao's avatar
dengyihao 已提交
305
void         doDestroyRequest(void* p);
306

dengyihao's avatar
dengyihao 已提交
307
char* getDbOfConnection(STscObj* pObj);
308
void  setConnectionDB(STscObj* pTscObj, const char* db);
H
Haojun Liao 已提交
309
void  resetConnectDB(STscObj* pTscObj);
310

L
Liu Jicong 已提交
311
int taos_options_imp(TSDB_OPTION option, const char* str);
312

dengyihao's avatar
dengyihao 已提交
313
void* openTransporter(const char* user, const char* auth, int32_t numOfThreads);
314

dengyihao's avatar
dengyihao 已提交
315
bool persistConnForSpecificMsg(void* parenct, tmsg_t msgType);
316
void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet);
317

318
STscObj* taos_connect_internal(const char* ip, const char* user, const char* pass, const char* auth, const char* db,
L
Liu Jicong 已提交
319
                               uint16_t port, int connType);
320

D
dapan1121 已提交
321
SRequestObj* launchQuery(STscObj* pTscObj, const char* sql, int sqlLen, bool validateOnly);
322

D
stmt  
dapan1121 已提交
323 324
int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery, SStmtCallback* pStmtCb);

D
dapan1121 已提交
325
int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArray* pNodeList);
326

dengyihao's avatar
dengyihao 已提交
327
int32_t buildRequest(STscObj* pTscObj, const char* sql, int sqlLen, SRequestObj** pRequest);
328

L
Liu Jicong 已提交
329
void taos_close_internal(void* taos);
330

dengyihao's avatar
dengyihao 已提交
331
// --- heartbeat
L
Liu Jicong 已提交
332 333 334 335 336 337
// global, called by mgmt
int  hbMgrInit();
void hbMgrCleanUp();
int  hbHandleRsp(SClientHbBatchRsp* hbRsp);

// cluster level
dengyihao's avatar
dengyihao 已提交
338 339
SAppHbMgr* appHbMgrInit(SAppInstInfo* pAppInstInfo, char* key);
void       appHbMgrCleanup(void);
dengyihao's avatar
dengyihao 已提交
340
void       hbRemoveAppHbMrg(SAppHbMgr** pAppHbMgr);
dengyihao's avatar
dengyihao 已提交
341 342
void       destroyAllRequests(SHashObj* pRequests);
void       stopAllRequests(SHashObj* pRequests);
L
Liu Jicong 已提交
343 344

// conn level
L
Liu Jicong 已提交
345
int  hbRegisterConn(SAppHbMgr* pAppHbMgr, int64_t tscRefId, int64_t clusterId, int8_t connType);
L
Liu Jicong 已提交
346 347 348 349 350
void hbDeregisterConn(SAppHbMgr* pAppHbMgr, SClientHbKey connKey);

// --- mq
void hbMgrInitMqHbRspHandle();

351
SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, bool keepQuery, void** res);
H
Haojun Liao 已提交
352
int32_t      scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList);
L
Liu Jicong 已提交
353
void         launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultMeta);
L
Liu Jicong 已提交
354
int32_t      refreshMeta(STscObj* pTscObj, SRequestObj* pRequest);
L
Liu Jicong 已提交
355
int32_t      updateQnodeList(SAppInstInfo* pInfo, SArray* pNodeList);
356
void         doAsyncQuery(SRequestObj* pRequest, bool forceUpdateMeta);
L
Liu Jicong 已提交
357 358
int32_t      removeMeta(STscObj* pTscObj, SArray* tbList);  // todo move to clientImpl.c and become a static function
int32_t      handleAlterTbExecRes(void* res, struct SCatalog* pCatalog);  // todo move to xxx
D
dapan1121 已提交
359
bool         qnodeRequired(SRequestObj* pRequest);
D
stmt  
dapan1121 已提交
360

dengyihao's avatar
dengyihao 已提交
361 362 363
void initTscQhandle();
void cleanupTscQhandle();

364 365 366 367 368
#ifdef __cplusplus
}
#endif

#endif  // TDENGINE_CLIENTINT_H