clientInt.h 10.8 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

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

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

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

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

50 51 52 53 54 55 56 57
enum {
  RES_TYPE__QUERY = 1,
  RES_TYPE__TMQ,
};

#define TD_RES_QUERY(res) (*(int8_t*)res == RES_TYPE__QUERY)
#define TD_RES_TMQ(res)   (*(int8_t*)res == RES_TYPE__TMQ)

L
Liu Jicong 已提交
58 59
typedef struct SAppInstInfo SAppInstInfo;

L
Liu Jicong 已提交
60
typedef struct {
dengyihao's avatar
dengyihao 已提交
61
  char* key;
L
Liu Jicong 已提交
62 63 64 65 66 67 68 69
  // statistics
  int32_t reportCnt;
  int32_t connKeyCnt;
  int64_t reportBytes;  // not implemented
  int64_t startTime;
  // ctl
  SRWLatch lock;  // lock is used in serialization
  SAppInstInfo* pAppInstInfo;
dengyihao's avatar
dengyihao 已提交
70
  SHashObj* activeInfo;  // hash<SClientHbKey, SClientHbReq>
L
Liu Jicong 已提交
71 72
} SAppHbMgr;

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

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

L
Liu Jicong 已提交
77
typedef struct {
L
Liu Jicong 已提交
78 79
  int8_t inited;
  // ctl
L
Liu Jicong 已提交
80 81
  int8_t        threadStop;
  TdThread      thread;
wafwerar's avatar
wafwerar 已提交
82
  TdThreadMutex lock;       // used when app init and cleanup
L
Liu Jicong 已提交
83
  SArray*       appHbMgrs;  // SArray<SAppHbMgr*> one for each cluster
D
dapan1121 已提交
84 85
  FHbReqHandle  reqHandle[CONN_TYPE__MAX];
  FHbRspHandle  rspHandle[CONN_TYPE__MAX];
L
Liu Jicong 已提交
86 87
} SClientHbMgr;

88
typedef struct SQueryExecMetric {
D
dapan1121 已提交
89 90 91 92
  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
93 94
} SQueryExecMetric;

95
typedef struct SInstanceSummary {
dengyihao's avatar
dengyihao 已提交
96 97 98 99 100 101 102 103 104 105
  uint64_t numOfInsertsReq;
  uint64_t numOfInsertRows;
  uint64_t insertElapsedTime;
  uint64_t insertBytes;  // submit to tsdb since launched.

  uint64_t fetchBytes;
  uint64_t queryElapsedTime;
  uint64_t numOfSlowQueries;
  uint64_t totalRequests;
  uint64_t currentRequests;  // the number of SRequestObj
106
} SInstanceSummary;
107 108

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

L
Liu Jicong 已提交
112
struct SAppInstInfo {
L
Liu Jicong 已提交
113 114
  int64_t          numOfConns;
  SCorEpSet        mgmtEp;
D
dapan1121 已提交
115 116
  TdThreadMutex    qnodeMutex;
  SArray*          pQnodeList;
L
Liu Jicong 已提交
117 118
  SInstanceSummary summary;
  SList*           pConnList;  // STscObj linked list
119
  uint64_t         clusterId;
L
Liu Jicong 已提交
120 121
  void*            pTransporter;
  SAppHbMgr*       pAppHbMgr;
L
Liu Jicong 已提交
122
};
123 124

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

typedef struct STscObj {
L
Liu Jicong 已提交
135 136 137 138
  char          user[TSDB_USER_LEN];
  char          pass[TSDB_PASSWORD_LEN];
  char          db[TSDB_DB_FNAME_LEN];
  char          ver[128];
L
Liu Jicong 已提交
139
  int8_t        connType;
L
Liu Jicong 已提交
140 141 142 143 144 145
  int32_t       acctId;
  uint32_t      connId;
  uint64_t      id;         // ref ID returned by taosAddRef
  TdThreadMutex mutex;      // used to protect the operation on db
  int32_t       numOfReqs;  // number of sqlObj bound to this connection
  SAppInstInfo* pAppInfo;
146
  SHashObj*     pRequests;
wmmhello's avatar
wmmhello 已提交
147
  int8_t        schemalessType;
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
  SQueryExecRes  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
  int32_t        payloadLen;
175 176 177
} SReqResultInfo;

typedef struct SRequestSendRecvBody {
178
  tsem_t             rspSem;  // not used now
179 180 181
  __taos_async_fn_t  queryFp;
  __taos_async_fn_t  fetchFp;
  void*              param;
182 183
  SDataBuf           requestMsg;
  int64_t            queryJob;  // query job, created according to sql query DAG.
L
Liu Jicong 已提交
184
  struct SQueryPlan* pDag;      // the query dag, generated according to the sql statement.
185
  SReqResultInfo     resInfo;
186
} SRequestSendRecvBody;
187

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

199
typedef struct SRequestObj {
L
Liu Jicong 已提交
200
  int8_t               resType;  // query or tmq
dengyihao's avatar
dengyihao 已提交
201 202 203
  uint64_t             requestId;
  int32_t              type;  // request type
  STscObj*             pTscObj;
204
  char*                pDb;     // current database string
dengyihao's avatar
dengyihao 已提交
205 206 207 208
  char*                sqlstr;  // sql string
  int32_t              sqlLen;
  int64_t              self;
  char*                msgBuf;
D
stmt  
dapan1121 已提交
209
  int32_t              msgBufLen;
dengyihao's avatar
dengyihao 已提交
210
  int32_t              code;
D
dapan1121 已提交
211 212
  SArray*              dbList;
  SArray*              tableList;
dengyihao's avatar
dengyihao 已提交
213
  SQueryExecMetric     metric;
H
Haojun Liao 已提交
214
  SRequestSendRecvBody body;
215 216 217

  uint32_t             prevCode; //previous error code: todo refactor, add update flag for catalog
  uint32_t             retry;
218 219
} SRequestObj;

220 221 222 223 224 225
typedef struct SSyncQueryParam {
  tsem_t       sem;
  SRequestObj* pRequest;
} SSyncQueryParam;

void*   doAsyncFetchRow(SRequestObj* pRequest, bool setupOneRowPtr, bool convertUcs4);
L
Liu Jicong 已提交
226
void*   doFetchRows(SRequestObj* pRequest, bool setupOneRowPtr, bool convertUcs4);
227

L
Liu Jicong 已提交
228 229
void    doSetOneRowPtr(SReqResultInfo* pResultInfo);
void    setResPrecision(SReqResultInfo* pResInfo, int32_t precision);
L
Liu Jicong 已提交
230 231
int32_t setQueryResultFromRsp(SReqResultInfo* pResultInfo, const SRetrieveTableRsp* pRsp, bool convertUcs4,
                              bool freeAfterUse);
L
Liu Jicong 已提交
232
void    setResSchemaInfo(SReqResultInfo* pResInfo, const SSchema* pSchema, int32_t numOfCols);
L
Liu Jicong 已提交
233
void    doFreeReqResultInfo(SReqResultInfo* pResInfo);
234
SRequestObj* execQuery(STscObj* pTscObj, const char* sql, int sqlLen);
L
Liu Jicong 已提交
235

L
Liu Jicong 已提交
236
static FORCE_INLINE SReqResultInfo* tmqGetCurResInfo(TAOS_RES* res) {
L
Liu Jicong 已提交
237
  SMqRspObj* msg = (SMqRspObj*)res;
L
Liu Jicong 已提交
238
  return (SReqResultInfo*)&msg->resInfo;
L
Liu Jicong 已提交
239 240
}

L
Liu Jicong 已提交
241
static FORCE_INLINE SReqResultInfo* tmqGetNextResInfo(TAOS_RES* res, bool convertUcs4) {
L
Liu Jicong 已提交
242
  SMqRspObj* msg = (SMqRspObj*)res;
L
Liu Jicong 已提交
243 244 245
  msg->resIter++;
  if (msg->resIter < msg->rsp.blockNum) {
    SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)taosArrayGetP(msg->rsp.blockData, msg->resIter);
L
Liu Jicong 已提交
246 247 248
    if (msg->rsp.withSchema) {
      SSchemaWrapper* pSW = (SSchemaWrapper*)taosArrayGetP(msg->rsp.blockSchema, msg->resIter);
      setResSchemaInfo(&msg->resInfo, pSW->pSchema, pSW->nCols);
L
Liu Jicong 已提交
249 250 251 252
      taosMemoryFreeClear(msg->resInfo.row);
      taosMemoryFreeClear(msg->resInfo.pCol);
      taosMemoryFreeClear(msg->resInfo.length);
      taosMemoryFreeClear(msg->resInfo.convertBuf);
L
Liu Jicong 已提交
253
    }
L
Liu Jicong 已提交
254
    setQueryResultFromRsp(&msg->resInfo, pRetrieve, convertUcs4, false);
L
Liu Jicong 已提交
255
    return &msg->resInfo;
L
Liu Jicong 已提交
256 257 258 259 260 261 262 263 264
  }
  return NULL;
}

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

265
extern SAppInfo appInfo;
H
Haojun Liao 已提交
266
extern int32_t  clientReqRefPool;
267
extern int32_t  clientConnRefPool;
268

H
Haojun Liao 已提交
269 270
__async_send_cb_fn_t getMsgRspHandle(int32_t msgType);

H
Haojun Liao 已提交
271
SMsgSendInfo* buildMsgInfoImpl(SRequestObj* pReqObj);
272

273
void*    createTscObj(const char* user, const char* auth, const char* db, int32_t connType, SAppInstInfo* pAppInfo);
L
Liu Jicong 已提交
274 275 276
void     destroyTscObj(void* pObj);
STscObj* acquireTscObj(int64_t rid);
int32_t  releaseTscObj(int64_t rid);
277

278 279
uint64_t generateRequestId();

280
void*        createRequest(STscObj* pObj, int32_t type);
L
Liu Jicong 已提交
281 282 283
void         destroyRequest(SRequestObj* pRequest);
SRequestObj* acquireRequest(int64_t rid);
int32_t      releaseRequest(int64_t rid);
284

dengyihao's avatar
dengyihao 已提交
285
char* getDbOfConnection(STscObj* pObj);
286
void  setConnectionDB(STscObj* pTscObj, const char* db);
H
Haojun Liao 已提交
287
void  resetConnectDB(STscObj* pTscObj);
288

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

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

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

dengyihao's avatar
dengyihao 已提交
296
TAOS* taos_connect_internal(const char* ip, const char* user, const char* pass, const char* auth, const char* db,
L
Liu Jicong 已提交
297
                            uint16_t port, int connType);
298

299 300
SRequestObj* launchQuery(STscObj* pTscObj, const char* sql, int sqlLen);

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

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

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

dengyihao's avatar
dengyihao 已提交
307
// --- heartbeat
L
Liu Jicong 已提交
308 309 310 311 312 313
// global, called by mgmt
int  hbMgrInit();
void hbMgrCleanUp();
int  hbHandleRsp(SClientHbBatchRsp* hbRsp);

// cluster level
dengyihao's avatar
dengyihao 已提交
314 315
SAppHbMgr* appHbMgrInit(SAppInstInfo* pAppInstInfo, char* key);
void       appHbMgrCleanup(void);
L
Liu Jicong 已提交
316 317

// conn level
L
Liu Jicong 已提交
318
int  hbRegisterConn(SAppHbMgr* pAppHbMgr, int64_t tscRefId, int64_t clusterId, int8_t connType);
L
Liu Jicong 已提交
319 320 321 322 323
void hbDeregisterConn(SAppHbMgr* pAppHbMgr, SClientHbKey connKey);

// --- mq
void hbMgrInitMqHbRspHandle();

324
SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, bool keepQuery, void** res);
H
Haojun Liao 已提交
325
int32_t      scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList);
326
void         launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery);
L
Liu Jicong 已提交
327
int32_t      refreshMeta(STscObj* pTscObj, SRequestObj* pRequest);
L
Liu Jicong 已提交
328
int32_t      updateQnodeList(SAppInstInfo* pInfo, SArray* pNodeList);
329 330 331
void         doAsyncQuery(SRequestObj* pRequest, bool forceUpdateMeta);
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
stmt  
dapan1121 已提交
332

333 334 335 336 337
#ifdef __cplusplus
}
#endif

#endif  // TDENGINE_CLIENTINT_H