tsclient.h 14.6 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
hzcheng 已提交
25
#include "taos.h"
26
#include "taosdef.h"
H
hzcheng 已提交
27
#include "taosmsg.h"
H
hjxilinx 已提交
28
#include "tarray.h"
S
slguan 已提交
29
#include "tglobal.h"
30
#include "tsqlfunction.h"
H
hjxilinx 已提交
31 32
#include "tutil.h"

33
#include "qExecutor.h"
H
hjxilinx 已提交
34 35
#include "qsqlparser.h"
#include "qtsbuf.h"
H
Haojun Liao 已提交
36
#include "tcmdtype.h"
H
hzcheng 已提交
37 38 39

// forward declaration
struct SSqlInfo;
H
hjxilinx 已提交
40 41 42 43 44 45 46
struct SLocalReducer;

// data source from sql string or from file
enum {
  DATA_FROM_SQL_STRING = 1,
  DATA_FROM_DATA_FILE = 2,
};
H
hzcheng 已提交
47

H
hjxilinx 已提交
48
typedef struct STableComInfo {
H
hjxilinx 已提交
49 50 51
  uint8_t numOfTags;
  uint8_t precision;
  int16_t numOfColumns;
B
Bomin Zhang 已提交
52
  int32_t rowSize;
H
hjxilinx 已提交
53
} STableComInfo;
H
hjxilinx 已提交
54 55

typedef struct STableMeta {
H
Haojun Liao 已提交
56
  STableComInfo tableInfo;
H
hjxilinx 已提交
57 58
  uint8_t       tableType;
  int16_t       sversion;
59
  int16_t       tversion;
H
hjxilinx 已提交
60
  SCMVgroupInfo vgroupInfo;
H
hjxilinx 已提交
61 62 63
  int32_t       sid;       // the index of one table in a virtual node
  uint64_t      uid;       // unique id of a table
  SSchema       schema[];  // if the table is TSDB_CHILD_TABLE, schema is acquired by super table meta info
H
hjxilinx 已提交
64 65 66
} STableMeta;

typedef struct STableMetaInfo {
67
  STableMeta *  pTableMeta;      // table meta, cached in client side and acquired by name
H
hjxilinx 已提交
68
  SVgroupsInfo *vgroupList;
69 70
  SArray       *pVgroupTables;   // SArray<SVgroupTableInfo>
  
H
hjxilinx 已提交
71
  /*
H
hjxilinx 已提交
72 73
   * 1. keep the vgroup index during the multi-vnode super table projection query
   * 2. keep the vgroup index for multi-vnode insertion
H
hjxilinx 已提交
74
   */
H
hjxilinx 已提交
75
  int32_t vgroupIndex;
H
hjxilinx 已提交
76
  char    name[TSDB_TABLE_ID_LEN];        // (super) table name
77
  SArray* tagColList;                     // SArray<SColumn*>, involved tag columns
H
hjxilinx 已提交
78
} STableMetaInfo;
H
hzcheng 已提交
79

S
slguan 已提交
80 81
/* the structure for sql function in select clause */
typedef struct SSqlExpr {
H
hjxilinx 已提交
82 83 84 85 86 87
  char      aliasName[TSDB_COL_NAME_LEN];  // as aliasName
  SColIndex colInfo;
  int64_t   uid;            // refactor use the pointer
  int16_t   functionId;     // function id in aAgg array
  int16_t   resType;        // return value type
  int16_t   resBytes;       // length of return value
H
Haojun Liao 已提交
88
  int32_t   interBytes;     // inter result buffer size
H
hjxilinx 已提交
89 90 91
  int16_t   numOfParams;    // argument value of each function
  tVariant  param[3];       // parameters are not more than 3
  int32_t   offset;         // sub result column value of arithmetic expression.
H
hzcheng 已提交
92 93
} SSqlExpr;

H
hjxilinx 已提交
94 95 96 97 98
typedef struct SColumnIndex {
  int16_t tableIndex;
  int16_t columnIndex;
} SColumnIndex;

H
hjxilinx 已提交
99 100
typedef struct SFieldSupInfo {
  bool            visible;
H
hjxilinx 已提交
101
  SExprInfo      *pArithExprInfo;
H
hjxilinx 已提交
102
  SSqlExpr *      pSqlExpr;
H
hjxilinx 已提交
103
} SFieldSupInfo;
S
slguan 已提交
104

H
hjxilinx 已提交
105
typedef struct SFieldInfo {
H
hjxilinx 已提交
106 107 108
  int16_t numOfOutput;   // number of column in result
  SArray *pFields;       // SArray<TAOS_FIELD>
  SArray *pSupportInfo;  // SArray<SFieldSupInfo>
H
hzcheng 已提交
109 110
} SFieldInfo;

111
typedef struct SColumn {
S
slguan 已提交
112 113 114
  SColumnIndex       colIndex;
  int32_t            numOfFilters;
  SColumnFilterInfo *filterInfo;
115
} SColumn;
H
hzcheng 已提交
116

S
slguan 已提交
117 118
typedef struct SCond {
  uint64_t uid;
H
hjxilinx 已提交
119
  int32_t  len;  // length of tag query condition data
H
hjxilinx 已提交
120
  char *   cond;
S
slguan 已提交
121 122 123
} SCond;

typedef struct SJoinNode {
S
slguan 已提交
124
  char     tableId[TSDB_TABLE_ID_LEN];
S
slguan 已提交
125
  uint64_t uid;
H
Haojun Liao 已提交
126
  int16_t  tagColId;
S
slguan 已提交
127 128 129 130 131 132 133 134
} SJoinNode;

typedef struct SJoinInfo {
  bool      hasJoin;
  SJoinNode left;
  SJoinNode right;
} SJoinInfo;

H
hzcheng 已提交
135
typedef struct STagCond {
S
slguan 已提交
136 137 138 139 140 141 142 143 144 145
  // relation between tbname list and query condition, including : TK_AND or TK_OR
  int16_t relType;

  // tbname query condition, only support tbname query condition on one table
  SCond tbnameCond;

  // join condition, only support two tables join currently
  SJoinInfo joinInfo;

  // for different table, the query condition must be seperated
H
hjxilinx 已提交
146
  SArray *pCond;
H
hzcheng 已提交
147 148
} STagCond;

S
slguan 已提交
149
typedef struct SParamInfo {
H
hjxilinx 已提交
150 151 152 153
  int32_t  idx;
  char     type;
  uint8_t  timePrec;
  short    bytes;
S
slguan 已提交
154 155 156
  uint32_t offset;
} SParamInfo;

S
slguan 已提交
157
typedef struct STableDataBlocks {
H
Haojun Liao 已提交
158 159 160 161 162 163 164
  char     tableId[TSDB_TABLE_ID_LEN];
  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
S
slguan 已提交
165
  uint32_t nAllocSize;
H
Haojun Liao 已提交
166
  uint32_t headerSize;   // header for table info (uid, tid, submit metadata)
S
slguan 已提交
167
  uint32_t size;
168

H
hjxilinx 已提交
169
  /*
H
Haojun Liao 已提交
170
   * the table meta of table, the table meta will be used during submit, keep a ref
H
hjxilinx 已提交
171 172
   * to avoid it to be removed from cache
   */
H
hjxilinx 已提交
173
  STableMeta *pTableMeta;
H
Haojun Liao 已提交
174
  char       *pData;
S
slguan 已提交
175 176

  // for parameter ('?') binding
H
hjxilinx 已提交
177 178 179
  uint32_t    numOfAllocedParams;
  uint32_t    numOfParams;
  SParamInfo *params;
S
slguan 已提交
180
} STableDataBlocks;
H
hzcheng 已提交
181

182 183 184 185 186
//typedef struct SDataBlockList {  // todo remove
//  uint32_t           nSize;
//  uint32_t           nAlloc;
//  STableDataBlocks **pData;
//} SDataBlockList;
H
hzcheng 已提交
187

188
typedef struct SQueryInfo {
H
Haojun Liao 已提交
189 190
  int16_t          command;       // the command may be different for each subclause, so keep it seperately.
  uint32_t         type;          // query/insert/import type
191 192 193 194 195
  char             slidingTimeUnit;
  STimeWindow      window;
  int64_t          intervalTime;  // aggregation time interval
  int64_t          slidingTime;   // sliding window in mseconds
  SSqlGroupbyExpr  groupbyExpr;   // group by tags info
H
Haojun Liao 已提交
196
  SArray *         colList;       // SArray<SColumn*>
197
  SFieldInfo       fieldsInfo;
H
Haojun Liao 已提交
198
  SArray *         exprList;      // SArray<SSqlExpr*>
199 200 201 202
  SLimitVal        limit;
  SLimitVal        slimit;
  STagCond         tagCond;
  SOrderVal        order;
H
Haojun Liao 已提交
203
  int16_t          fillType;      // final result fill type
204
  int16_t          numOfTables;
H
hjxilinx 已提交
205
  STableMetaInfo **pTableMetaInfo;
206
  struct STSBuf *  tsBuf;
H
Haojun Liao 已提交
207 208 209 210
  int64_t *        fillVal;       // default value for fill
  char *           msg;           // pointer to the pCmd->payload to keep error message temporarily
  int64_t          clauseLimit;   // limit for current sub clause
  int64_t          prjOffset;     // offset value in the original sql expression, only applied at client side
211 212
} SQueryInfo;

H
hzcheng 已提交
213
typedef struct {
214 215
  int     command;
  uint8_t msgType;
216
  bool    autoCreated;        // create table if it is not existed during retrieve table meta in mnode
H
hjxilinx 已提交
217

218 219 220 221
  union {
    int32_t count;
    int32_t numOfTablesInSubmit;
  };
H
hzcheng 已提交
222

223
  int32_t      insertType;
224 225 226
  int32_t      clauseIndex;   // index of multiple subclause query

  char *       curSql;       // current sql, resume position of sql after parsing paused
227
  int8_t       parseFinished;
228

229 230 231 232
  short        numOfCols;
  uint32_t     allocSize;
  char *       payload;
  int32_t      payloadLen;
233 234
  SQueryInfo **pQueryInfo;
  int32_t      numOfClause;
235 236
  int32_t      batchSize;    // for parameter ('?') binding and batch processing
  int32_t      numOfParams;
237 238 239 240

  int8_t       dataSourceType;     // load data from file or not
  int8_t       submitSchema;  // submit block is built with table schema
  SHashObj    *pTableList;   // referred table involved in sql
241
  SArray      *pDataBlocks;  // SArray<STableDataBlocks*> submit data blocks after parsing sql
H
hzcheng 已提交
242 243 244 245 246 247 248 249
} SSqlCmd;

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

typedef struct {
H
hjxilinx 已提交
250 251
  int64_t               numOfRows;                  // num of results in current retrieved
  int64_t               numOfTotal;                 // num of total results
H
Haojun Liao 已提交
252
  int64_t               numOfClauseTotal;           // num of total result in current subclause
H
hjxilinx 已提交
253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268
  char *                pRsp;
  int32_t               rspType;
  int32_t               rspLen;
  uint64_t              qhandle;
  int64_t               uid;
  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;
  void **               tsrow;
269
  int32_t*              length;  // length for each field for current row
H
hjxilinx 已提交
270 271
  char **               buffer;  // Buffer used to put multibytes encoded using unicode (wchar_t)
  SColumnIndex *        pColumnIndex;
272 273
  SArithmeticSupport*   pArithSup;   // support the arithmetic expression calculation on agg functions
  
H
hzcheng 已提交
274 275 276
  struct SLocalReducer *pLocalReducer;
} SSqlRes;

H
hjxilinx 已提交
277
typedef struct STscObj {
H
hjxilinx 已提交
278 279 280 281
  void *             signature;
  void *             pTimer;
  char               user[TSDB_USER_LEN];
  char               pass[TSDB_KEY_LEN];
B
Bomin Zhang 已提交
282
  char               acctId[TSDB_ACCT_LEN];
B
Bomin Zhang 已提交
283
  char               db[TSDB_ACCT_LEN + TSDB_DB_NAME_LEN];
H
hjxilinx 已提交
284 285 286
  char               sversion[TSDB_VERSION_LEN];
  char               writeAuth : 1;
  char               superAuth : 1;
S
Shengliang Guan 已提交
287
  uint32_t           connId;
H
hjxilinx 已提交
288 289
  struct SSqlObj *   pHb;
  struct SSqlObj *   sqlList;
H
hjxilinx 已提交
290
  struct SSqlStream *streamList;
291
  void*              pDnodeConn;
H
hjxilinx 已提交
292
  pthread_mutex_t    mutex;
H
hzcheng 已提交
293 294
} STscObj;

295
typedef struct SSqlObj {
296 297
  void            *signature;
  STscObj         *pTscObj;
H
Haojun Liao 已提交
298
  void            *pRpcCtx;
299 300 301
  void            (*fp)();
  void            (*fetchFp)();
  void            *param;
H
hjxilinx 已提交
302 303 304 305 306 307 308 309
  int64_t          stime;
  uint32_t         queryId;
  void *           pStream;
  void *           pSubscription;
  char *           sqlstr;
  char             retry;
  char             maxRetry;
  SRpcIpSet        ipList;
H
Haojun Liao 已提交
310
  char             listed;
H
hjxilinx 已提交
311 312 313
  tsem_t           rspSem;
  SSqlCmd          cmd;
  SSqlRes          res;
H
Haojun Liao 已提交
314
  uint16_t         numOfSubs;
315 316
  struct SSqlObj **pSubs;
  struct SSqlObj * prev, *next;
H
hzcheng 已提交
317 318
} SSqlObj;

H
hjxilinx 已提交
319
typedef struct SSqlStream {
H
hzcheng 已提交
320 321 322
  SSqlObj *pSql;
  uint32_t streamId;
  char     listed;
323 324
  bool     isProject;
  int16_t  precision;
H
hzcheng 已提交
325 326 327
  int64_t  num;  // number of computing count

  /*
S
slguan 已提交
328
   * keep the number of current result in computing,
H
hzcheng 已提交
329 330 331 332 333 334 335
   * 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 已提交
336
  int64_t etime;     // stream end query time, when time is larger then etime, the stream will be closed
H
hzcheng 已提交
337 338 339 340 341 342 343
  int64_t interval;
  int64_t slidingTime;
  void *  pTimer;

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

S
slguan 已提交
344
  void (*callback)(void *);  // Callback function when stream is stopped from client level
H
hjxilinx 已提交
345
  struct SSqlStream *prev, *next;
H
hzcheng 已提交
346 347
} SSqlStream;

348
int32_t tscInitRpc(const char *user, const char *secret, void** pDnodeConn);
H
hjxilinx 已提交
349
void    tscInitMsgsFp();
S
slguan 已提交
350

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

J
jtao1735 已提交
353
void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcIpSet *pIpSet);
S
slguan 已提交
354
int  tscProcessSql(SSqlObj *pSql);
H
hzcheng 已提交
355

H
Haojun Liao 已提交
356
int  tscRenewTableMeta(SSqlObj *pSql, char *tableId);
H
hzcheng 已提交
357
void tscQueueAsyncRes(SSqlObj *pSql);
S
slguan 已提交
358

H
[td-99]  
hjxilinx 已提交
359
void tscQueueAsyncError(void(*fp), void *param, int32_t code);
H
hzcheng 已提交
360 361 362 363 364

int tscProcessLocalCmd(SSqlObj *pSql);
int tscCfgDynamicOptions(char *msg);
int taos_retrieve(TAOS_RES *res);

H
hjxilinx 已提交
365 366
int32_t tscTansformSQLFuncForSTableQuery(SQueryInfo *pQueryInfo);
void    tscRestoreSQLFuncForSTableQuery(SQueryInfo *pQueryInfo);
H
hzcheng 已提交
367

368
int32_t tscCreateResPointerInfo(SSqlRes *pRes, SQueryInfo *pQueryInfo);
H
hjxilinx 已提交
369
void    tscDestroyResPointerInfo(SSqlRes *pRes);
H
hzcheng 已提交
370

371
void tscResetSqlCmdObj(SSqlCmd *pCmd);
H
hzcheng 已提交
372

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

H
hzcheng 已提交
379 380 381 382 383
/**
 * only free part of resources allocated during query.
 * Note: this function is multi-thread safe.
 * @param pObj
 */
H
hjxilinx 已提交
384
void tscPartiallyFreeSqlObj(SSqlObj *pObj);
H
hzcheng 已提交
385 386 387 388 389 390 391 392 393 394

/**
 * free sql object, release allocated resource
 * @param pObj  Free metric/meta information, dynamically allocated payload, and
 * response buffer, object itself
 */
void tscFreeSqlObj(SSqlObj *pObj);

void tscCloseTscObj(STscObj *pObj);

H
hjxilinx 已提交
395 396
TAOS *taos_connect_a(char *ip, char *user, char *pass, char *db, uint16_t port, void (*fp)(void *, TAOS_RES *, int),
                     void *param, void **taos);
H
Hui Li 已提交
397
void waitForQueryRsp(void *param, TAOS_RES *tres, int code) ;
H
hjxilinx 已提交
398 399

void doAsyncQuery(STscObj *pObj, SSqlObj *pSql, void (*fp)(), void *param, const char *sqlstr, size_t sqlLen);
400

H
Haojun Liao 已提交
401
void tscProcessMultiVnodesImportFromFile(SSqlObj *pSql);
H
hjxilinx 已提交
402
void tscKillSTableQuery(SSqlObj *pSql);
403
void tscInitResObjForLocalQuery(SSqlObj *pObj, int32_t numOfRes, int32_t rowLen);
H
Haojun Liao 已提交
404
bool tscIsUpdateQuery(SSqlObj* pSql);
405
bool tscHasReachLimitation(SQueryInfo *pQueryInfo, SSqlRes *pRes);
H
Haojun Liao 已提交
406 407

// todo remove this function.
408
bool tscResultsetFetchCompleted(TAOS_RES *result);
H
hjxilinx 已提交
409

410
char *tscGetErrorMsgPayload(SSqlCmd *pCmd);
411

H
hjxilinx 已提交
412
int32_t tscInvalidSQLErrMsg(char *msg, const char *additionalInfo, const char *sql);
H
hzcheng 已提交
413

H
hjxilinx 已提交
414
int32_t tscToSQLCmd(SSqlObj *pSql, struct SSqlInfo *pInfo);
H
Haojun Liao 已提交
415 416 417
//void    tscGetResultColumnChr(SSqlRes *pRes, SFieldInfo* pFieldInfo, int32_t column);

static FORCE_INLINE void tscGetResultColumnChr(SSqlRes* pRes, SFieldInfo* pFieldInfo, int32_t columnIndex) {
H
Haojun Liao 已提交
418
  SFieldSupInfo* pInfo = (SFieldSupInfo*) TARRAY_GET_ELEM(pFieldInfo->pSupportInfo, columnIndex);
H
Haojun Liao 已提交
419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452
  assert(pInfo->pSqlExpr != NULL);

  int32_t type = pInfo->pSqlExpr->resType;
  int32_t bytes = pInfo->pSqlExpr->resBytes;

  char* pData = pRes->data + pInfo->pSqlExpr->offset * pRes->numOfRows + bytes * pRes->row;

  if (type == TSDB_DATA_TYPE_NCHAR || type == TSDB_DATA_TYPE_BINARY) {
    int32_t realLen = varDataLen(pData);
    assert(realLen <= bytes - VARSTR_HEADER_SIZE);

    if (isNull(pData, type)) {
      pRes->tsrow[columnIndex] = NULL;
    } else {
      pRes->tsrow[columnIndex] = ((tstr*)pData)->data;
    }

    if (realLen < pInfo->pSqlExpr->resBytes - VARSTR_HEADER_SIZE) { // todo refactor
      *(pData + realLen + VARSTR_HEADER_SIZE) = 0;
    }

    pRes->length[columnIndex] = realLen;
  } else {
    assert(bytes == tDataTypeDesc[type].nSize);

    if (isNull(pData, type)) {
      pRes->tsrow[columnIndex] = NULL;
    } else {
      pRes->tsrow[columnIndex] = pData;
    }

    pRes->length[columnIndex] = bytes;
  }
}
H
hjxilinx 已提交
453 454 455 456 457 458 459

extern void *    tscCacheHandle;
extern void *    tscTmr;
extern void *    tscQhandle;
extern int       tscKeepConn[];
extern int       tsInsertHeadSize;
extern int       tscNumOfThreads;
460
extern SRpcCorIpSet tscMgmtIpSet;
H
hjxilinx 已提交
461 462

extern int (*tscBuildMsg[TSDB_SQL_MAX])(SSqlObj *pSql, SSqlInfo *pInfo);
H
hzcheng 已提交
463

464 465
typedef void (*__async_cb_func_t)(void *param, TAOS_RES *tres, int numOfRows);

weixin_48148422's avatar
weixin_48148422 已提交
466
int32_t tscCompareTidTags(const void* p1, const void* p2);
H
Haojun Liao 已提交
467
void tscBuildVgroupTableInfo(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo, SArray* tables);
weixin_48148422's avatar
weixin_48148422 已提交
468

H
hzcheng 已提交
469 470 471 472 473
#ifdef __cplusplus
}
#endif

#endif