tsclient.h 14.0 KB
Newer Older
H
hzcheng 已提交
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_TSCLIENT_H
#define TDENGINE_TSCLIENT_H

#ifdef __cplusplus
extern "C" {
#endif

S
slguan 已提交
23
#include "os.h"
24

H
Haojun Liao 已提交
25
#include "qAggMain.h"
H
hzcheng 已提交
26
#include "taos.h"
27
#include "taosdef.h"
H
hzcheng 已提交
28
#include "taosmsg.h"
H
hjxilinx 已提交
29
#include "tarray.h"
H
Haojun Liao 已提交
30
#include "tcache.h"
H
Haojun Liao 已提交
31
#include "tglobal.h"
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
32
#include "tref.h"
H
Haojun Liao 已提交
33
#include "tutil.h"
H
hjxilinx 已提交
34

H
Haojun Liao 已提交
35
#include "qExecutor.h"
36
#include "qSqlparser.h"
H
Haojun Liao 已提交
37
#include "qTsbuf.h"
H
Haojun Liao 已提交
38
#include "qUtil.h"
H
Haojun Liao 已提交
39
#include "tcmdtype.h"
H
hzcheng 已提交
40 41 42

// forward declaration
struct SSqlInfo;
H
hjxilinx 已提交
43

H
Haojun Liao 已提交
44 45
typedef void (*__async_cb_func_t)(void *param, TAOS_RES *tres, int32_t numOfRows);

46 47 48 49 50 51 52 53 54 55 56
typedef struct SNewVgroupInfo {
  int32_t    vgId;
  int8_t     inUse;
  int8_t     numOfEps;
  SEpAddrMsg ep[TSDB_MAX_REPLICA];
} SNewVgroupInfo;

typedef struct CChildTableMeta {
  int32_t        vgId;
  STableId       id;
  uint8_t        tableType;
57 58
  char           sTableName[TSDB_TABLE_FNAME_LEN];  // TODO: refactor super table name, not full name
  uint64_t       suid;                              // super table id
59 60
} CChildTableMeta;

H
hjxilinx 已提交
61 62 63 64 65
typedef struct SColumnIndex {
  int16_t tableIndex;
  int16_t columnIndex;
} SColumnIndex;

66
typedef struct SColumn {
H
Haojun Liao 已提交
67 68 69
  uint64_t     tableUid;
  int32_t      columnIndex;
  SColumnInfo  info;
70
} SColumn;
H
hzcheng 已提交
71

H
Haojun Liao 已提交
72 73
typedef struct SInternalField {
  TAOS_FIELD      field;
H
hjxilinx 已提交
74
  bool            visible;
H
Haojun Liao 已提交
75
  SExprInfo      *pExpr;
H
Haojun Liao 已提交
76
} SInternalField;
S
slguan 已提交
77 78

typedef struct SParamInfo {
H
hjxilinx 已提交
79
  int32_t  idx;
80
  uint8_t  type;
H
hjxilinx 已提交
81
  uint8_t  timePrec;
82
  int16_t  bytes;
S
slguan 已提交
83 84 85
  uint32_t offset;
} SParamInfo;

H
Haojun Liao 已提交
86 87 88 89 90
typedef struct SBoundColumn {
  bool    hasVal;  // denote if current column has bound or not
  int32_t offset;  // all column offset value
} SBoundColumn;

C
Cary Xu 已提交
91 92 93 94 95
typedef struct {
  uint16_t schemaColIdx;
  uint16_t boundIdx;
  uint16_t finalIdx;
} SBoundIdxInfo;
96 97 98 99 100 101 102

typedef enum _COL_ORDER_STATUS {
  ORDER_STATUS_UNKNOWN = 0,
  ORDER_STATUS_ORDERED = 1,
  ORDER_STATUS_DISORDERED = 2,
} EOrderStatus;

H
Haojun Liao 已提交
103
typedef struct SParsedDataColInfo {
C
Cary Xu 已提交
104 105 106 107 108
  int16_t        numOfCols;
  int16_t        numOfBound;
  int32_t *      boundedColumns;  // bounded column idx according to schema
  SBoundColumn * cols;
  SBoundIdxInfo *colIdxInfo;
109
  int8_t         orderStatus;  // bounded columns: 
H
Haojun Liao 已提交
110 111
} SParsedDataColInfo;

112
#define IS_DATA_COL_ORDERED(s) ((s) == (int8_t)ORDER_STATUS_ORDERED)
K
kailixu 已提交
113

C
Cary Xu 已提交
114
typedef struct {
K
kailixu 已提交
115 116 117 118 119
  SSchema *   pSchema;
  int16_t     sversion;
  int32_t     flen;
  uint16_t    nCols;
  void *      buf;
C
Cary Xu 已提交
120 121 122 123
  void *      pDataBlock;
  SSubmitBlk *pSubmitBlk;
} SMemRowBuilder;

C
Cary Xu 已提交
124 125 126 127
typedef struct {
  TDRowLenT allNullLen;
} SMemRowHelper;

S
slguan 已提交
128
typedef struct STableDataBlocks {
129
  SName       tableName;
130 131 132 133 134 135 136 137 138 139
  int8_t      tsSource;     // where does the UNIX timestamp come from, server or client
  bool        ordered;      // if current rows are ordered or not
  int64_t     vgId;         // virtual group id
  int64_t     prevTS;       // previous timestamp, recorded to decide if the records array is ts ascending
  int32_t     numOfTables;  // number of tables in current submit block
  int32_t     rowSize;      // row size for current table
  uint32_t    nAllocSize;
  uint32_t    headerSize;   // header for table info (uid, tid, submit metadata)
  uint32_t    size;
  STableMeta *pTableMeta;   // the tableMeta of current table, the table meta will be used during submit, keep a ref to avoid to be removed from cache
H
Haojun Liao 已提交
140
  char       *pData;
141 142
  bool        cloned;
  
C
Cary Xu 已提交
143
  SParsedDataColInfo boundColumnInfo;
H
Haojun Liao 已提交
144

S
slguan 已提交
145
  // for parameter ('?') binding
C
Cary Xu 已提交
146 147 148
  uint32_t       numOfAllocedParams;
  uint32_t       numOfParams;
  SParamInfo *   params;
C
Cary Xu 已提交
149
  SMemRowHelper  rowHelper;
S
slguan 已提交
150
} STableDataBlocks;
H
hzcheng 已提交
151

H
Haojun Liao 已提交
152 153 154 155 156
typedef struct {
  STableMeta   *pTableMeta;
  SVgroupsInfo *pVgroupInfo;
} STableMetaVgroupInfo;

H
Haojun Liao 已提交
157 158
typedef struct SInsertStatementParam {
  SName      **pTableNameList;          // all involved tableMeta list of current insert sql statement.
H
Haojun Liao 已提交
159
  int32_t      numOfTables;             // number of tables in table name list
H
Haojun Liao 已提交
160 161 162
  SHashObj    *pTableBlockHashList;     // data block for each table
  SArray      *pDataBlocks;             // SArray<STableDataBlocks*>. Merged submit block for each vgroup
  int8_t       schemaAttached;          // denote if submit block is built with table schema or not
163
  uint8_t      payloadType;             // EPayloadType. 0: K-V payload for non-prepare insert, 1: rawPayload for prepare insert
H
Haojun Liao 已提交
164 165
  STagData     tagData;                 // NOTE: pTagData->data is used as a variant length array

H
Haojun Liao 已提交
166 167 168
  int32_t      batchSize;               // for parameter ('?') binding and batch processing
  int32_t      numOfParams;

H
Haojun Liao 已提交
169
  char         msg[512];                // error message
H
Haojun Liao 已提交
170
  uint32_t     insertType;              // insert data from [file|sql statement| bound statement]
H
Haojun Liao 已提交
171 172
  uint64_t     objectId;                // sql object id
  char        *sql;                     // current sql statement position
H
Haojun Liao 已提交
173 174
} SInsertStatementParam;

175 176 177 178 179 180 181 182
typedef enum {
  PAYLOAD_TYPE_KV = 0,
  PAYLOAD_TYPE_RAW = 1,
} EPayloadType;

#define IS_RAW_PAYLOAD(t) \
  (((int)(t)) == PAYLOAD_TYPE_RAW)  // 0: K-V payload for non-prepare insert, 1: rawPayload for prepare insert

H
Haojun Liao 已提交
183
// TODO extract sql parser supporter
H
hzcheng 已提交
184
typedef struct {
185 186
  int     command;
  uint8_t msgType;
H
Haojun Liao 已提交
187
  SInsertStatementParam insertParam;
188
  char    reserve1[3];        // fix bus error on arm32
H
Haojun Liao 已提交
189
  int32_t count;   // todo remove it
D
dapan1121 已提交
190
  bool    subCmd;
H
hzcheng 已提交
191

192
  char         reserve2[3];        // fix bus error on arm32
193
  int16_t      numOfCols;
194
  char         reserve3[2];        // fix bus error on arm32
195 196 197
  uint32_t     allocSize;
  char *       payload;
  int32_t      payloadLen;
H
Haojun Liao 已提交
198

H
Haojun Liao 已提交
199
  SHashObj    *pTableMetaMap;  // local buffer to keep the queried table meta, before validating the AST
H
Haojun Liao 已提交
200
  SQueryInfo  *pQueryInfo;
H
Haojun Liao 已提交
201 202
  SQueryInfo  *active;         // current active query info
  int32_t      batchSize;      // for parameter ('?') binding and batch processing
H
Haojun Liao 已提交
203
  int32_t      resColumnId;
H
hzcheng 已提交
204 205 206 207 208 209 210 211
} SSqlCmd;

typedef struct SResRec {
  int numOfRows;
  int numOfTotal;
} SResRec;

typedef struct {
H
Haojun Liao 已提交
212 213 214 215 216 217 218
  int32_t        numOfRows;                  // num of results in current retrieval
  int64_t        numOfRowsGroup;             // num of results of current group
  int64_t        numOfTotal;                 // num of total results
  int64_t        numOfClauseTotal;           // num of total result in current subclause
  char *         pRsp;
  int32_t        rspType;
  int32_t        rspLen;
H
Haojun Liao 已提交
219
  uint64_t       qId;
H
Haojun Liao 已提交
220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235
  int64_t        useconds;
  int64_t        offset;  // offset value from vnode during projection query of stable
  int32_t        row;
  int16_t        numOfCols;
  int16_t        precision;
  bool           completed;
  int32_t        code;
  int32_t        numOfGroups;
  SResRec *      pGroupRec;
  char *         data;
  TAOS_ROW       tsrow;
  TAOS_ROW       urow;
  int32_t*       length;  // length for each field for current row
  char **        buffer;  // Buffer used to put multibytes encoded using unicode (wchar_t)
  SColumnIndex*  pColumnIndex;

D
fix bug  
dapan1121 已提交
236
  TAOS_FIELD*           final;
237
  SArithmeticSupport   *pArithSup;   // support the arithmetic expression calculation on agg functions
H
Haojun Liao 已提交
238
  struct SGlobalMerger *pMerger;
H
hzcheng 已提交
239 240
} SSqlRes;

Y
yihaoDeng 已提交
241 242 243 244 245
typedef struct {
  char         key[512]; 
  void         *pDnodeConn; 
} SRpcObj;

H
hjxilinx 已提交
246
typedef struct STscObj {
H
hjxilinx 已提交
247 248 249 250
  void *             signature;
  void *             pTimer;
  char               user[TSDB_USER_LEN];
  char               pass[TSDB_KEY_LEN];
251 252
  char               acctId[TSDB_ACCT_ID_LEN];
  char               db[TSDB_ACCT_ID_LEN + TSDB_DB_NAME_LEN];
H
hjxilinx 已提交
253 254 255
  char               sversion[TSDB_VERSION_LEN];
  char               writeAuth : 1;
  char               superAuth : 1;
S
Shengliang Guan 已提交
256
  uint32_t           connId;
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
257
  uint64_t           rid;      // ref ID returned by taosAddRef
258
  int64_t            hbrid;
H
hjxilinx 已提交
259
  struct SSqlObj *   sqlList;
H
hjxilinx 已提交
260
  struct SSqlStream *streamList;
Y
yihaoDeng 已提交
261
  SRpcObj           *pRpcObj;
Y
yihaoDeng 已提交
262
  SRpcCorEpSet      *tscCorMgmtEpSet;
H
hjxilinx 已提交
263
  pthread_mutex_t    mutex;
264
  int32_t            numOfObj; // number of sqlObj from this tscObj
H
hzcheng 已提交
265 266
} STscObj;

H
Haojun Liao 已提交
267
typedef struct SSubqueryState {
D
fix bug  
dapan1121 已提交
268
  pthread_mutex_t mutex;
269
  int8_t  *states;
270 271
  int32_t  numOfSub;            // the number of total sub-queries
  uint64_t numOfRetrievedRows;  // total number of points in this query
H
Haojun Liao 已提交
272 273
} SSubqueryState;

274
typedef struct SSqlObj {
275
  void            *signature;
276
  int64_t          owner;        // owner of sql object, by which it is executed
277
  STscObj         *pTscObj;
陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
278
  int64_t          rpcRid;
H
Haojun Liao 已提交
279 280
  __async_cb_func_t  fp;
  __async_cb_func_t  fetchFp;
281
  void            *param;
H
hjxilinx 已提交
282 283 284 285 286
  int64_t          stime;
  uint32_t         queryId;
  void *           pStream;
  void *           pSubscription;
  char *           sqlstr;
287
  void *           pBuf;  // table meta buffer
288
  char             parseRetry;
H
hjxilinx 已提交
289 290
  char             retry;
  char             maxRetry;
291
  SRpcEpSet        epSet;
H
Haojun Liao 已提交
292
  char             listed;
H
hjxilinx 已提交
293 294 295
  tsem_t           rspSem;
  SSqlCmd          cmd;
  SSqlRes          res;
D
dapan1121 已提交
296
  bool             isBind;
297

H
Haojun Liao 已提交
298
  SSubqueryState   subState;
299
  struct SSqlObj **pSubs;
H
Haojun Liao 已提交
300

D
TD-2516  
dapan1121 已提交
301 302 303 304
  int64_t          metaRid;
  int64_t          svgroupRid;

  int64_t          squeryLock;
305
  int32_t          retryReason;  // previous error code
H
Haojun Liao 已提交
306
  struct SSqlObj  *prev, *next;
307
  int64_t          self;
H
hzcheng 已提交
308 309
} SSqlObj;

H
hjxilinx 已提交
310
typedef struct SSqlStream {
H
hzcheng 已提交
311
  SSqlObj *pSql;
312
  void *  cqhandle;  // stream belong to SCQContext handle
313
  const char* dstTable;
H
hzcheng 已提交
314 315
  uint32_t streamId;
  char     listed;
316 317
  bool     isProject;
  int16_t  precision;
H
hzcheng 已提交
318 319 320
  int64_t  num;  // number of computing count

  /*
S
slguan 已提交
321
   * keep the number of current result in computing,
H
hzcheng 已提交
322 323 324 325 326 327 328
   * the value will be set to 0 before set timer for next computing
   */
  int64_t numOfRes;

  int64_t useconds;  // total  elapsed time
  int64_t ctime;     // stream created time
  int64_t stime;     // stream next executed time
S
slguan 已提交
329
  int64_t etime;     // stream end query time, when time is larger then etime, the stream will be closed
330
  int64_t ltime;     // stream last row time in stream table
B
Bomin Zhang 已提交
331
  SInterval interval;
H
hzcheng 已提交
332 333 334 335 336
  void *  pTimer;

  void (*fp)();
  void *param;

S
slguan 已提交
337
  void (*callback)(void *);  // Callback function when stream is stopped from client level
H
hjxilinx 已提交
338
  struct SSqlStream *prev, *next;
H
hzcheng 已提交
339 340
} SSqlStream;

341 342
void tscSetStreamDestTable(SSqlStream* pStream, const char* dstTable);

Y
yihaoDeng 已提交
343
int  tscAcquireRpc(const char *key, const char *user, const char *secret,void **pRpcObj);
Y
yihaoDeng 已提交
344
void tscReleaseRpc(void *param);
Y
yihaoDeng 已提交
345
void tscInitMsgsFp();
S
slguan 已提交
346

H
Haojun Liao 已提交
347
int tsParseSql(SSqlObj *pSql, bool initial);
H
hzcheng 已提交
348

349
void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet);
350
int  tscBuildAndSendRequest(SSqlObj *pSql, SQueryInfo* pQueryInfo);
H
hzcheng 已提交
351

B
Bomin Zhang 已提交
352
int  tscRenewTableMeta(SSqlObj *pSql, int32_t tableIndex);
H
Haojun Liao 已提交
353
void tscAsyncResultOnError(SSqlObj *pSql);
S
slguan 已提交
354

H
[td-99]  
hjxilinx 已提交
355
void tscQueueAsyncError(void(*fp), void *param, int32_t code);
H
hzcheng 已提交
356 357 358 359

int tscProcessLocalCmd(SSqlObj *pSql);
int tscCfgDynamicOptions(char *msg);

H
Haojun Liao 已提交
360 361
int32_t tscTansformFuncForSTableQuery(SQueryInfo *pQueryInfo);
void    tscRestoreFuncForSTableQuery(SQueryInfo *pQueryInfo);
H
hzcheng 已提交
362

H
Haojun Liao 已提交
363 364
int32_t tscCreateResPointerInfo(SSqlRes *pRes, SQueryInfo *pQueryInfo);
void tscSetResRawPtr(SSqlRes* pRes, SQueryInfo* pQueryInfo);
365
void tscSetResRawPtrRv(SSqlRes* pRes, SQueryInfo* pQueryInfo, SSDataBlock* pBlock, bool convertNchar);
H
Haojun Liao 已提交
366

367
void handleDownstreamOperator(SSqlObj** pSqlList, int32_t numOfUpstream, SQueryInfo* px, SSqlObj* pParent);
H
Haojun Liao 已提交
368
void destroyTableNameList(SInsertStatementParam* pInsertParam);
H
hzcheng 已提交
369

370
void tscResetSqlCmd(SSqlCmd *pCmd, bool removeMeta);
H
hzcheng 已提交
371

weixin_48148422's avatar
weixin_48148422 已提交
372 373 374 375
/**
 * free query result of the sql object
 * @param pObj
 */
H
hjxilinx 已提交
376
void tscFreeSqlResult(SSqlObj *pSql);
weixin_48148422's avatar
weixin_48148422 已提交
377

H
hzcheng 已提交
378 379
/**
 * free sql object, release allocated resource
H
Haojun Liao 已提交
380
 * @param pObj
H
hzcheng 已提交
381
 */
H
Haojun Liao 已提交
382
void tscFreeSqlObj(SSqlObj *pSql);
D
dapan1121 已提交
383 384
void tscFreeSubobj(SSqlObj* pSql);

H
Haojun Liao 已提交
385
void tscFreeRegisteredSqlObj(void *pSql);
H
hzcheng 已提交
386

陶建辉(Jeff)'s avatar
陶建辉(Jeff) 已提交
387
void tscCloseTscObj(void *pObj);
H
hzcheng 已提交
388

H
Haojun Liao 已提交
389
// todo move to taos? or create a new file: taos_internal.h
H
hjxilinx 已提交
390
TAOS *taos_connect_a(char *ip, char *user, char *pass, char *db, uint16_t port, void (*fp)(void *, TAOS_RES *, int),
391
                     void *param, TAOS **taos);
D
fix bug  
dapan1121 已提交
392
TAOS_RES* taos_query_h(TAOS* taos, const char *sqlstr, int64_t* res);
D
dapan1121 已提交
393 394
TAOS_RES * taos_query_ra(TAOS *taos, const char *sqlstr, __async_cb_func_t fp, void *param);

H
Haojun Liao 已提交
395
void waitForQueryRsp(void *param, TAOS_RES *tres, int code);
H
hjxilinx 已提交
396

H
Haojun Liao 已提交
397
void doAsyncQuery(STscObj *pObj, SSqlObj *pSql, __async_cb_func_t fp, void *param, const char *sqlstr, size_t sqlLen);
398

H
Haojun Liao 已提交
399
void tscImportDataFromFile(SSqlObj *pSql);
H
Haojun Liao 已提交
400
struct SGlobalMerger* tscInitResObjForLocalQuery(int32_t numOfRes, int32_t rowLen, uint64_t id);
H
Haojun Liao 已提交
401
bool tscIsUpdateQuery(SSqlObj* pSql);
D
fix bug  
dapan1121 已提交
402 403 404
char* tscGetSqlStr(SSqlObj* pSql);
bool tscIsQueryWithLimit(SSqlObj* pSql);

H
Haojun Liao 已提交
405
bool tscHasReachLimitation(SQueryInfo *pQueryInfo, SSqlRes *pRes);
H
Haojun Liao 已提交
406
void tscSetBoundColumnInfo(SParsedDataColInfo *pColInfo, SSchema *pSchema, int32_t numOfCols);
H
Haojun Liao 已提交
407

408
char *tscGetErrorMsgPayload(SSqlCmd *pCmd);
409

H
Haojun Liao 已提交
410
int32_t tscInvalidOperationMsg(char *msg, const char *additionalInfo, const char *sql);
Y
TD-934  
yihaoDeng 已提交
411
int32_t tscSQLSyntaxErrMsg(char* msg, const char* additionalInfo,  const char* sql);
H
hzcheng 已提交
412

413
int32_t tscValidateSqlInfo(SSqlObj *pSql, struct SSqlInfo *pInfo);
H
Haojun Liao 已提交
414

415
int32_t tsSetBlockInfo(SSubmitBlk *pBlocks, const STableMeta *pTableMeta, int32_t numOfRows);
416 417
extern int32_t    sentinel;
extern SHashObj  *tscVgroupMap;
418
extern SHashObj  *tscTableMetaInfo;
419 420 421 422 423 424 425

extern int   tscObjRef;
extern void *tscTmr;
extern void *tscQhandle;
extern int   tscKeepConn[];
extern int   tscRefId;
extern int   tscNumOfObj;     // number of existed sqlObj in current process.
H
hjxilinx 已提交
426 427

extern int (*tscBuildMsg[TSDB_SQL_MAX])(SSqlObj *pSql, SSqlInfo *pInfo);
C
Cary Xu 已提交
428
 
H
Haojun Liao 已提交
429
void tscBuildVgroupTableInfo(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo, SArray* tables);
H
Haojun Liao 已提交
430
int16_t getNewResColId(SSqlCmd* pCmd);
weixin_48148422's avatar
weixin_48148422 已提交
431

432 433 434 435 436
int32_t schemaIdxCompar(const void *lhs, const void *rhs);
int32_t boundIdxCompar(const void *lhs, const void *rhs);
int     initSMemRowHelper(SMemRowHelper *pHelper, SSchema *pSSchema, uint16_t nCols, uint16_t allNullColsLen);
int32_t getExtendedRowSize(STableComInfo *tinfo);

H
hzcheng 已提交
437 438 439 440 441
#ifdef __cplusplus
}
#endif

#endif