Skip to content

TDengine
TDengine

taosdata / TDengine
11 个月 前同步成功

通知 1179
Star 22014
Fork 4786
TDengine
TDengine

体验新版 GitCode，发现更多精彩内容 >>

提交 863bda56 编写于 作者: H hzcheng
浏览文件
操作

Merge branch 'develop' into feature/2.0tsdb

上级 512572e8 99392bee
展开全部 隐藏空白更改
内联 并排
Showing with 3219 addition and 2441 deletion
.travis.yml
浏览文件 @ 863bda56
......@@ -26,10 +26,6 @@ matrix:
- python3-setuptools
- valgrind
before_install:
- sudo apt update -y -qq
- sudo apt install -y net-tools python-pip python-setuptools python3-pip python3-setuptools valgrind
before_script:
- cd ${TRAVIS_BUILD_DIR}
- mkdir debug
......@@ -63,11 +59,14 @@ matrix:
GREEN_UNDERLINE='\033[4;32m'
NC='\033[0m'
tail -10 mem-error-out.txt
defiMemError=`grep -m 1 'definitely lost' mem-error-out.txt | awk '{print $7}'`
memError=`grep -m 1 'ERROR SUMMARY' mem-error-out.txt | awk '{print $4}'`
if [ -n "$memError" ]; then
if [ "$memError" -gt 23 ]; then
echo -e "${RED} ## Memory errors number valgrind reports is $memError. More than our threshold! ## ${NC} "
if [ "$memError" -gt 16 ] && [ "$defiMemError" -gt 0 ]; then
echo -e "${RED} ## Memory errors number valgrind reports is $memError.\
Definitely lost is $defiMemError. More than our threshold! ## ${NC}"
travis_terminate $memError
fi
fi
......@@ -131,10 +130,6 @@ matrix:
- python3-setuptools
- lcov
before_install:
- sudo apt update -y -qq
- sudo apt install -y net-tools python-pip python-setuptools python3-pip python3-setuptools lcov
before_script:
- cd ${TRAVIS_BUILD_DIR}
- mkdir debug
......
src/client/inc/tscUtil.h
浏览文件 @ 863bda56
......@@ -57,8 +57,8 @@ typedef struct SJoinSubquerySupporter {
int64_t interval; // interval time
SLimitVal limit; // limit info
uint64_t uid; // query meter uid
SColumnBaseInfo colList; // previous query information
SSqlExprInfo exprsInfo;
SArray* colList; // previous query information
SArray* exprsInfo;
SFieldInfo fieldsInfo;
STagCond tagCond;
SSqlGroupbyExpr groupbyExpr;
......@@ -69,8 +69,9 @@ typedef struct SJoinSubquerySupporter {
int32_t tscCreateDataBlock(size_t initialSize, int32_t rowSize, int32_t startOffset, const char* name,
STableMeta* pTableMeta, STableDataBlocks** dataBlocks);
void tscAppendDataBlock(SDataBlockList* pList, STableDataBlocks* pBlocks);
void tscDestroyDataBlock(STableDataBlocks* pDataBlock);
void tscAppendDataBlock(SDataBlockList* pList, STableDataBlocks* pBlocks);
void tscDestroyDataBlock(STableDataBlocks* pDataBlock);
void tscSortRemoveDataBlockDupRows(STableDataBlocks* dataBuf);
SParamInfo* tscAddParamToDataBlock(STableDataBlocks* pDataBlock, char type, uint8_t timePrec, short bytes,
uint32_t offset);
......@@ -85,8 +86,7 @@ int32_t tscGetDataBlockFromList(void* pHashList, SDataBlockList* pDataBlockList,
int32_t startOffset, int32_t rowSize, const char* tableId, STableMeta* pTableMeta,
STableDataBlocks** dataBlocks);
SVnodeSidList* tscGetVnodeSidList(SSuperTableMeta* pMetricmeta, int32_t vnodeIdx);
STableIdInfo* tscGetMeterSidInfo(SVnodeSidList* pSidList, int32_t idx);
UNUSED_FUNC STableIdInfo* tscGetMeterSidInfo(SVnodeSidList* pSidList, int32_t idx);
/**
*
......@@ -106,7 +106,7 @@ bool tscIsProjectionQueryOnSTable(SQueryInfo* pQueryInfo, int32_t tableIndex);
bool tscProjectionQueryOnTable(SQueryInfo* pQueryInfo);
bool tscIsTwoStageSTableQuery(SQueryInfo* pQueryInfo, int32_t tableIndex);
bool tscQueryOnMetric(SSqlCmd* pCmd);
bool tscQueryOnSTable(SSqlCmd* pCmd);
bool tscQueryTags(SQueryInfo* pQueryInfo);
bool tscIsSelectivityWithTagQuery(SSqlCmd* pCmd);
......@@ -115,60 +115,58 @@ void tscAddSpecialColumnForSelect(SQueryInfo* pQueryInfo, int32_t outputColIndex
void addRequiredTagColumn(SQueryInfo* pQueryInfo, int32_t tagColIndex, int32_t tableIndex);
int32_t setMeterID(STableMetaInfo* pTableMetaInfo, SSQLToken* pzTableName, SSqlObj* pSql);
int32_t tscSetTableId(STableMetaInfo* pTableMetaInfo, SSQLToken* pzTableName, SSqlObj* pSql);
void tscClearInterpInfo(SQueryInfo* pQueryInfo);
bool tscIsInsertOrImportData(char* sqlstr);
/* use for keep current db info temporarily, for handle table with db prefix */
// todo remove it
void tscGetDBInfoFromMeterId(char* tableId, char* db);
int tscAllocPayload(SSqlCmd* pCmd, int size);
void tscFieldInfoSetValFromSchema(SFieldInfo* pFieldInfo, int32_t index, SSchema* pSchema);
void tscFieldInfoSetValFromField(SFieldInfo* pFieldInfo, int32_t index, TAOS_FIELD* pField);
void tscFieldInfoSetValue(SFieldInfo* pFieldInfo, int32_t index, int8_t type, const char* name, int16_t bytes);
void tscFieldInfoUpdateVisible(SFieldInfo* pFieldInfo, int32_t index, bool visible);
void tscFieldInfoSetExpr(SFieldInfo* pFieldInfo, int32_t index, SSqlExpr* pExpr);
void tscFieldInfoSetBinExpr(SFieldInfo* pFieldInfo, int32_t index, SSqlFunctionExpr* pExpr);
TAOS_FIELD tscCreateField(int8_t type, const char* name, int16_t bytes);
void tscFieldInfoCalOffset(SQueryInfo* pQueryInfo);
void tscFieldInfoCopy(SFieldInfo* src, SFieldInfo* dst, const int32_t* indexList, int32_t size);
void tscFieldInfoCopyAll(SFieldInfo* dst, SFieldInfo* src);
SFieldSupInfo* tscFieldInfoAppend(SFieldInfo* pFieldInfo, TAOS_FIELD* pField);
SFieldSupInfo* tscFieldInfoInsert(SFieldInfo* pFieldInfo, int32_t index, TAOS_FIELD* field);
SFieldSupInfo* tscFieldInfoGetSupp(SFieldInfo* pFieldInfo, int32_t index);
TAOS_FIELD* tscFieldInfoGetField(SFieldInfo* pFieldInfo, int32_t index);
void tscFieldInfoUpdateOffset(SQueryInfo* pQueryInfo);
void tscFieldInfoCopy(SFieldInfo* dst, const SFieldInfo* src);
void tscFieldInfoUpdateOffsetForInterResult(SQueryInfo* pQueryInfo);
TAOS_FIELD* tscFieldInfoGetField(SQueryInfo* pQueryInfo, int32_t index);
int16_t tscFieldInfoGetOffset(SQueryInfo* pQueryInfo, int32_t index);
int32_t tscGetResRowLength(SQueryInfo* pQueryInfo);
void tscClearFieldInfo(SFieldInfo* pFieldInfo);
int16_t tscFieldInfoGetOffset(SQueryInfo* pQueryInfo, int32_t index);
void tscFieldInfoClear(SFieldInfo* pFieldInfo);
int32_t tscNumOfFields(SQueryInfo* pQueryInfo);
int32_t tscFieldInfoCompare(SFieldInfo* pFieldInfo1, SFieldInfo* pFieldInfo2);
int32_t tscFieldInfoCompare(const SFieldInfo* pFieldInfo1, const SFieldInfo* pFieldInfo2);
void addExprParams(SSqlExpr* pExpr, char* argument, int32_t type, int32_t bytes, int16_t tableIndex);
int32_t tscGetResRowLength(SArray* pExprList);
SSqlExpr* tscSqlExprInsert(SQueryInfo* pQueryInfo, int32_t index, int16_t functionId, SColumnIndex* pColIndex, int16_t type,
int16_t size, int16_t interSize);
SSqlExpr* tscSqlExprAppend(SQueryInfo* pQueryInfo, int16_t functionId, SColumnIndex* pColIndex, int16_t type,
int16_t size, int16_t interSize);
SSqlExpr* tscSqlExprInsertEmpty(SQueryInfo* pQueryInfo, int32_t index, int16_t functionId);
SSqlExpr* tscSqlExprUpdate(SQueryInfo* pQueryInfo, int32_t index, int16_t functionId, int16_t srcColumnIndex, int16_t type,
int16_t size);
int32_t tscSqlExprNumOfExprs(SQueryInfo* pQueryInfo);
SSqlExpr* tscSqlExprGet(SQueryInfo* pQueryInfo, int32_t index);
void tscSqlExprCopy(SSqlExprInfo* dst, const SSqlExprInfo* src, uint64_t uid, bool deepcopy);
void* tscSqlExprDestroy(SSqlExpr* pExpr);
void tscSqlExprInfoDestroy(SSqlExprInfo* pExprInfo);
SArray* tscSqlExprCopy(const SArray* src, uint64_t uid, bool deepcopy);
void tscSqlExprInfoDestroy(SArray* pExprInfo);
SColumnBase* tscColumnBaseInfoInsert(SQueryInfo* pQueryInfo, SColumnIndex* colIndex);
void tscColumnFilterInfoCopy(SColumnFilterInfo* dst, const SColumnFilterInfo* src);
void tscColumnBaseCopy(SColumnBase* dst, const SColumnBase* src);
SColumn* tscColumnClone(const SColumn* src);
SColumn* tscColumnListInsert(SArray* pColList, SColumnIndex* colIndex);
void tscColumnListCopy(SArray* dst, const SArray* src, int16_t tableIndex);
void tscColumnListDestroy(SArray* pColList);
void tscColumnBaseInfoCopy(SColumnBaseInfo* dst, const SColumnBaseInfo* src, int16_t tableIndex);
SColumnBase* tscColumnBaseInfoGet(SColumnBaseInfo* pColumnBaseInfo, int32_t index);
void tscColumnBaseInfoUpdateTableIndex(SColumnBaseInfo* pColList, int16_t tableIndex);
void tscColumnBaseInfoReserve(SColumnBaseInfo* pColumnBaseInfo, int32_t size);
void tscColumnBaseInfoDestroy(SColumnBaseInfo* pColumnBaseInfo);
SColumnFilterInfo* tscFilterInfoClone(const SColumnFilterInfo* src, int32_t numOfFilters);
int32_t tscValidateName(SSQLToken* pToken);
......@@ -188,17 +186,16 @@ void tscGetSrcColumnInfo(SSrcColumnInfo* pColInfo, SQueryInfo* pQueryInfo);
void tscSetFreeHeatBeat(STscObj* pObj);
bool tscShouldFreeHeatBeat(SSqlObj* pHb);
void tscCleanSqlCmd(SSqlCmd* pCmd);
bool tscShouldFreeAsyncSqlObj(SSqlObj* pSql);
bool tscShouldBeFreed(SSqlObj* pSql);
void tscRemoveAllMeterMetaInfo(SQueryInfo* pQueryInfo, const char* address, bool removeFromCache);
void tscClearAllTableMetaInfo(SQueryInfo* pQueryInfo, const char* address, bool removeFromCache);
STableMetaInfo* tscGetTableMetaInfoFromCmd(SSqlCmd *pCmd, int32_t subClauseIndex, int32_t tableIndex);
STableMetaInfo* tscGetMetaInfo(SQueryInfo *pQueryInfo, int32_t tableIndex);
SQueryInfo *tscGetQueryInfoDetail(SSqlCmd* pCmd, int32_t subClauseIndex);
int32_t tscGetQueryInfoDetailSafely(SSqlCmd *pCmd, int32_t subClauseIndex, SQueryInfo** pQueryInfo);
STableMetaInfo* tscGetMeterMetaInfoByUid(SQueryInfo* pQueryInfo, uint64_t uid, int32_t* index);
void tscClearMeterMetaInfo(STableMetaInfo* pTableMetaInfo, bool removeFromCache);
void tscClearMeterMetaInfo(STableMetaInfo* pTableMetaInfo, bool removeFromCache);
STableMetaInfo* tscAddTableMetaInfo(SQueryInfo* pQueryInfo, const char* name, STableMeta* pTableMeta,
SVgroupsInfo* vgroupList, int16_t numOfTags, int16_t* tags);
......@@ -208,7 +205,6 @@ int32_t tscAddSubqueryInfo(SSqlCmd *pCmd);
void tscFreeSubqueryInfo(SSqlCmd* pCmd);
void tscClearSubqueryInfo(SSqlCmd* pCmd);
void tscGetMetricMetaCacheKey(SQueryInfo* pQueryInfo, char* keyStr, uint64_t uid);
int tscGetSTableVgroupInfo(SSqlObj* pSql, int32_t clauseIndex);
int tscGetTableMeta(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo);
int tscGetMeterMetaEx(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo, bool createIfNotExists);
......@@ -242,11 +238,6 @@ void doAddGroupColumnForSubquery(SQueryInfo* pQueryInfo, int32_t tagIndex);
int16_t tscGetJoinTagColIndexByUid(STagCond* pTagCond, uint64_t uid);
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);
void sortRemoveDuplicates(STableDataBlocks* dataBuf);
void tscPrintSelectClause(SSqlObj* pSql, int32_t subClauseIndex);
bool hasMoreVnodesToTry(SSqlObj *pSql);
......
src/client/inc/tsclient.h
浏览文件 @ 863bda56
......@@ -22,32 +22,30 @@ extern "C" {
#include "os.h"
#include "qsqlparser.h"
#include "qsqltype.h"
#include "qtsbuf.h"
#include "taos.h"
#include "taosdef.h"
#include "taosmsg.h"
#include "tarray.h"
#include "tglobal.h"
#include "trpc.h"
#include "tsqlfunction.h"
#include "tutil.h"
#define TSC_GET_RESPTR_BASE(res, _queryinfo, col) (res->data + ((_queryinfo)->fieldsInfo.pSqlExpr[col]->offset) * res->numOfRows)
#include "qsqlparser.h"
#include "qsqltype.h"
#include "qtsbuf.h"
#include "queryExecutor.h"
// forward declaration
struct SSqlInfo;
struct SLocalReducer;
typedef SCMSTableVgroupRspMsg SVgroupsInfo;
// data source from sql string or from file
enum {
DATA_FROM_SQL_STRING = 1,
DATA_FROM_DATA_FILE = 2,
};
typedef struct SSqlGroupbyExpr {
int16_t tableIndex;
int16_t numOfGroupCols;
SColIndex columnInfo[TSDB_MAX_TAGS]; // group by columns information
int16_t orderIndex; // order by column index
int16_t orderType; // order by type: asc/desc
} SSqlGroupbyExpr;
typedef SCMSTableVgroupRspMsg SVgroupsInfo;
typedef struct STableComInfo {
uint8_t numOfTags;
......@@ -57,20 +55,23 @@ typedef struct STableComInfo {
} STableComInfo;
typedef struct STableMeta {
//super table if it is created according to super table, otherwise, tableInfo is used
union { struct STableMeta* pSTable; STableComInfo tableInfo; };
uint8_t tableType;
int16_t sversion;
// super table if it is created according to super table, otherwise, tableInfo is used
union {
struct STableMeta *pSTable;
STableComInfo tableInfo;
};
uint8_t tableType;
int16_t sversion;
SCMVgroupInfo vgroupInfo;
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
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
} STableMeta;
typedef struct STableMetaInfo {
STableMeta * pTableMeta; // table meta, cached in client side and acquried by name
SVgroupsInfo* vgroupList;
STableMeta * pTableMeta; // table meta, cached in client side and acquried by name
SVgroupsInfo *vgroupList;
/*
* 1. keep the vnode index during the multi-vnode super table projection query
* 2. keep the vnode index for multi-vnode insertion
......@@ -83,16 +84,16 @@ typedef struct STableMetaInfo {
/* the structure for sql function in select clause */
typedef struct SSqlExpr {
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
int16_t interResBytes; // inter result buffer size
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.
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
int16_t interResBytes; // inter result buffer size
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.
} SSqlExpr;
typedef struct SColumnIndex {
......@@ -100,46 +101,27 @@ typedef struct SColumnIndex {
int16_t columnIndex;
} SColumnIndex;
typedef struct SFieldInfo {
int16_t numOfOutputCols; // number of column in result
int16_t numOfAlloc; // allocated size
TAOS_FIELD *pFields;
typedef struct SFieldSupInfo {
bool visible;
SArithExprInfo *pArithExprInfo;
SSqlExpr * pSqlExpr;
} SFieldSupInfo;
/*
* define if this column is belong to the queried result, it may be add by parser to faciliate
* the query process
*
* NOTE: these hidden columns always locate at the end of the output columns
*/
bool * pVisibleCols;
int32_t numOfHiddenCols; // the number of column not belongs to the queried result columns
SSqlFunctionExpr** pExpr; // used for aggregation arithmetic express,such as count(*)+count(*)
SSqlExpr** pSqlExpr;
typedef struct SFieldInfo {
int16_t numOfOutput; // number of column in result
SArray *pFields; // SArray<TAOS_FIELD>
SArray *pSupportInfo; // SArray<SFieldSupInfo>
} SFieldInfo;
typedef struct SSqlExprInfo {
int16_t numOfAlloc;
int16_t numOfExprs;
SSqlExpr** pExprs;
} SSqlExprInfo;
typedef struct SColumnBase {
typedef struct SColumn {
SColumnIndex colIndex;
int32_t numOfFilters;
SColumnFilterInfo *filterInfo;
} SColumnBase;
typedef struct SColumnBaseInfo {
int16_t numOfAlloc;
int16_t numOfCols;
SColumnBase *pColList;
} SColumnBaseInfo;
struct SLocalReducer;
} SColumn;
typedef struct SCond {
uint64_t uid;
int32_t len; // length of tag query condition data
int32_t len; // length of tag query condition data
char * cond;
} SCond;
......@@ -166,7 +148,7 @@ typedef struct STagCond {
SJoinInfo joinInfo;
// for different table, the query condition must be seperated
SArray* pCond;
SArray *pCond;
} STagCond;
typedef struct SParamInfo {
......@@ -207,7 +189,7 @@ typedef struct STableDataBlocks {
SParamInfo *params;
} STableDataBlocks;
typedef struct SDataBlockList {
typedef struct SDataBlockList { // todo remove
uint32_t nSize;
uint32_t nAlloc;
STableDataBlocks **pData;
......@@ -218,14 +200,14 @@ typedef struct SQueryInfo {
uint16_t type; // query/insert/import type
char slidingTimeUnit;
int64_t etime, stime;
STimeWindow window;
int64_t intervalTime; // aggregation time interval
int64_t slidingTime; // sliding window in mseconds
SSqlGroupbyExpr groupbyExpr; // group by tags info
SColumnBaseInfo colList;
SArray * colList; // SArray<SColumn*>
SFieldInfo fieldsInfo;
SSqlExprInfo exprsInfo;
SArray * exprsInfo; // SArray<SSqlExpr*>
SLimitVal limit;
SLimitVal slimit;
STagCond tagCond;
......@@ -242,19 +224,13 @@ typedef struct SQueryInfo {
int64_t prjOffset;
} SQueryInfo;
// data source from sql string or from file
enum {
DATA_FROM_SQL_STRING = 1,
DATA_FROM_DATA_FILE = 2,
};
typedef struct {
int command;
uint8_t msgType;
union {
bool existsCheck; // check if the table exists or not
bool autoCreated; // if the table is missing, on-the-fly create it. during getmeterMeta
bool autoCreated; // if the table is missing, on-the-fly create it. during getmeterMeta
int8_t dataSourceType; // load data from file or not
};
......@@ -271,11 +247,11 @@ typedef struct {
int32_t payloadLen;
SQueryInfo **pQueryInfo;
int32_t numOfClause;
SDataBlockList *pDataBlocks; // submit data blocks after parsing sql
char * curSql; // current sql, resume position of sql after parsing paused
void * pTableList; // referred table involved in sql
SDataBlockList *pDataBlocks; // submit data blocks after parsing sql
char * curSql; // current sql, resume position of sql after parsing paused
void * pTableList; // referred table involved in sql
// for parameter ('?') binding and batch processing
int32_t batchSize;
int32_t numOfParams;
......@@ -286,50 +262,48 @@ typedef struct SResRec {
int numOfTotal;
} SResRec;
struct STSBuf;
typedef struct {
int64_t numOfRows; // num of results in current retrieved
int64_t numOfTotal; // num of total results
int64_t numOfTotalInCurrentClause; // num of total result in current subclause
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;
char ** buffer; // Buffer used to put multibytes encoded using unicode (wchar_t)
SColumnIndex *pColumnIndex;
int64_t numOfRows; // num of results in current retrieved
int64_t numOfTotal; // num of total results
int64_t numOfTotalInCurrentClause; // num of total result in current subclause
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;
char ** buffer; // Buffer used to put multibytes encoded using unicode (wchar_t)
SColumnIndex * pColumnIndex;
struct SLocalReducer *pLocalReducer;
} SSqlRes;
typedef struct STscObj {
void * signature;
void * pTimer;
char mgmtIp[TSDB_USER_LEN];
uint16_t mgmtPort;
char user[TSDB_USER_LEN];
char pass[TSDB_KEY_LEN];
char acctId[TSDB_DB_NAME_LEN];
char db[TSDB_TABLE_ID_LEN];
char sversion[TSDB_VERSION_LEN];
char writeAuth : 1;
char superAuth : 1;
struct SSqlObj *pSql;
struct SSqlObj *pHb;
struct SSqlObj *sqlList;
void * signature;
void * pTimer;
char mgmtIp[TSDB_USER_LEN];
uint16_t mgmtPort;
char user[TSDB_USER_LEN];
char pass[TSDB_KEY_LEN];
char acctId[TSDB_DB_NAME_LEN];
char db[TSDB_TABLE_ID_LEN];
char sversion[TSDB_VERSION_LEN];
char writeAuth : 1;
char superAuth : 1;
struct SSqlObj * pSql;
struct SSqlObj * pHb;
struct SSqlObj * sqlList;
struct SSqlStream *streamList;
pthread_mutex_t mutex;
pthread_mutex_t mutex;
} STscObj;
typedef struct SSqlObj {
......@@ -337,23 +311,23 @@ typedef struct SSqlObj {
STscObj *pTscObj;
void (*fp)();
void (*fetchFp)();
void * param;
uint32_t ip;
short vnode;
int64_t stime;
uint32_t queryId;
void * pStream;
void * pSubscription;
char * sqlstr;
char retry;
char maxRetry;
SRpcIpSet ipList;
char freed : 4;
char listed : 4;
tsem_t rspSem;
SSqlCmd cmd;
SSqlRes res;
uint8_t numOfSubs;
void * param;
uint32_t ip;
short vnode;
int64_t stime;
uint32_t queryId;
void * pStream;
void * pSubscription;
char * sqlstr;
char retry;
char maxRetry;
SRpcIpSet ipList;
char freed : 4;
char listed : 4;
tsem_t rspSem;
SSqlCmd cmd;
SSqlRes res;
uint8_t numOfSubs;
struct SSqlObj **pSubs;
struct SSqlObj * prev, *next;
} SSqlObj;
......@@ -387,13 +361,10 @@ typedef struct SSqlStream {
} SSqlStream;
int32_t tscInitRpc(const char *user, const char *secret);
void tscInitMsgsFp();
// tscSql API
int tsParseSql(SSqlObj *pSql, bool multiVnodeInsertion);
void tscInitMsgsFp();
extern int (*tscBuildMsg[TSDB_SQL_MAX])(SSqlObj *pSql, SSqlInfo *pInfo);
void tscProcessMsgFromServer(SRpcMsg *rpcMsg);
int tscProcessSql(SSqlObj *pSql);
......@@ -406,12 +377,8 @@ int tscProcessLocalCmd(SSqlObj *pSql);
int tscCfgDynamicOptions(char *msg);
int taos_retrieve(TAOS_RES *res);
/*
* transfer function for metric query in stream computing, the function need to be change
* before send query message to vnode
*/
int32_t tscTansformSQLFunctionForSTableQuery(SQueryInfo *pQueryInfo);
void tscRestoreSQLFunctionForMetricQuery(SQueryInfo *pQueryInfo);
int32_t tscTansformSQLFuncForSTableQuery(SQueryInfo *pQueryInfo);
void tscRestoreSQLFuncForSTableQuery(SQueryInfo *pQueryInfo);
int32_t tscCreateResPointerInfo(SSqlRes *pRes, SQueryInfo *pQueryInfo);
void tscDestroyResPointerInfo(SSqlRes *pRes);
......@@ -441,10 +408,13 @@ void tscFreeSqlObj(SSqlObj *pObj);
void tscCloseTscObj(STscObj *pObj);
void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const char* sqlstr, size_t sqlLen);
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);
void doAsyncQuery(STscObj *pObj, SSqlObj *pSql, void (*fp)(), void *param, const char *sqlstr, size_t sqlLen);
void tscProcessMultiVnodesInsertFromFile(SSqlObj *pSql);
void tscKillMetricQuery(SSqlObj *pSql);
void tscKillSTableQuery(SSqlObj *pSql);
void tscInitResObjForLocalQuery(SSqlObj *pObj, int32_t numOfRes, int32_t rowLen);
bool tscIsUpdateQuery(STscObj *pObj);
bool tscHasReachLimitation(SQueryInfo *pQueryInfo, SSqlRes *pRes);
......@@ -453,19 +423,22 @@ char *tscGetErrorMsgPayload(SSqlCmd *pCmd);
int32_t tscInvalidSQLErrMsg(char *msg, const char *additionalInfo, const char *sql);
void tscQueueAsyncFreeResult(SSqlObj *pSql);
int32_t tscToSQLCmd(SSqlObj* pSql, struct SSqlInfo* pInfo);
extern void * pVnodeConn;
extern void * pTscMgmtConn;
extern void * tscCacheHandle;
extern int slaveIndex;
extern void * tscTmr;
extern void * tscQhandle;
extern int tscKeepConn[];
extern int tsInsertHeadSize;
extern int tscNumOfThreads;
extern SRpcIpSet tscMgmtIpSet;
void tscQueueAsyncFreeResult(SSqlObj *pSql);
int32_t tscToSQLCmd(SSqlObj *pSql, struct SSqlInfo *pInfo);
char * tscGetResultColumnChr(SSqlRes *pRes, SQueryInfo *pQueryInfo, int32_t column);
extern void * pVnodeConn;
extern void * pTscMgmtConn;
extern void * tscCacheHandle;
extern int slaveIndex;
extern void * tscTmr;
extern void * tscQhandle;
extern int tscKeepConn[];
extern int tsInsertHeadSize;
extern int tscNumOfThreads;
extern SRpcIpSet tscMgmtIpSet;
extern int (*tscBuildMsg[TSDB_SQL_MAX])(SSqlObj *pSql, SSqlInfo *pInfo);
typedef void (*__async_cb_func_t)(void *param, TAOS_RES *tres, int numOfRows);
......
src/client/src/tscAsync.c
浏览文件 @ 863bda56
......@@ -287,9 +287,9 @@ void tscAsyncFetchSingleRowProxy(void *param, TAOS_RES *tres, int numOfRows) {
}
for (int i = 0; i < pCmd->numOfCols; ++i){
SSqlExpr* pExpr = pQueryInfo->fieldsInfo.pSqlExpr[i];
if (pExpr != NULL) {
pRes->tsrow[i] = TSC_GET_RESPTR_BASE(pRes, pQueryInfo, i) + pExpr->resBytes * pRes->row;
SFieldSupInfo* pSup = taosArrayGet(pQueryInfo->fieldsInfo.pSupportInfo, i);
if (pSup->pSqlExpr != NULL) {
// pRes->tsrow[i] = TSC_GET_RESPTR_BASE(pRes, pQueryInfo, i) + pSup->pSqlExpr->resBytes * pRes->row;
} else {
//todo add
}
......@@ -308,11 +308,12 @@ void tscProcessFetchRow(SSchedMsg *pMsg) {
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
for (int i = 0; i < pCmd->numOfCols; ++i) {
SSqlExpr* pExpr = pQueryInfo->fieldsInfo.pSqlExpr[i];
if (pExpr != NULL) {
pRes->tsrow[i] = TSC_GET_RESPTR_BASE(pRes, pQueryInfo, i) + pExpr->resBytes * pRes->row;
SFieldSupInfo* pSup = taosArrayGet(pQueryInfo->fieldsInfo.pSupportInfo, i);
if (pSup->pSqlExpr != NULL) {
pRes->tsrow[i] = tscGetResultColumnChr(pRes, pQueryInfo, i);
} else {
//todo add
// todo add
}
}
......@@ -332,7 +333,7 @@ void tscProcessAsyncRes(SSchedMsg *pMsg) {
int code = pRes->code;
// in case of async insert, restore the user specified callback function
bool shouldFree = tscShouldFreeAsyncSqlObj(pSql);
bool shouldFree = tscShouldBeFreed(pSql);
if (cmd == TSDB_SQL_INSERT) {
assert(pSql->fp != NULL);
......@@ -487,7 +488,7 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
*/
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
tscTansformSQLFunctionForSTableQuery(pQueryInfo);
tscTansformSQLFuncForSTableQuery(pQueryInfo);
tscIncStreamExecutionCount(pSql->pStream);
} else {
tscTrace("%p get tableMeta successfully", pSql);
......
src/client/src/tscFunctionImpl.c
浏览文件 @ 863bda56
......@@ -2940,7 +2940,7 @@ static void col_project_function_f(SQLFunctionCtx *pCtx, int32_t index) {
char *pData = GET_INPUT_CHAR_INDEX(pCtx, index);
memcpy(pCtx->aOutputBuf, pData, pCtx->inputBytes);
pCtx->aOutputBuf += pCtx->inputBytes/* * GET_FORWARD_DIRECTION_FACTOR(pCtx->order)*/;
pCtx->aOutputBuf += pCtx->inputBytes;
}
/**
......@@ -3297,7 +3297,7 @@ static void diff_function_f(SQLFunctionCtx *pCtx, int32_t index) {
char *arithmetic_callback_function(void *param, char *name, int32_t colId) {
SArithmeticSupport *pSupport = (SArithmeticSupport *)param;
SSqlFunctionExpr *pExpr = pSupport->pExpr;
SArithExprInfo *pExpr = pSupport->pArithExpr;
int32_t colIndex = -1;
for (int32_t i = 0; i < pExpr->binExprInfo.numOfCols; ++i) {
......@@ -3315,7 +3315,7 @@ static void arithmetic_function(SQLFunctionCtx *pCtx) {
GET_RES_INFO(pCtx)->numOfRes += pCtx->size;
SArithmeticSupport *sas = (SArithmeticSupport *)pCtx->param[1].pz;
tSQLBinaryExprCalcTraverse(sas->pExpr->binExprInfo.pBinExpr, pCtx->size, pCtx->aOutputBuf, sas, pCtx->order,
tSQLBinaryExprCalcTraverse(sas->pArithExpr->binExprInfo.pBinExpr, pCtx->size, pCtx->aOutputBuf, sas, pCtx->order,
arithmetic_callback_function);
pCtx->aOutputBuf += pCtx->outputBytes * pCtx->size;
......@@ -3327,10 +3327,10 @@ static void arithmetic_function_f(SQLFunctionCtx *pCtx, int32_t index) {
SArithmeticSupport *sas = (SArithmeticSupport *)pCtx->param[1].pz;
sas->offset = index;
tSQLBinaryExprCalcTraverse(sas->pExpr->binExprInfo.pBinExpr, 1, pCtx->aOutputBuf, sas, pCtx->order,
tSQLBinaryExprCalcTraverse(sas->pArithExpr->binExprInfo.pBinExpr, 1, pCtx->aOutputBuf, sas, pCtx->order,
arithmetic_callback_function);
pCtx->aOutputBuf += pCtx->outputBytes/* * GET_FORWARD_DIRECTION_FACTOR(pCtx->order)*/;
pCtx->aOutputBuf += pCtx->outputBytes;
}
#define LIST_MINMAX_N(ctx, minOutput, maxOutput, elemCnt, data, type, tsdbType, numOfNotNullElem) \
......
src/client/src/tscLocal.c
浏览文件 @ 863bda56
......@@ -130,13 +130,13 @@ static int32_t tscSetValueToResObj(SSqlObj *pSql, int32_t rowLen) {
SSchema *pSchema = tscGetTableSchema(pMeta);
for (int32_t i = 0; i < numOfRows; ++i) {
TAOS_FIELD *pField = tscFieldInfoGetField(pQueryInfo, 0);
TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, 0);
strncpy(pRes->data + tscFieldInfoGetOffset(pQueryInfo, 0) * totalNumOfRows + pField->bytes * i, pSchema[i].name,
TSDB_COL_NAME_LEN);
char *type = tDataTypeDesc[pSchema[i].type].aName;
pField = tscFieldInfoGetField(pQueryInfo, 1);
pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, 1);
strncpy(pRes->data + tscFieldInfoGetOffset(pQueryInfo, 1) * totalNumOfRows + pField->bytes * i, type, pField->bytes);
int32_t bytes = pSchema[i].bytes;
......@@ -144,10 +144,10 @@ static int32_t tscSetValueToResObj(SSqlObj *pSql, int32_t rowLen) {
bytes = bytes / TSDB_NCHAR_SIZE;
}
pField = tscFieldInfoGetField(pQueryInfo, 2);
pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, 2);
*(int32_t *)(pRes->data + tscFieldInfoGetOffset(pQueryInfo, 2) * totalNumOfRows + pField->bytes * i) = bytes;
pField = tscFieldInfoGetField(pQueryInfo, 3);
pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, 3);
if (i >= tscGetNumOfColumns(pMeta) && tscGetNumOfTags(pMeta) != 0) {
strncpy(pRes->data + tscFieldInfoGetOffset(pQueryInfo, 3) * totalNumOfRows + pField->bytes * i, "tag",
strlen("tag") + 1);
......@@ -162,18 +162,18 @@ static int32_t tscSetValueToResObj(SSqlObj *pSql, int32_t rowLen) {
char *pTagValue = tsGetTagsValue(pMeta);
for (int32_t i = numOfRows; i < totalNumOfRows; ++i) {
// field name
TAOS_FIELD *pField = tscFieldInfoGetField(pQueryInfo, 0);
TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, 0);
strncpy(pRes->data + tscFieldInfoGetOffset(pQueryInfo, 0) * totalNumOfRows + pField->bytes * i, pSchema[i].name,
TSDB_COL_NAME_LEN);
// type name
pField = tscFieldInfoGetField(pQueryInfo, 1);
pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, 1);
char *type = tDataTypeDesc[pSchema[i].type].aName;
strncpy(pRes->data + tscFieldInfoGetOffset(pQueryInfo, 1) * totalNumOfRows + pField->bytes * i, type, pField->bytes);
// type length
int32_t bytes = pSchema[i].bytes;
pField = tscFieldInfoGetField(pQueryInfo, 2);
pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, 2);
if (pSchema[i].type == TSDB_DATA_TYPE_NCHAR) {
bytes = bytes / TSDB_NCHAR_SIZE;
}
......@@ -181,7 +181,7 @@ static int32_t tscSetValueToResObj(SSqlObj *pSql, int32_t rowLen) {
*(int32_t *)(pRes->data + tscFieldInfoGetOffset(pQueryInfo, 2) * totalNumOfRows + pField->bytes * i) = bytes;
// tag value
pField = tscFieldInfoGetField(pQueryInfo, 3);
pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, 3);
char *target = pRes->data + tscFieldInfoGetOffset(pQueryInfo, 3) * totalNumOfRows + pField->bytes * i;
if (isNull(pTagValue, pSchema[i].type)) {
......@@ -236,31 +236,47 @@ static int32_t tscSetValueToResObj(SSqlObj *pSql, int32_t rowLen) {
static int32_t tscBuildMeterSchemaResultFields(SSqlObj *pSql, int32_t numOfCols, int32_t typeColLength,
int32_t noteColLength) {
int32_t rowLen = 0;
SSqlCmd *pCmd = &pSql->cmd;
pCmd->numOfCols = numOfCols;
SColumnIndex index = {0};
pSql->cmd.numOfCols = numOfCols;
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
pQueryInfo->order.order = TSDB_ORDER_ASC;
tscFieldInfoSetValue(&pQueryInfo->fieldsInfo, 0, TSDB_DATA_TYPE_BINARY, "Field", TSDB_COL_NAME_LEN);
TAOS_FIELD f = {.type = TSDB_DATA_TYPE_BINARY, .bytes = TSDB_COL_NAME_LEN};
strncpy(f.name, "Field", TSDB_COL_NAME_LEN);
SFieldSupInfo* pInfo = tscFieldInfoAppend(&pQueryInfo->fieldsInfo, &f);
pInfo->pSqlExpr = tscSqlExprAppend(pQueryInfo, TSDB_FUNC_TS_DUMMY, &index, TSDB_DATA_TYPE_BINARY, TSDB_COL_NAME_LEN, TSDB_COL_NAME_LEN);
rowLen += TSDB_COL_NAME_LEN;
tscFieldInfoSetValue(&pQueryInfo->fieldsInfo, 1, TSDB_DATA_TYPE_BINARY, "Type", typeColLength);
f.bytes = typeColLength;
f.type = TSDB_DATA_TYPE_BINARY;
strncpy(f.name, "Type", TSDB_COL_NAME_LEN);
pInfo = tscFieldInfoAppend(&pQueryInfo->fieldsInfo, &f);
pInfo->pSqlExpr = tscSqlExprAppend(pQueryInfo, TSDB_FUNC_TS_DUMMY, &index, TSDB_DATA_TYPE_BINARY, typeColLength, typeColLength);
rowLen += typeColLength;
tscFieldInfoSetValue(&pQueryInfo->fieldsInfo, 2, TSDB_DATA_TYPE_INT, "Length", sizeof(int32_t));
f.bytes = sizeof(int32_t);
f.type = TSDB_DATA_TYPE_INT;
strncpy(f.name, "Length", TSDB_COL_NAME_LEN);
pInfo = tscFieldInfoAppend(&pQueryInfo->fieldsInfo, &f);
pInfo->pSqlExpr = tscSqlExprAppend(pQueryInfo, TSDB_FUNC_TS_DUMMY, &index, TSDB_DATA_TYPE_INT, sizeof(int32_t), sizeof(int32_t));
rowLen += sizeof(int32_t);
tscFieldInfoSetValue(&pQueryInfo->fieldsInfo, 3, TSDB_DATA_TYPE_BINARY, "Note", noteColLength);
rowLen += noteColLength;
f.bytes = noteColLength;
f.type = TSDB_DATA_TYPE_BINARY;
strncpy(f.name, "Note", TSDB_COL_NAME_LEN);
//set the sqlexpr part
SColumnIndex index = {0};
pQueryInfo->fieldsInfo.pSqlExpr[0] = tscSqlExprInsert(pQueryInfo, 0, TSDB_FUNC_TS_DUMMY, &index, TSDB_DATA_TYPE_BINARY, TSDB_COL_NAME_LEN, TSDB_COL_NAME_LEN);
pQueryInfo->fieldsInfo.pSqlExpr[1] = tscSqlExprInsert(pQueryInfo, 1, TSDB_FUNC_TS_DUMMY, &index, TSDB_DATA_TYPE_BINARY, typeColLength, typeColLength);
pQueryInfo->fieldsInfo.pSqlExpr[2] = tscSqlExprInsert(pQueryInfo, 2, TSDB_FUNC_TS_DUMMY, &index, TSDB_DATA_TYPE_INT, sizeof(int32_t), sizeof(int32_t));
pQueryInfo->fieldsInfo.pSqlExpr[3] = tscSqlExprInsert(pQueryInfo, 3, TSDB_FUNC_TS_DUMMY, &index, TSDB_DATA_TYPE_BINARY, noteColLength, noteColLength);
pInfo = tscFieldInfoAppend(&pQueryInfo->fieldsInfo, &f);
pInfo->pSqlExpr = tscSqlExprAppend(pQueryInfo, TSDB_FUNC_TS_DUMMY, &index, TSDB_DATA_TYPE_BINARY, noteColLength, noteColLength);
rowLen += noteColLength;
return rowLen;
}
......@@ -280,7 +296,7 @@ static int32_t tscProcessDescribeTable(SSqlObj *pSql) {
int32_t rowLen =
tscBuildMeterSchemaResultFields(pSql, NUM_OF_DESCRIBE_TABLE_COLUMNS, TYPE_COLUMN_LENGTH, note_field_length);
tscFieldInfoCalOffset(pQueryInfo);
tscFieldInfoUpdateOffset(pQueryInfo);
return tscSetValueToResObj(pSql, rowLen);
}
......@@ -310,7 +326,7 @@ static int tscBuildMetricTagProjectionResult(SSqlObj *pSql) {
}
int32_t totalNumOfResults = pMetricMeta->numOfTables;
int32_t rowLen = tscGetResRowLength(pQueryInfo);
int32_t rowLen = tscGetResRowLength(pQueryInfo->exprsInfo);
tscInitResObjForLocalQuery(pSql, totalNumOfResults, rowLen);
......@@ -321,7 +337,7 @@ static int tscBuildMetricTagProjectionResult(SSqlObj *pSql) {
for (int32_t j = 0; j < pSidList->numOfSids; ++j) {
STableIdInfo *pSidExt = tscGetMeterSidInfo(pSidList, j);
for (int32_t k = 0; k < pQueryInfo->fieldsInfo.numOfOutputCols; ++k) {
for (int32_t k = 0; k < pQueryInfo->fieldsInfo.numOfOutput; ++k) {
SColIndex *pColIndex = &tscSqlExprGet(pQueryInfo, k)->colInfo;
int16_t offsetId = pColIndex->colIdx;
......@@ -329,7 +345,7 @@ static int tscBuildMetricTagProjectionResult(SSqlObj *pSql) {
assert(0);
char * val = NULL;//pSidExt->tags + vOffset[offsetId];
TAOS_FIELD *pField = tscFieldInfoGetField(pQueryInfo, k);
TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, k);
memcpy(pRes->data + tscFieldInfoGetOffset(pQueryInfo, k) * totalNumOfResults + pField->bytes * rowIdx, val,
(size_t)pField->bytes);
......@@ -350,17 +366,17 @@ static int tscBuildMetricTagSqlFunctionResult(SSqlObj *pSql) {
#if 0
SSuperTableMeta *pMetricMeta = tscGetMetaInfo(pQueryInfo, 0)->pMetricMeta;
int32_t totalNumOfResults = 1; // count function only produce one result
int32_t rowLen = tscGetResRowLength(pQueryInfo);
int32_t rowLen = tscGetResRowLength(pQueryInfo->exprsInfo);
tscInitResObjForLocalQuery(pSql, totalNumOfResults, rowLen);
int32_t rowIdx = 0;
for (int32_t i = 0; i < totalNumOfResults; ++i) {
for (int32_t k = 0; k < pQueryInfo->fieldsInfo.numOfOutputCols; ++k) {
for (int32_t k = 0; k < pQueryInfo->fieldsInfo.numOfOutput; ++k) {
SSqlExpr *pExpr = tscSqlExprGet(pQueryInfo, i);
if (pExpr->colInfo.colIdx == -1 && pExpr->functionId == TSDB_FUNC_COUNT) {
TAOS_FIELD *pField = tscFieldInfoGetField(pQueryInfo, k);
TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, k);
memcpy(pRes->data + tscFieldInfoGetOffset(pQueryInfo, i) * totalNumOfResults + pField->bytes * rowIdx,
&pMetricMeta->numOfTables, sizeof(pMetricMeta->numOfTables));
......@@ -388,7 +404,7 @@ static int tscProcessQueryTags(SSqlObj *pSql) {
return pSql->res.code;
}
SSqlExpr *pExpr = tscSqlExprGet(pQueryInfo, 0);
SSqlExpr *pExpr = taosArrayGetP(pQueryInfo->exprsInfo, 0);
if (pExpr->functionId == TSDB_FUNC_COUNT) {
return tscBuildMetricTagSqlFunctionResult(pSql);
} else {
......@@ -399,7 +415,7 @@ static int tscProcessQueryTags(SSqlObj *pSql) {
static void tscProcessCurrentUser(SSqlObj *pSql) {
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
SSqlExpr* pExpr = tscSqlExprGet(pQueryInfo, 0);
SSqlExpr* pExpr = taosArrayGetP(pQueryInfo->exprsInfo, 0);
tscSetLocalQueryResult(pSql, pSql->pTscObj->user, pExpr->aliasName, TSDB_USER_LEN);
}
......@@ -414,7 +430,7 @@ static void tscProcessCurrentDB(SSqlObj *pSql) {
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
SSqlExpr* pExpr = tscSqlExprGet(pQueryInfo, 0);
SSqlExpr* pExpr = taosArrayGetP(pQueryInfo->exprsInfo, 0);
tscSetLocalQueryResult(pSql, db, pExpr->aliasName, TSDB_DB_NAME_LEN);
}
......@@ -422,14 +438,14 @@ static void tscProcessServerVer(SSqlObj *pSql) {
const char* v = pSql->pTscObj->sversion;
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
SSqlExpr* pExpr = tscSqlExprGet(pQueryInfo, 0);
SSqlExpr* pExpr = taosArrayGetP(pQueryInfo->exprsInfo, 0);
tscSetLocalQueryResult(pSql, v, pExpr->aliasName, tListLen(pSql->pTscObj->sversion));
}
static void tscProcessClientVer(SSqlObj *pSql) {
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
SSqlExpr* pExpr = tscSqlExprGet(pQueryInfo, 0);
SSqlExpr* pExpr = taosArrayGetP(pQueryInfo->exprsInfo, 0);
tscSetLocalQueryResult(pSql, version, pExpr->aliasName, strlen(version));
}
......@@ -449,7 +465,7 @@ static void tscProcessServStatus(SSqlObj *pSql) {
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
SSqlExpr* pExpr = tscSqlExprGet(pQueryInfo, 0);
SSqlExpr* pExpr = taosArrayGetP(pQueryInfo->exprsInfo, 0);
tscSetLocalQueryResult(pSql, "1", pExpr->aliasName, 2);
}
......@@ -462,13 +478,16 @@ void tscSetLocalQueryResult(SSqlObj *pSql, const char *val, const char *columnNa
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
pQueryInfo->order.order = TSDB_ORDER_ASC;
tscClearFieldInfo(&pQueryInfo->fieldsInfo);
tscFieldInfoClear(&pQueryInfo->fieldsInfo);
TAOS_FIELD f = tscCreateField(TSDB_DATA_TYPE_BINARY, columnName, valueLength);
tscFieldInfoAppend(&pQueryInfo->fieldsInfo, &f);
tscFieldInfoSetValue(&pQueryInfo->fieldsInfo, 0, TSDB_DATA_TYPE_BINARY, columnName, valueLength);
tscInitResObjForLocalQuery(pSql, 1, valueLength);
TAOS_FIELD *pField = tscFieldInfoGetField(pQueryInfo, 0);
pQueryInfo->fieldsInfo.pSqlExpr[0] = pQueryInfo->exprsInfo.pExprs[0];
TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, 0);
SFieldSupInfo* pInfo = tscFieldInfoGetSupp(&pQueryInfo->fieldsInfo, 0);
pInfo->pSqlExpr = taosArrayGetP(pQueryInfo->exprsInfo, 0);
strncpy(pRes->data, val, pField->bytes);
}
......
src/client/src/tscParseInsert.c
浏览文件 @ 863bda56
......@@ -613,7 +613,7 @@ static void tsSetBlockInfo(SSubmitBlk *pBlocks, const STableMeta *pTableMeta, in
}
// data block is disordered, sort it in ascending order
void sortRemoveDuplicates(STableDataBlocks *dataBuf) {
void tscSortRemoveDataBlockDupRows(STableDataBlocks *dataBuf) {
SSubmitBlk *pBlocks = (SSubmitBlk *)dataBuf->pData;
// size is less than the total size, since duplicated rows may be removed yet.
......@@ -779,7 +779,7 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql) {
}
STableMetaInfo *pSTableMeterMetaInfo = tscGetMetaInfo(pQueryInfo, STABLE_INDEX);
setMeterID(pSTableMeterMetaInfo, &sToken, pSql);
tscSetTableId(pSTableMeterMetaInfo, &sToken, pSql);
strncpy(pTag->name, pSTableMeterMetaInfo->name, TSDB_TABLE_ID_LEN);
code = tscGetTableMeta(pSql, pSTableMeterMetaInfo);
......@@ -922,7 +922,7 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql) {
return tscInvalidSQLErrMsg(pCmd->payload, "invalid table name", *sqlstr);
}
int32_t ret = setMeterID(pTableMetaInfo, &tableToken, pSql);
int32_t ret = tscSetTableId(pTableMetaInfo, &tableToken, pSql);
if (ret != TSDB_CODE_SUCCESS) {
return ret;
}
......@@ -1059,7 +1059,7 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
goto _error_clean;
}
if ((code = setMeterID(pTableMetaInfo, &sToken, pSql)) != TSDB_CODE_SUCCESS) {
if ((code = tscSetTableId(pTableMetaInfo, &sToken, pSql)) != TSDB_CODE_SUCCESS) {
goto _error_clean;
}
......
src/client/src/tscSQLParser.c
浏览文件 @ 863bda56
此差异已折叠。
src/client/src/tscSchemaUtil.c
浏览文件 @ 863bda56
......@@ -161,9 +161,9 @@ STableMeta* tscCreateTableMetaFromMsg(STableMetaMsg* pTableMetaMsg, size_t* size
pTableMeta->tableType = pTableMetaMsg->tableType;
pTableMeta->tableInfo = (STableComInfo) {
.numOfTags = pTableMetaMsg->numOfTags,
.numOfTags = pTableMetaMsg->numOfTags,
.precision = pTableMetaMsg->precision,
.numOfColumns = pTableMetaMsg->numOfColumns,
.precision = pTableMetaMsg->precision
};
pTableMeta->sid = pTableMetaMsg->sid;
......@@ -172,7 +172,7 @@ STableMeta* tscCreateTableMetaFromMsg(STableMetaMsg* pTableMetaMsg, size_t* size
memcpy(pTableMeta->schema, pTableMetaMsg->schema, schemaSize);
int32_t numOfTotalCols = pTableMeta->tableInfo.numOfColumns + pTableMeta->tableInfo.numOfTags;
int32_t numOfTotalCols = pTableMeta->tableInfo.numOfColumns;
for(int32_t i = 0; i < numOfTotalCols; ++i) {
pTableMeta->tableInfo.rowSize += pTableMeta->schema[i].bytes;
}
......
src/client/src/tscSecondaryMerge.c
浏览文件 @ 863bda56
......@@ -61,7 +61,9 @@ static void tscInitSqlContext(SSqlCmd *pCmd, SSqlRes *pRes, SLocalReducer *pRedu
* merge requirement. So, the final result in pRes structure is formatted in accordance with the pCmd object.
*/
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
for (int32_t i = 0; i < pQueryInfo->exprsInfo.numOfExprs; ++i) {
size_t size = tscSqlExprNumOfExprs(pQueryInfo);
for (int32_t i = 0; i < size; ++i) {
SQLFunctionCtx *pCtx = &pReducer->pCtx[i];
SSqlExpr * pExpr = tscSqlExprGet(pQueryInfo, i);
......@@ -108,10 +110,10 @@ static void tscInitSqlContext(SSqlCmd *pCmd, SSqlRes *pRes, SLocalReducer *pRedu
int16_t n = 0;
int16_t tagLen = 0;
SQLFunctionCtx **pTagCtx = calloc(pQueryInfo->fieldsInfo.numOfOutputCols, POINTER_BYTES);
SQLFunctionCtx **pTagCtx = calloc(pQueryInfo->fieldsInfo.numOfOutput, POINTER_BYTES);
SQLFunctionCtx *pCtx = NULL;
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutputCols; ++i) {
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
SSqlExpr *pExpr = tscSqlExprGet(pQueryInfo, i);
if (pExpr->functionId == TSDB_FUNC_TAG_DUMMY || pExpr->functionId == TSDB_FUNC_TS_DUMMY) {
tagLen += pExpr->resBytes;
......@@ -254,12 +256,13 @@ void tscCreateLocalReducer(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrd
// the input data format follows the old format, but output in a new format.
// so, all the input must be parsed as old format
pReducer->pCtx = (SQLFunctionCtx *)calloc(pQueryInfo->exprsInfo.numOfExprs, sizeof(SQLFunctionCtx));
size_t size = tscSqlExprNumOfExprs(pQueryInfo);
pReducer->pCtx = (SQLFunctionCtx *)calloc(size, sizeof(SQLFunctionCtx));
pReducer->rowSize = pMemBuffer[0]->nElemSize;
tscRestoreSQLFunctionForMetricQuery(pQueryInfo);
tscFieldInfoCalOffset(pQueryInfo);
tscRestoreSQLFuncForSTableQuery(pQueryInfo);
tscFieldInfoUpdateOffset(pQueryInfo);
if (pReducer->rowSize > pMemBuffer[0]->pageSize) {
assert(false); // todo fixed row size is larger than the minimum page size;
......@@ -279,7 +282,7 @@ void tscCreateLocalReducer(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrd
pReducer->nResultBufSize = pMemBuffer[0]->pageSize * 16;
pReducer->pResultBuf = (tFilePage *)calloc(1, pReducer->nResultBufSize + sizeof(tFilePage));
int32_t finalRowLength = tscGetResRowLength(pQueryInfo);
int32_t finalRowLength = tscGetResRowLength(pQueryInfo->exprsInfo);
pReducer->resColModel = finalmodel;
pReducer->resColModel->capacity = pReducer->nResultBufSize / finalRowLength;
assert(finalRowLength <= pReducer->rowSize);
......@@ -301,7 +304,7 @@ void tscCreateLocalReducer(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrd
}
pReducer->pTempBuffer->numOfElems = 0;
pReducer->pResInfo = calloc((size_t)pQueryInfo->exprsInfo.numOfExprs, sizeof(SResultInfo));
pReducer->pResInfo = calloc(size, sizeof(SResultInfo));
tscCreateResPointerInfo(pRes, pQueryInfo);
tscInitSqlContext(pCmd, pRes, pReducer, pDesc);
......@@ -324,7 +327,7 @@ void tscCreateLocalReducer(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrd
STableComInfo tinfo = tscGetTableInfo(pTableMetaInfo->pTableMeta);
int16_t prec = tinfo.precision;
int64_t stime = (pQueryInfo->stime < pQueryInfo->etime) ? pQueryInfo->stime : pQueryInfo->etime;
int64_t stime = (pQueryInfo->window.skey < pQueryInfo->window.ekey) ? pQueryInfo->window.skey : pQueryInfo->window.ekey;
int64_t revisedSTime =
taosGetIntervalStartTimestamp(stime, pQueryInfo->intervalTime, pQueryInfo->slidingTimeUnit, prec);
......@@ -332,7 +335,7 @@ void tscCreateLocalReducer(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrd
taosInitInterpoInfo(pInterpoInfo, pQueryInfo->order.order, revisedSTime, pQueryInfo->groupbyExpr.numOfGroupCols,
pReducer->rowSize);
int32_t startIndex = pQueryInfo->fieldsInfo.numOfOutputCols - pQueryInfo->groupbyExpr.numOfGroupCols;
int32_t startIndex = pQueryInfo->fieldsInfo.numOfOutput - pQueryInfo->groupbyExpr.numOfGroupCols;
if (pQueryInfo->groupbyExpr.numOfGroupCols > 0) {
pInterpoInfo->pTags[0] = (char *)pInterpoInfo->pTags + POINTER_BYTES * pQueryInfo->groupbyExpr.numOfGroupCols;
......@@ -462,7 +465,7 @@ void tscDestroyLocalReducer(SSqlObj *pSql) {
taosDestoryInterpoInfo(&pLocalReducer->interpolationInfo);
if (pLocalReducer->pCtx != NULL) {
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutputCols; ++i) {
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
SQLFunctionCtx *pCtx = &pLocalReducer->pCtx[i];
tVariantDestroy(&pCtx->tag);
......@@ -480,7 +483,7 @@ void tscDestroyLocalReducer(SSqlObj *pSql) {
tfree(pLocalReducer->pResultBuf);
if (pLocalReducer->pResInfo != NULL) {
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutputCols; ++i) {
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
tfree(pLocalReducer->pResInfo[i].interResultBuf);
}
......@@ -532,7 +535,7 @@ static int32_t createOrderDescriptor(tOrderDescriptor **pOrderDesc, SSqlCmd *pCm
}
if (numOfGroupByCols > 0) {
int32_t startCols = pQueryInfo->fieldsInfo.numOfOutputCols - pQueryInfo->groupbyExpr.numOfGroupCols;
int32_t startCols = pQueryInfo->fieldsInfo.numOfOutput - pQueryInfo->groupbyExpr.numOfGroupCols;
// tags value locate at the last columns
for (int32_t i = 0; i < pQueryInfo->groupbyExpr.numOfGroupCols; ++i) {
......@@ -612,8 +615,10 @@ int32_t tscLocalReducerEnvCreate(SSqlObj *pSql, tExtMemBuffer ***pMemBuffer, tOr
pRes->code = TSDB_CODE_CLI_OUT_OF_MEMORY;
return pRes->code;
}
pSchema = (SSchema *)calloc(1, sizeof(SSchema) * pQueryInfo->exprsInfo.numOfExprs);
size_t size = tscSqlExprNumOfExprs(pQueryInfo);
pSchema = (SSchema *)calloc(1, sizeof(SSchema) * size);
if (pSchema == NULL) {
tscError("%p failed to allocate memory", pSql);
pRes->code = TSDB_CODE_CLI_OUT_OF_MEMORY;
......@@ -621,7 +626,7 @@ int32_t tscLocalReducerEnvCreate(SSqlObj *pSql, tExtMemBuffer ***pMemBuffer, tOr
}
int32_t rlen = 0;
for (int32_t i = 0; i < pQueryInfo->exprsInfo.numOfExprs; ++i) {
for (int32_t i = 0; i < size; ++i) {
SSqlExpr *pExpr = tscSqlExprGet(pQueryInfo, i);
pSchema[i].bytes = pExpr->resBytes;
......@@ -634,8 +639,8 @@ int32_t tscLocalReducerEnvCreate(SSqlObj *pSql, tExtMemBuffer ***pMemBuffer, tOr
if (rlen != 0) {
capacity = nBufferSizes / rlen;
}
pModel = createColumnModel(pSchema, pQueryInfo->exprsInfo.numOfExprs, capacity);
pModel = createColumnModel(pSchema, size, capacity);
size_t numOfSubs = pTableMetaInfo->vgroupList->numOfVgroups;
for (int32_t i = 0; i < numOfSubs; ++i) {
......@@ -649,8 +654,8 @@ int32_t tscLocalReducerEnvCreate(SSqlObj *pSql, tExtMemBuffer ***pMemBuffer, tOr
}
// final result depends on the fields number
memset(pSchema, 0, sizeof(SSchema) * pQueryInfo->exprsInfo.numOfExprs);
for (int32_t i = 0; i < pQueryInfo->exprsInfo.numOfExprs; ++i) {
memset(pSchema, 0, sizeof(SSchema) * size);
for (int32_t i = 0; i < size; ++i) {
SSqlExpr *pExpr = tscSqlExprGet(pQueryInfo, i);
SSchema *p1 = tscGetTableColumnSchema(pTableMetaInfo->pTableMeta, pExpr->colInfo.colIndex);
......@@ -683,8 +688,8 @@ int32_t tscLocalReducerEnvCreate(SSqlObj *pSql, tExtMemBuffer ***pMemBuffer, tOr
pSchema[i].bytes = bytes;
strcpy(pSchema[i].name, pModel->pFields[i].field.name);
}
*pFinalModel = createColumnModel(pSchema, pQueryInfo->exprsInfo.numOfExprs, capacity);
*pFinalModel = createColumnModel(pSchema, size, capacity);
tfree(pSchema);
return TSDB_CODE_SUCCESS;
......@@ -782,7 +787,7 @@ void savePrevRecordAndSetupInterpoInfo(SLocalReducer *pLocalReducer, SQueryInfo
STableComInfo tinfo = tscGetTableInfo(pTableMetaInfo->pTableMeta);
int16_t prec = tinfo.precision;
int64_t stime = (pQueryInfo->stime < pQueryInfo->etime) ? pQueryInfo->stime : pQueryInfo->etime;
int64_t stime = (pQueryInfo->window.skey < pQueryInfo->window.ekey) ? pQueryInfo->window.skey : pQueryInfo->window.ekey;
int64_t revisedSTime =
taosGetIntervalStartTimestamp(stime, pQueryInfo->intervalTime, pQueryInfo->slidingTimeUnit, prec);
......@@ -800,7 +805,7 @@ void savePrevRecordAndSetupInterpoInfo(SLocalReducer *pLocalReducer, SQueryInfo
// static void reversedCopyResultToDstBuf(SQueryInfo* pQueryInfo, SSqlRes *pRes, tFilePage *pFinalDataPage) {
//
// for (int32_t i = 0; i < pQueryInfo->exprsInfo.numOfExprs; ++i) {
// TAOS_FIELD *pField = tscFieldInfoGetField(pQueryInfo, i);
// TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
//
// int32_t offset = tscFieldInfoGetOffset(pQueryInfo, i);
// char * src = pFinalDataPage->data + (pRes->numOfRows - 1) * pField->bytes + pRes->numOfRows * offset;
......@@ -817,8 +822,10 @@ void savePrevRecordAndSetupInterpoInfo(SLocalReducer *pLocalReducer, SQueryInfo
static void reversedCopyFromInterpolationToDstBuf(SQueryInfo *pQueryInfo, SSqlRes *pRes, tFilePage **pResPages,
SLocalReducer *pLocalReducer) {
assert(0);
for (int32_t i = 0; i < pQueryInfo->exprsInfo.numOfExprs; ++i) {
TAOS_FIELD *pField = tscFieldInfoGetField(pQueryInfo, i);
size_t size = tscSqlExprNumOfExprs(pQueryInfo);
for (int32_t i = 0; i < size; ++i) {
TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
int32_t offset = tscFieldInfoGetOffset(pQueryInfo, i);
assert(offset == getColumnModelOffset(pLocalReducer->resColModel, i));
......@@ -894,7 +901,7 @@ static void doInterpolateResult(SSqlObj *pSql, SLocalReducer *pLocalReducer, boo
savePrevRecordAndSetupInterpoInfo(pLocalReducer, pQueryInfo, &pLocalReducer->interpolationInfo);
}
int32_t rowSize = tscGetResRowLength(pQueryInfo);
int32_t rowSize = tscGetResRowLength(pQueryInfo->exprsInfo);
memcpy(pRes->data, pFinalDataPage->data, pRes->numOfRows * rowSize);
pFinalDataPage->numOfElems = 0;
......@@ -905,18 +912,18 @@ static void doInterpolateResult(SSqlObj *pSql, SLocalReducer *pLocalReducer, boo
SInterpolationInfo *pInterpoInfo = &pLocalReducer->interpolationInfo;
int64_t actualETime = (pQueryInfo->stime < pQueryInfo->etime) ? pQueryInfo->etime : pQueryInfo->stime;
int64_t actualETime = (pQueryInfo->window.skey < pQueryInfo->window.ekey) ? pQueryInfo->window.ekey : pQueryInfo->window.skey;
tFilePage **pResPages = malloc(POINTER_BYTES * pQueryInfo->fieldsInfo.numOfOutputCols);
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutputCols; ++i) {
TAOS_FIELD *pField = tscFieldInfoGetField(pQueryInfo, i);
tFilePage **pResPages = malloc(POINTER_BYTES * pQueryInfo->fieldsInfo.numOfOutput);
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
pResPages[i] = calloc(1, sizeof(tFilePage) + pField->bytes * pLocalReducer->resColModel->capacity);
}
char ** srcData = (char **)malloc((POINTER_BYTES + sizeof(int32_t)) * pQueryInfo->fieldsInfo.numOfOutputCols);
int32_t *functions = (int32_t *)((char *)srcData + pQueryInfo->fieldsInfo.numOfOutputCols * sizeof(void *));
char ** srcData = (char **)malloc((POINTER_BYTES + sizeof(int32_t)) * pQueryInfo->fieldsInfo.numOfOutput);
int32_t *functions = (int32_t *)((char *)srcData + pQueryInfo->fieldsInfo.numOfOutput * sizeof(void *));
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutputCols; ++i) {
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
srcData[i] =
pLocalReducer->pBufForInterpo + tscFieldInfoGetOffset(pQueryInfo, i) * pInterpoInfo->numOfRawDataInRows;
functions[i] = tscSqlExprGet(pQueryInfo, i)->functionId;
......@@ -943,8 +950,8 @@ static void doInterpolateResult(SSqlObj *pSql, SLocalReducer *pLocalReducer, boo
newRows -= pQueryInfo->limit.offset;
if (pQueryInfo->limit.offset > 0) {
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutputCols; ++i) {
TAOS_FIELD *pField = tscFieldInfoGetField(pQueryInfo, i);
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
memmove(pResPages[i]->data, pResPages[i]->data + pField->bytes * pQueryInfo->limit.offset,
newRows * pField->bytes);
}
......@@ -992,8 +999,8 @@ static void doInterpolateResult(SSqlObj *pSql, SLocalReducer *pLocalReducer, boo
}
if (pQueryInfo->order.order == TSDB_ORDER_ASC) {
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutputCols; ++i) {
TAOS_FIELD *pField = tscFieldInfoGetField(pQueryInfo, i);
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
int16_t offset = getColumnModelOffset(pLocalReducer->resColModel, i);
memcpy(pRes->data + offset * pRes->numOfRows, pResPages[i]->data, pField->bytes * pRes->numOfRows);
}
......@@ -1003,7 +1010,7 @@ static void doInterpolateResult(SSqlObj *pSql, SLocalReducer *pLocalReducer, boo
}
pFinalDataPage->numOfElems = 0;
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutputCols; ++i) {
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
tfree(pResPages[i]);
}
tfree(pResPages);
......@@ -1030,8 +1037,9 @@ static void savePreviousRow(SLocalReducer *pLocalReducer, tFilePage *tmpBuffer)
static void doExecuteSecondaryMerge(SSqlCmd *pCmd, SLocalReducer *pLocalReducer, bool needInit) {
// the tag columns need to be set before all functions execution
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
size_t size = tscSqlExprNumOfExprs(pQueryInfo);
for (int32_t j = 0; j < pQueryInfo->exprsInfo.numOfExprs; ++j) {
for (int32_t j = 0; j < size; ++j) {
SSqlExpr * pExpr = tscSqlExprGet(pQueryInfo, j);
SQLFunctionCtx *pCtx = &pLocalReducer->pCtx[j];
......@@ -1051,7 +1059,7 @@ static void doExecuteSecondaryMerge(SSqlCmd *pCmd, SLocalReducer *pLocalReducer,
}
}
for (int32_t j = 0; j < pQueryInfo->exprsInfo.numOfExprs; ++j) {
for (int32_t j = 0; j < size; ++j) {
int32_t functionId = tscSqlExprGet(pQueryInfo, j)->functionId;
if (functionId == TSDB_FUNC_TAG_DUMMY || functionId == TSDB_FUNC_TS_DUMMY) {
continue;
......@@ -1071,8 +1079,9 @@ static void handleUnprocessedRow(SSqlCmd *pCmd, SLocalReducer *pLocalReducer, tF
static int64_t getNumOfResultLocal(SQueryInfo *pQueryInfo, SQLFunctionCtx *pCtx) {
int64_t maxOutput = 0;
for (int32_t j = 0; j < pQueryInfo->exprsInfo.numOfExprs; ++j) {
size_t size = tscSqlExprNumOfExprs(pQueryInfo);
for (int32_t j = 0; j < size; ++j) {
// SSqlExpr* pExpr = pQueryInfo->fieldsInfo.pSqlExpr[j];
// if (pExpr == NULL) {
// assert(pQueryInfo->fieldsInfo.pExpr[j] != NULL);
......@@ -1107,7 +1116,9 @@ static int64_t getNumOfResultLocal(SQueryInfo *pQueryInfo, SQLFunctionCtx *pCtx)
*/
static void fillMultiRowsOfTagsVal(SQueryInfo *pQueryInfo, int32_t numOfRes, SLocalReducer *pLocalReducer) {
int32_t maxBufSize = 0; // find the max tags column length to prepare the buffer
for (int32_t k = 0; k < pQueryInfo->exprsInfo.numOfExprs; ++k) {
size_t size = tscSqlExprNumOfExprs(pQueryInfo);
for (int32_t k = 0; k < size; ++k) {
SSqlExpr *pExpr = tscSqlExprGet(pQueryInfo, k);
if (maxBufSize < pExpr->resBytes && pExpr->functionId == TSDB_FUNC_TAG) {
maxBufSize = pExpr->resBytes;
......@@ -1117,7 +1128,7 @@ static void fillMultiRowsOfTagsVal(SQueryInfo *pQueryInfo, int32_t numOfRes, SLo
assert(maxBufSize >= 0);
char *buf = malloc((size_t)maxBufSize);
for (int32_t k = 0; k < pQueryInfo->exprsInfo.numOfExprs; ++k) {
for (int32_t k = 0; k < size; ++k) {
SSqlExpr *pExpr = tscSqlExprGet(pQueryInfo, k);
if (pExpr->functionId != TSDB_FUNC_TAG) {
continue;
......@@ -1139,7 +1150,9 @@ static void fillMultiRowsOfTagsVal(SQueryInfo *pQueryInfo, int32_t numOfRes, SLo
}
int32_t finalizeRes(SQueryInfo *pQueryInfo, SLocalReducer *pLocalReducer) {
for (int32_t k = 0; k < pQueryInfo->exprsInfo.numOfExprs; ++k) {
size_t size = tscSqlExprNumOfExprs(pQueryInfo);
for (int32_t k = 0; k < size; ++k) {
SSqlExpr *pExpr = tscSqlExprGet(pQueryInfo, k);
aAggs[pExpr->functionId].xFinalize(&pLocalReducer->pCtx[k]);
}
......@@ -1242,7 +1255,7 @@ bool doGenerateFinalResults(SSqlObj *pSql, SLocalReducer *pLocalReducer, bool no
#endif
SInterpolationInfo *pInterpoInfo = &pLocalReducer->interpolationInfo;
int32_t startIndex = pQueryInfo->fieldsInfo.numOfOutputCols - pQueryInfo->groupbyExpr.numOfGroupCols;
int32_t startIndex = pQueryInfo->fieldsInfo.numOfOutput - pQueryInfo->groupbyExpr.numOfGroupCols;
for (int32_t i = 0; i < pQueryInfo->groupbyExpr.numOfGroupCols; ++i) {
int16_t offset = getColumnModelOffset(pModel, startIndex + i);
......@@ -1258,7 +1271,7 @@ bool doGenerateFinalResults(SSqlObj *pSql, SLocalReducer *pLocalReducer, bool no
}
void resetOutputBuf(SQueryInfo *pQueryInfo, SLocalReducer *pLocalReducer) { // reset output buffer to the beginning
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutputCols; ++i) {
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
pLocalReducer->pCtx[i].aOutputBuf =
pLocalReducer->pResultBuf->data + tscFieldInfoGetOffset(pQueryInfo, i) * pLocalReducer->resColModel->capacity;
}
......@@ -1282,7 +1295,7 @@ static void resetEnvForNewResultset(SSqlRes *pRes, SSqlCmd *pCmd, SLocalReducer
// for group result interpolation, do not return if not data is generated
if (pQueryInfo->interpoType != TSDB_INTERPO_NONE) {
int64_t stime = (pQueryInfo->stime < pQueryInfo->etime) ? pQueryInfo->stime : pQueryInfo->etime;
int64_t stime = (pQueryInfo->window.skey < pQueryInfo->window.ekey) ? pQueryInfo->window.skey : pQueryInfo->window.ekey;
int64_t newTime =
taosGetIntervalStartTimestamp(stime, pQueryInfo->intervalTime, pQueryInfo->slidingTimeUnit, precision);
......@@ -1348,7 +1361,7 @@ static bool doHandleLastRemainData(SSqlObj *pSql) {
prevGroupCompleted) {
// if interpoType == TSDB_INTERPO_NONE, return directly
if (pQueryInfo->interpoType != TSDB_INTERPO_NONE) {
int64_t etime = (pQueryInfo->stime < pQueryInfo->etime) ? pQueryInfo->etime : pQueryInfo->stime;
int64_t etime = (pQueryInfo->window.skey < pQueryInfo->window.ekey) ? pQueryInfo->window.ekey : pQueryInfo->window.skey;
etime = taosGetRevisedEndKey(etime, pQueryInfo->order.order, pQueryInfo->intervalTime,
pQueryInfo->slidingTimeUnit, precision);
......@@ -1386,8 +1399,9 @@ static void doProcessResultInNextWindow(SSqlObj *pSql, int32_t numOfRes) {
SLocalReducer *pLocalReducer = pRes->pLocalReducer;
SQueryInfo * pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
size_t size = tscSqlExprNumOfExprs(pQueryInfo);
for (int32_t k = 0; k < pQueryInfo->exprsInfo.numOfExprs; ++k) {
for (int32_t k = 0; k < size; ++k) {
SSqlExpr * pExpr = tscSqlExprGet(pQueryInfo, k);
SQLFunctionCtx *pCtx = &pLocalReducer->pCtx[k];
......
src/client/src/tscServer.c
浏览文件 @ 863bda56
......@@ -359,13 +359,13 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg) {
/*
* Whether to free sqlObj or not should be decided before call the user defined function, since this SqlObj
* may be freed in UDF, and reused by other threads before tscShouldFreeAsyncSqlObj called, in which case
* tscShouldFreeAsyncSqlObj checks an object which is actually allocated by other threads.
* may be freed in UDF, and reused by other threads before tscShouldBeFreed called, in which case
* tscShouldBeFreed checks an object which is actually allocated by other threads.
*
* If this block of memory is re-allocated for an insert thread, in which tscKeepConn[command] equals to 0,
* the tscShouldFreeAsyncSqlObj will success and tscFreeSqlObj free it immediately.
* the tscShouldBeFreed will success and tscFreeSqlObj free it immediately.
*/
bool shouldFree = tscShouldFreeAsyncSqlObj(pSql);
bool shouldFree = tscShouldBeFreed(pSql);
(*pSql->fp)(pSql->param, taosres, rpcMsg->code);
if (shouldFree) {
......@@ -458,7 +458,7 @@ int tscProcessSql(SSqlObj *pSql) {
return doProcessSql(pSql);
}
void tscKillMetricQuery(SSqlObj *pSql) {
void tscKillSTableQuery(SSqlObj *pSql) {
SSqlCmd* pCmd = &pSql->cmd;
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
......@@ -572,9 +572,11 @@ static int32_t tscEstimateQueryMsgSize(SSqlCmd *pCmd, int32_t clauseIndex) {
const static int32_t MIN_QUERY_MSG_PKT_SIZE = TSDB_MAX_BYTES_PER_ROW * 5;
SQueryInfo * pQueryInfo = tscGetQueryInfoDetail(pCmd, clauseIndex);
int32_t srcColListSize = pQueryInfo->colList.numOfCols * sizeof(SColumnInfo);
int32_t exprSize = sizeof(SSqlFuncExprMsg) * pQueryInfo->exprsInfo.numOfExprs;
int32_t srcColListSize = taosArrayGetSize(pQueryInfo->colList) * sizeof(SColumnInfo);
size_t numOfExprs = tscSqlExprNumOfExprs(pQueryInfo);
int32_t exprSize = sizeof(SSqlFuncMsg) * numOfExprs;
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
// meter query without tags values
......@@ -583,20 +585,6 @@ static int32_t tscEstimateQueryMsgSize(SSqlCmd *pCmd, int32_t clauseIndex) {
}
int32_t size = 4096;
#if 0
SSuperTableMeta *pMetricMeta = pTableMetaInfo->pMetricMeta;
SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, pTableMetaInfo->vgroupIndex);
int32_t meterInfoSize = (pMetricMeta->tagLen + sizeof(STableIdInfo)) * pVnodeSidList->numOfSids;
int32_t outputColumnSize = pQueryInfo->exprsInfo.numOfExprs * sizeof(SSqlFuncExprMsg);
int32_t size = meterInfoSize + outputColumnSize + srcColListSize + exprSize + MIN_QUERY_MSG_PKT_SIZE;
if (pQueryInfo->tsBuf != NULL) {
size += pQueryInfo->tsBuf->fileSize;
}
#endif
return size;
}
......@@ -629,7 +617,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
STableMeta * pTableMeta = pTableMetaInfo->pTableMeta;
if (pQueryInfo->colList.numOfCols <= 0) {
if (taosArrayGetSize(pQueryInfo->colList) <= 0) {
tscError("%p illegal value of numOfCols in query msg: %d", pSql, tscGetNumOfColumns(pTableMeta));
return -1;
}
......@@ -674,17 +662,6 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pSql->ipList.ip[i] = pVgroupInfo->ipAddr[i].ip;
}
#if 0
SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, pTableMetaInfo->vgroupIndex);
uint32_t vnodeId = pVnodeSidList->vpeerDesc[pVnodeSidList->index].vnode;
numOfTables = pVnodeSidList->numOfSids;
if (numOfTables <= 0) {
tscError("%p vid:%d,error numOfTables in query message:%d", pSql, vnodeId, numOfTables);
return -1; // error
}
#endif
tscTrace("%p query on super table, numOfVgroup:%d, vgroupIndex:%d", pSql, pTableMetaInfo->vgroupList->numOfVgroups, index);
pQueryMsg->head.vgId = htonl(pVgroupInfo->vgId);
......@@ -692,11 +669,11 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
}
if (pQueryInfo->order.order == TSDB_ORDER_ASC) {
pQueryMsg->window.skey = htobe64(pQueryInfo->stime);
pQueryMsg->window.ekey = htobe64(pQueryInfo->etime);
pQueryMsg->window.skey = htobe64(pQueryInfo->window.skey);
pQueryMsg->window.ekey = htobe64(pQueryInfo->window.ekey);
} else {
pQueryMsg->window.skey = htobe64(pQueryInfo->etime);
pQueryMsg->window.ekey = htobe64(pQueryInfo->stime);
pQueryMsg->window.skey = htobe64(pQueryInfo->window.ekey);
pQueryMsg->window.ekey = htobe64(pQueryInfo->window.skey);
}
pQueryMsg->numOfTables = htonl(numOfTables);
......@@ -705,37 +682,38 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pQueryMsg->interpoType = htons(pQueryInfo->interpoType);
pQueryMsg->limit = htobe64(pQueryInfo->limit.limit);
pQueryMsg->offset = htobe64(pQueryInfo->limit.offset);
pQueryMsg->numOfCols = htons(pQueryInfo->colList.numOfCols);
pQueryMsg->numOfCols = htons(taosArrayGetSize(pQueryInfo->colList));
pQueryMsg->intervalTime = htobe64(pQueryInfo->intervalTime);
pQueryMsg->slidingTime = htobe64(pQueryInfo->slidingTime);
pQueryMsg->slidingTimeUnit = pQueryInfo->slidingTimeUnit;
pQueryMsg->numOfGroupCols = htons(pQueryInfo->groupbyExpr.numOfGroupCols);
pQueryMsg->queryType = htons(pQueryInfo->type);
pQueryMsg->numOfOutputCols = htons(pQueryInfo->exprsInfo.numOfExprs);
int32_t numOfOutput = pQueryInfo->fieldsInfo.numOfOutputCols;
size_t numOfOutput = tscSqlExprNumOfExprs(pQueryInfo);
pQueryMsg->numOfOutput = htons(numOfOutput);
if (numOfOutput < 0) {
tscError("%p illegal value of number of output columns in query msg: %d", pSql, numOfOutput);
return -1;
}
// set column list ids
char *pMsg = (char *)(pQueryMsg->colList) + pQueryInfo->colList.numOfCols * sizeof(SColumnInfo);
size_t numOfCols = taosArrayGetSize(pQueryInfo->colList);
char *pMsg = (char *)(pQueryMsg->colList) + numOfCols * sizeof(SColumnInfo);
SSchema *pSchema = tscGetTableSchema(pTableMeta);
for (int32_t i = 0; i < pQueryInfo->colList.numOfCols; ++i) {
SColumnBase *pCol = tscColumnBaseInfoGet(&pQueryInfo->colList, i);
SSchema * pColSchema = &pSchema[pCol->colIndex.columnIndex];
for (int32_t i = 0; i < numOfCols; ++i) {
SColumn *pCol = taosArrayGetP(pQueryInfo->colList, i);
SSchema *pColSchema = &pSchema[pCol->colIndex.columnIndex];
if (pCol->colIndex.columnIndex >= tscGetNumOfColumns(pTableMeta) || pColSchema->type < TSDB_DATA_TYPE_BOOL ||
pColSchema->type > TSDB_DATA_TYPE_NCHAR) {
tscError("%p sid:%d uid:%" PRIu64" id:%s, column index out of range, numOfColumns:%d, index:%d, column name:%s",
pSql, pTableMeta->sid, pTableMeta->uid, pTableMetaInfo->name, tscGetNumOfColumns(pTableMeta), pCol->colIndex,
pColSchema->name);
// if (pCol->colIndex.columnIndex >= tscGetNumOfColumns(pTableMeta) || pColSchema->type < TSDB_DATA_TYPE_BOOL ||
// pColSchema->type > TSDB_DATA_TYPE_NCHAR) {
// tscError("%p vid:%d sid:%d id:%s, column index out of range, numOfColumns:%d, index:%d, column name:%s", pSql,
// htons(pQueryMsg->vnode), pTableMeta->sid, pTableMetaInfo->name, tscGetNumOfColumns(pTableMeta), pCol->colIndex,
// pColSchema->name);
//
// return -1; // 0 means build msg failed
// }
return -1; // 0 means build msg failed
}
pQueryMsg->colList[i].colId = htons(pColSchema->colId);
pQueryMsg->colList[i].bytes = htons(pColSchema->bytes);
......@@ -747,11 +725,11 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
SColumnFilterInfo *pColFilter = &pCol->filterInfo[f];
SColumnFilterInfo *pFilterMsg = (SColumnFilterInfo *)pMsg;
pFilterMsg->filterOnBinary = htons(pColFilter->filterOnBinary);
pFilterMsg->filterstr = htons(pColFilter->filterstr);
pMsg += sizeof(SColumnFilterInfo);
if (pColFilter->filterOnBinary) {
if (pColFilter->filterstr) {
pFilterMsg->len = htobe64(pColFilter->len);
memcpy(pMsg, (void *)pColFilter->pz, pColFilter->len + 1);
pMsg += (pColFilter->len + 1); // append the additional filter binary info
......@@ -772,7 +750,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
bool hasArithmeticFunction = false;
SSqlFuncExprMsg *pSqlFuncExpr = (SSqlFuncExprMsg *)pMsg;
SSqlFuncMsg *pSqlFuncExpr = (SSqlFuncMsg *)pMsg;
for (int32_t i = 0; i < tscSqlExprNumOfExprs(pQueryInfo); ++i) {
SSqlExpr *pExpr = tscSqlExprGet(pQueryInfo, i);
......@@ -792,7 +770,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pSqlFuncExpr->functionId = htons(pExpr->functionId);
pSqlFuncExpr->numOfParams = htons(pExpr->numOfParams);
pMsg += sizeof(SSqlFuncExprMsg);
pMsg += sizeof(SSqlFuncMsg);
for (int32_t j = 0; j < pExpr->numOfParams; ++j) {
pSqlFuncExpr->arg[j].argType = htons((uint16_t)pExpr->param[j].nType);
......@@ -808,13 +786,14 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
}
}
pSqlFuncExpr = (SSqlFuncExprMsg *)pMsg;
pSqlFuncExpr = (SSqlFuncMsg *)pMsg;
}
int32_t len = 0;
if (hasArithmeticFunction) {
SColumnBase *pColBase = pQueryInfo->colList.pColList;
for (int32_t i = 0; i < pQueryInfo->colList.numOfCols; ++i) {
for (int32_t i = 0; i < numOfCols; ++i) {
SColumn* pColBase = taosArrayGetP(pQueryInfo->colList, i);
char * name = pSchema[pColBase[i].colIndex.columnIndex].name;
int32_t lenx = strlen(name);
memcpy(pMsg, name, lenx);
......@@ -853,7 +832,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
}
if (pQueryInfo->interpoType != TSDB_INTERPO_NONE) {
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutputCols; ++i) {
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
*((int64_t *)pMsg) = htobe64(pQueryInfo->defaultVal[i]);
pMsg += sizeof(pQueryInfo->defaultVal[0]);
}
......@@ -1258,7 +1237,7 @@ int tscBuildCreateTableMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pSchema = (SSchema *)pCreateTableMsg->schema;
for (int i = 0; i < pCmd->numOfCols + pCmd->count; ++i) {
TAOS_FIELD *pField = tscFieldInfoGetField(pQueryInfo, i);
TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
pSchema->type = pField->type;
strcpy(pSchema->name, pField->name);
......@@ -1277,7 +1256,7 @@ int tscBuildCreateTableMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
}
}
tscClearFieldInfo(&pQueryInfo->fieldsInfo);
tscFieldInfoClear(&pQueryInfo->fieldsInfo);
msgLen = pMsg - (char*)pCreateTableMsg;
pCreateTableMsg->contLen = htonl(msgLen);
......@@ -1325,7 +1304,7 @@ int tscBuildAlterTableMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
SSchema *pSchema = pAlterTableMsg->schema;
for (int i = 0; i < tscNumOfFields(pQueryInfo); ++i) {
TAOS_FIELD *pField = tscFieldInfoGetField(pQueryInfo, i);
TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
pSchema->type = pField->type;
strcpy(pSchema->name, pField->name);
......@@ -1384,9 +1363,9 @@ static int tscSetResultPointer(SQueryInfo *pQueryInfo, SSqlRes *pRes) {
return pRes->code;
}
for (int i = 0; i < pQueryInfo->fieldsInfo.numOfOutputCols; ++i) {
for (int i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
int16_t offset = tscFieldInfoGetOffset(pQueryInfo, i);
pRes->tsrow[i] = (pRes->data + offset * pRes->numOfRows);
pRes->tsrow[i] = ((char*) pRes->data + offset * pRes->numOfRows);
}
return 0;
......@@ -2190,28 +2169,35 @@ int tscProcessShowRsp(SSqlObj *pSql) {
strcpy(key + 1, "showlist");
taosCacheRelease(tscCacheHandle, (void *)&(pTableMetaInfo->pTableMeta), false);
size_t size = 0;
STableMeta* pTableMeta = tscCreateTableMetaFromMsg(pMetaMsg, &size);
pTableMetaInfo->pTableMeta =
(STableMeta *)taosCachePut(tscCacheHandle, key, (char *)pTableMeta, size, tsMeterMetaKeepTimer);
pCmd->numOfCols = pQueryInfo->fieldsInfo.numOfOutputCols;
pTableMetaInfo->pTableMeta = taosCachePut(tscCacheHandle, key, (char *)pTableMeta, size, tsMeterMetaKeepTimer);
SSchema *pTableSchema = tscGetTableSchema(pTableMetaInfo->pTableMeta);
tscColumnBaseInfoReserve(&pQueryInfo->colList, pMetaMsg->numOfColumns);
if (pQueryInfo->colList == NULL) {
pQueryInfo->colList = taosArrayInit(4, POINTER_BYTES);
}
SFieldInfo* pFieldInfo = &pQueryInfo->fieldsInfo;
SColumnIndex index = {0};
for (int16_t i = 0; i < pMetaMsg->numOfColumns; ++i) {
pSchema = pMetaMsg->schema;
for (int16_t i = 0; i < pMetaMsg->numOfColumns; ++i, ++pSchema) {
index.columnIndex = i;
tscColumnBaseInfoInsert(pQueryInfo, &index);
tscFieldInfoSetValFromSchema(&pQueryInfo->fieldsInfo, i, &pTableSchema[i]);
tscColumnListInsert(pQueryInfo->colList, &index);
TAOS_FIELD f = tscCreateField(pSchema->type, pSchema->name, pSchema->bytes);
SFieldSupInfo* pInfo = tscFieldInfoAppend(pFieldInfo, &f);
pQueryInfo->fieldsInfo.pSqlExpr[i] = tscSqlExprInsert(pQueryInfo, i, TSDB_FUNC_TS_DUMMY, &index,
pInfo->pSqlExpr = tscSqlExprAppend(pQueryInfo, TSDB_FUNC_TS_DUMMY, &index,
pTableSchema[i].type, pTableSchema[i].bytes, pTableSchema[i].bytes);
}
tscFieldInfoCalOffset(pQueryInfo);
pCmd->numOfCols = pQueryInfo->fieldsInfo.numOfOutput;
tscFieldInfoUpdateOffset(pQueryInfo);
tfree(pTableMeta);
return 0;
......@@ -2322,7 +2308,7 @@ int tscProcessQueryRsp(SSqlObj *pSql) {
return 0;
}
int tscProcessRetrieveRspFromVnode(SSqlObj *pSql) {
int tscProcessRetrieveRspFromNode(SSqlObj *pSql) {
SSqlRes *pRes = &pSql->res;
SSqlCmd *pCmd = &pSql->cmd;
......@@ -2339,9 +2325,9 @@ int tscProcessRetrieveRspFromVnode(SSqlObj *pSql) {
tscSetResultPointer(pQueryInfo, pRes);
if (pSql->pSubscription != NULL) {
int32_t numOfCols = pQueryInfo->fieldsInfo.numOfOutputCols;
int32_t numOfCols = pQueryInfo->fieldsInfo.numOfOutput;
TAOS_FIELD *pField = tscFieldInfoGetField(pQueryInfo, numOfCols - 1);
TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, numOfCols - 1);
int16_t offset = tscFieldInfoGetOffset(pQueryInfo, numOfCols - 1);
char* p = pRes->data + (pField->bytes + offset) * pRes->numOfRows;
......@@ -2479,7 +2465,7 @@ int tscRenewMeterMeta(SSqlObj *pSql, char *tableId) {
* 1. only update the metermeta in force model metricmeta is not updated
* 2. if get metermeta failed, still get the metermeta
*/
if (pTableMetaInfo->pTableMeta == NULL || !tscQueryOnMetric(pCmd)) {
if (pTableMetaInfo->pTableMeta == NULL || !tscQueryOnSTable(pCmd)) {
STableMeta* pTableMeta = pTableMetaInfo->pTableMeta;
if (pTableMetaInfo->pTableMeta) {
tscTrace("%p update table meta, old: numOfTags:%d, numOfCols:%d, uid:%" PRId64 ", addr:%p", pSql,
......@@ -2624,7 +2610,7 @@ void tscInitMsgsFp() {
tscBuildMsg[TSDB_SQL_KILL_CONNECTION] = tscBuildKillMsg;
tscProcessMsgRsp[TSDB_SQL_SELECT] = tscProcessQueryRsp;
tscProcessMsgRsp[TSDB_SQL_FETCH] = tscProcessRetrieveRspFromVnode;
tscProcessMsgRsp[TSDB_SQL_FETCH] = tscProcessRetrieveRspFromNode;
tscProcessMsgRsp[TSDB_SQL_DROP_DB] = tscProcessDropDbRsp;
tscProcessMsgRsp[TSDB_SQL_DROP_TABLE] = tscProcessDropTableRsp;
......@@ -2635,7 +2621,7 @@ void tscInitMsgsFp() {
tscProcessMsgRsp[TSDB_SQL_MULTI_META] = tscProcessMultiMeterMetaRsp;
tscProcessMsgRsp[TSDB_SQL_SHOW] = tscProcessShowRsp;
tscProcessMsgRsp[TSDB_SQL_RETRIEVE] = tscProcessRetrieveRspFromVnode; // rsp handled by same function.
tscProcessMsgRsp[TSDB_SQL_RETRIEVE] = tscProcessRetrieveRspFromNode; // rsp handled by same function.
tscProcessMsgRsp[TSDB_SQL_DESCRIBE_TABLE] = tscProcessDescribeTableRsp;
tscProcessMsgRsp[TSDB_SQL_RETRIEVE_TAGS] = tscProcessTagRetrieveRsp;
......
src/client/src/tscSql.c
浏览文件 @ 863bda56
......@@ -326,16 +326,21 @@ int taos_num_fields(TAOS_RES *res) {
SSqlObj *pSql = (SSqlObj *)res;
if (pSql == NULL || pSql->signature != pSql) return 0;
int32_t num = 0;
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
if (pQueryInfo == NULL) {
return 0;
return num;
}
SFieldInfo *pFieldsInfo = &pQueryInfo->fieldsInfo;
if (pFieldsInfo)
return (pFieldsInfo->numOfOutputCols - pFieldsInfo->numOfHiddenCols);
else
return 0;
size_t numOfCols = tscNumOfFields(pQueryInfo);
for(int32_t i = 0; i < numOfCols; ++i) {
SFieldSupInfo* pInfo = taosArrayGet(pQueryInfo->fieldsInfo.pSupportInfo, i);
if (pInfo->visible) {
num++;
}
}
return num;
}
int taos_field_count(TAOS *taos) {
......@@ -357,11 +362,16 @@ TAOS_FIELD *taos_fetch_fields(TAOS_RES *res) {
if (pSql == NULL || pSql->signature != pSql) return 0;
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
if (pQueryInfo)
return pQueryInfo->fieldsInfo.pFields;
else
if (pQueryInfo == NULL) {
return NULL;
}
size_t numOfCols = tscNumOfFields(pQueryInfo);
if (numOfCols == 0) {
return NULL;
}
return pQueryInfo->fieldsInfo.pFields->pData;
}
int taos_retrieve(TAOS_RES *res) {
......@@ -414,8 +424,8 @@ int taos_fetch_block_impl(TAOS_RES *res, TAOS_ROW *rows) {
if (pQueryInfo == NULL)
return 0;
for (int i = 0; i < pQueryInfo->fieldsInfo.numOfOutputCols; ++i) {
pRes->tsrow[i] = TSC_GET_RESPTR_BASE(pRes, pQueryInfo, i);
for (int i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
pRes->tsrow[i] = tscGetResultColumnChr(pRes, pQueryInfo, i);
}
*rows = pRes->tsrow;
......@@ -448,7 +458,7 @@ static void transferNcharData(SSqlObj *pSql, int32_t columnIndex, TAOS_FIELD *pF
static char *getArithemicInputSrc(void *param, char *name, int32_t colId) {
SArithmeticSupport *pSupport = (SArithmeticSupport *)param;
SSqlFunctionExpr * pExpr = pSupport->pExpr;
SArithExprInfo * pExpr = pSupport->pArithExpr;
int32_t index = -1;
for (int32_t i = 0; i < pExpr->binExprInfo.numOfCols; ++i) {
......@@ -476,7 +486,8 @@ static void **doSetResultRowData(SSqlObj *pSql) {
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
//todo refactor move away
for(int32_t k = 0; k < pQueryInfo->exprsInfo.numOfExprs; ++k) {
size_t numOfExprs = tscSqlExprNumOfExprs(pQueryInfo);
for(int32_t k = 0; k < numOfExprs; ++k) {
SSqlExpr* pExpr = tscSqlExprGet(pQueryInfo, k);
if (k > 0) {
......@@ -487,9 +498,9 @@ static void **doSetResultRowData(SSqlObj *pSql) {
int32_t num = 0;
for (int i = 0; i < tscNumOfFields(pQueryInfo); ++i) {
if (pQueryInfo->fieldsInfo.pSqlExpr[i] != NULL) {
SSqlExpr* pExpr = pQueryInfo->fieldsInfo.pSqlExpr[i];
pRes->tsrow[i] = TSC_GET_RESPTR_BASE(pRes, pQueryInfo, i) + pExpr->resBytes * pRes->row;
SFieldSupInfo* pInfo = tscFieldInfoGetSupp(&pQueryInfo->fieldsInfo, i);
if (pInfo->pSqlExpr != NULL) {
pRes->tsrow[i] = tscGetResultColumnChr(pRes, pQueryInfo, i) + pInfo->pSqlExpr->resBytes * pRes->row;
} else {
assert(0);
}
......@@ -499,37 +510,37 @@ static void **doSetResultRowData(SSqlObj *pSql) {
continue;
}
TAOS_FIELD *pField = tscFieldInfoGetField(pQueryInfo, i);
TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
transferNcharData(pSql, i, pField);
// calculate the result from serveral other columns
if (pQueryInfo->fieldsInfo.pExpr != NULL && pQueryInfo->fieldsInfo.pExpr[i] != NULL) {
// calculate the result from several other columns
if (pInfo->pArithExprInfo != NULL) {
SArithmeticSupport *sas = (SArithmeticSupport *)calloc(1, sizeof(SArithmeticSupport));
sas->offset = 0;
sas->pExpr = pQueryInfo->fieldsInfo.pExpr[i];
sas->pArithExpr = pInfo->pArithExprInfo;
sas->numOfCols = sas->pExpr->binExprInfo.numOfCols;
sas->numOfCols = sas->pArithExpr->binExprInfo.numOfCols;
if (pRes->buffer[i] == NULL) {
pRes->buffer[i] = malloc(tscFieldInfoGetField(pQueryInfo, i)->bytes);
pRes->buffer[i] = malloc(tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i)->bytes);
}
for(int32_t k = 0; k < sas->numOfCols; ++k) {
int32_t columnIndex = sas->pExpr->binExprInfo.pReqColumns[k].colIndex;
int32_t columnIndex = sas->pArithExpr->binExprInfo.pReqColumns[k].colIndex;
SSqlExpr* pExpr = tscSqlExprGet(pQueryInfo, columnIndex);
sas->elemSize[k] = pExpr->resBytes;
sas->data[k] = (pRes->data + pRes->numOfRows* pExpr->offset) + pRes->row*pExpr->resBytes;
}
tSQLBinaryExprCalcTraverse(sas->pExpr->binExprInfo.pBinExpr, 1, pRes->buffer[i], sas, TSDB_ORDER_ASC, getArithemicInputSrc);
tSQLBinaryExprCalcTraverse(sas->pArithExpr->binExprInfo.pBinExpr, 1, pRes->buffer[i], sas, TSDB_ORDER_ASC, getArithemicInputSrc);
pRes->tsrow[i] = pRes->buffer[i];
free(sas); //todo optimization
}
}
assert(num <= pQueryInfo->fieldsInfo.numOfOutputCols);
assert(num <= pQueryInfo->fieldsInfo.numOfOutput);
pRes->row++; // index increase one-step
return pRes->tsrow;
......@@ -591,11 +602,13 @@ static bool tscHashRemainDataInSubqueryResultSet(SSqlObj *pSql) {
static UNUSED_FUNC void **tscBuildResFromSubqueries(SSqlObj *pSql) {
SSqlRes *pRes = &pSql->res;
while (1) {
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, pSql->cmd.clauseIndex);
size_t numOfExprs = tscSqlExprNumOfExprs(pQueryInfo);
if (pRes->tsrow == NULL) {
pRes->tsrow = calloc(pQueryInfo->exprsInfo.numOfExprs, POINTER_BYTES);
pRes->tsrow = calloc(numOfExprs, POINTER_BYTES);
}
bool success = false;
......@@ -619,7 +632,9 @@ static UNUSED_FUNC void **tscBuildResFromSubqueries(SSqlObj *pSql) {
}
if (success) { // current row of final output has been built, return to app
for (int32_t i = 0; i < pQueryInfo->exprsInfo.numOfExprs; ++i) {
size_t numOfExprs = tscSqlExprNumOfExprs(pQueryInfo);
for (int32_t i = 0; i < numOfExprs; ++i) {
int32_t tableIndex = pRes->pColumnIndex[i].tableIndex;
int32_t columnIndex = pRes->pColumnIndex[i].columnIndex;
......@@ -949,7 +964,7 @@ void taos_stop_query(TAOS_RES *res) {
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
if (tscIsTwoStageSTableQuery(pQueryInfo, 0)) {
tscKillMetricQuery(pSql);
tscKillSTableQuery(pSql);
return;
}
......@@ -1126,7 +1141,7 @@ static int tscParseTblNameList(SSqlObj *pSql, const char *tblNameList, int32_t t
return code;
}
if ((code = setMeterID(pTableMetaInfo, &sToken, pSql)) != TSDB_CODE_SUCCESS) {
if ((code = tscSetTableId(pTableMetaInfo, &sToken, pSql)) != TSDB_CODE_SUCCESS) {
return code;
}
......
src/client/src/tscStream.c
浏览文件 @ 863bda56
......@@ -36,7 +36,7 @@ static int64_t getDelayValueAfterTimewindowClosed(SSqlStream* pStream, int64_t l
}
static bool isProjectStream(SQueryInfo* pQueryInfo) {
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutputCols; ++i) {
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
SSqlExpr *pExpr = tscSqlExprGet(pQueryInfo, i);
if (pExpr->functionId != TSDB_FUNC_PRJ) {
return false;
......@@ -86,7 +86,7 @@ static void tscProcessStreamLaunchQuery(SSchedMsg *pMsg) {
if (code == TSDB_CODE_ACTION_IN_PROGRESS) return;
}
tscTansformSQLFunctionForSTableQuery(pQueryInfo);
tscTansformSQLFuncForSTableQuery(pQueryInfo);
// failed to get meter/metric meta, retry in 10sec.
if (code != TSDB_CODE_SUCCESS) {
......@@ -116,18 +116,18 @@ static void tscProcessStreamTimer(void *handle, void *tmrId) {
if (isProjectStream(pQueryInfo)) {
/*
* pQueryInfo->etime, which is the start time, does not change in case of
* pQueryInfo->window.ekey, which is the start time, does not change in case of
* repeat first execution, once the first execution failed.
*/
pQueryInfo->stime = pStream->stime; // start time
pQueryInfo->window.skey = pStream->stime; // start time
pQueryInfo->etime = taosGetTimestamp(pStream->precision); // end time
if (pQueryInfo->etime > pStream->etime) {
pQueryInfo->etime = pStream->etime;
pQueryInfo->window.ekey = taosGetTimestamp(pStream->precision); // end time
if (pQueryInfo->window.ekey > pStream->etime) {
pQueryInfo->window.ekey = pStream->etime;
}
} else {
pQueryInfo->stime = pStream->stime - pStream->interval;
pQueryInfo->etime = pStream->stime - 1;
pQueryInfo->window.skey = pStream->stime - pStream->interval;
pQueryInfo->window.ekey = pStream->stime - 1;
}
// launch stream computing in a new thread
......@@ -219,9 +219,9 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
void *oldPtr = pSql->res.data;
pSql->res.data = tmpRes;
for (int32_t i = 1; i < pQueryInfo->fieldsInfo.numOfOutputCols; ++i) {
for (int32_t i = 1; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
int16_t offset = tscFieldInfoGetOffset(pQueryInfo, i);
TAOS_FIELD *pField = tscFieldInfoGetField(pQueryInfo, i);
TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
assignVal(pSql->res.data + offset, (char *)(&pQueryInfo->defaultVal[i]), pField->bytes, pField->type);
row[i] = pSql->res.data + offset;
......@@ -231,7 +231,7 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
row[0] = pRes->data;
// char result[512] = {0};
// taos_print_row(result, row, pQueryInfo->fieldsInfo.pFields, pQueryInfo->fieldsInfo.numOfOutputCols);
// taos_print_row(result, row, pQueryInfo->fieldsInfo.pFields, pQueryInfo->fieldsInfo.numOfOutput);
// tscPrint("%p stream:%p query result: %s", pSql, pStream, result);
tscTrace("%p stream:%p fetch result", pSql, pStream);
......@@ -425,10 +425,10 @@ static int64_t tscGetStreamStartTimestamp(SSqlObj *pSql, SSqlStream *pStream, in
pStream->slidingTime = tsProjectExecInterval;
if (stime != 0) { // first projection start from the latest event timestamp
assert(stime >= pQueryInfo->stime);
assert(stime >= pQueryInfo->window.skey);
stime += 1; // exclude the last records from table
} else {
stime = pQueryInfo->stime;
stime = pQueryInfo->window.skey;
}
} else { // timewindow based aggregation stream
if (stime == 0) { // no data in meter till now
......@@ -548,7 +548,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
pStream->precision = tinfo.precision;
pStream->ctime = taosGetTimestamp(pStream->precision);
pStream->etime = pQueryInfo->etime;
pStream->etime = pQueryInfo->window.ekey;
pSql->pStream = pStream;
tscAddIntoStreamList(pStream);
......
src/client/src/tscSubquery.c
浏览文件 @ 863bda56
......@@ -190,10 +190,10 @@ void tscDestroyJoinSupporter(SJoinSubquerySupporter* pSupporter) {
return;
}
tscSqlExprInfoDestroy(&pSupporter->exprsInfo);
tscColumnBaseInfoDestroy(&pSupporter->colList);
tscSqlExprInfoDestroy(pSupporter->exprsInfo);
tscColumnListDestroy(pSupporter->colList);
tscClearFieldInfo(&pSupporter->fieldsInfo);
tscFieldInfoClear(&pSupporter->fieldsInfo);
if (pSupporter->f != NULL) {
fclose(pSupporter->f);
......@@ -211,8 +211,10 @@ void tscDestroyJoinSupporter(SJoinSubquerySupporter* pSupporter) {
*
*/
bool needSecondaryQuery(SQueryInfo* pQueryInfo) {
for (int32_t i = 0; i < pQueryInfo->colList.numOfCols; ++i) {
SColumnBase* pBase = tscColumnBaseInfoGet(&pQueryInfo->colList, i);
size_t numOfCols = taosArrayGetSize(pQueryInfo->colList);
for (int32_t i = 0; i < numOfCols; ++i) {
SColumn* pBase = taosArrayGet(pQueryInfo->colList, i);
if (pBase->colIndex.columnIndex != PRIMARYKEY_TIMESTAMP_COL_INDEX) {
return true;
}
......@@ -236,7 +238,7 @@ int32_t tscLaunchSecondPhaseSubqueries(SSqlObj* pSql) {
for (int32_t i = 0; i < pSql->numOfSubs; ++i) {
pSupporter = pSql->pSubs[i]->param;
if (pSupporter->exprsInfo.numOfExprs > 0) {
if (taosArrayGetSize(pSupporter->exprsInfo) > 0) {
++numOfSub;
}
}
......@@ -262,7 +264,7 @@ int32_t tscLaunchSecondPhaseSubqueries(SSqlObj* pSql) {
pSupporter = pPrevSub->param;
if (pSupporter->exprsInfo.numOfExprs == 0) {
if (taosArrayGetSize(pSupporter->exprsInfo) == 0) {
tscTrace("%p subIndex: %d, not need to launch query, ignore it", pSql, i);
tscDestroyJoinSupporter(pSupporter);
......@@ -277,7 +279,7 @@ int32_t tscLaunchSecondPhaseSubqueries(SSqlObj* pSql) {
pSubQueryInfo->tsBuf = NULL;
// free result for async object will also free sqlObj
assert(pSubQueryInfo->exprsInfo.numOfExprs == 1); // ts_comp query only requires one resutl columns
assert(tscSqlExprNumOfExprs(pSubQueryInfo) == 1); // ts_comp query only requires one resutl columns
taos_free_result(pPrevSub);
SSqlObj *pNew = createSubqueryObj(pSql, (int16_t) i, tscJoinQueryCallback, pSupporter, TSDB_SQL_SELECT, NULL);
......@@ -299,27 +301,27 @@ int32_t tscLaunchSecondPhaseSubqueries(SSqlObj* pSql) {
pQueryInfo->intervalTime = pSupporter->interval;
pQueryInfo->groupbyExpr = pSupporter->groupbyExpr;
tscColumnBaseInfoCopy(&pQueryInfo->colList, &pSupporter->colList, 0);
tscColumnListCopy(pQueryInfo->colList, pSupporter->colList, 0);
tscTagCondCopy(&pQueryInfo->tagCond, &pSupporter->tagCond);
tscSqlExprCopy(&pQueryInfo->exprsInfo, &pSupporter->exprsInfo, pSupporter->uid, false);
tscFieldInfoCopyAll(&pQueryInfo->fieldsInfo, &pSupporter->fieldsInfo);
pQueryInfo->exprsInfo = tscSqlExprCopy(pSupporter->exprsInfo, pSupporter->uid, false);
tscFieldInfoCopy(&pQueryInfo->fieldsInfo, &pSupporter->fieldsInfo);
pSupporter->exprsInfo.numOfExprs = 0;
pSupporter->fieldsInfo.numOfOutputCols = 0;
pSupporter->fieldsInfo.numOfOutput = 0;
/*
* if the first column of the secondary query is not ts function, add this function.
* Because this column is required to filter with timestamp after intersecting.
*/
if (pSupporter->exprsInfo.pExprs[0]->functionId != TSDB_FUNC_TS) {
SSqlExpr* pExpr = taosArrayGet(pSupporter->exprsInfo, 0);
if (pExpr->functionId != TSDB_FUNC_TS) {
tscAddTimestampColumn(pQueryInfo, TSDB_FUNC_TS, 0);
}
SQueryInfo *pNewQueryInfo = tscGetQueryInfoDetail(&pNew->cmd, 0);
assert(pNew->numOfSubs == 0 && pNew->cmd.numOfClause == 1 && pNewQueryInfo->numOfTables == 1);
tscFieldInfoCalOffset(pNewQueryInfo);
tscFieldInfoUpdateOffset(pNewQueryInfo);
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pNewQueryInfo, 0);
......@@ -342,10 +344,11 @@ int32_t tscLaunchSecondPhaseSubqueries(SSqlObj* pSql) {
tscPrintSelectClause(pNew, 0);
size_t numOfCols = taosArrayGetSize(pNewQueryInfo->colList);
tscTrace("%p subquery:%p tableIndex:%d, vgroupIndex:%d, type:%d, exprInfo:%d, colList:%d, fieldsInfo:%d, name:%s",
pSql, pNew, 0, pTableMetaInfo->vgroupIndex, pNewQueryInfo->type,
pNewQueryInfo->exprsInfo.numOfExprs, pNewQueryInfo->colList.numOfCols,
pNewQueryInfo->fieldsInfo.numOfOutputCols, pNewQueryInfo->pTableMetaInfo[0]->name);
taosArrayGetSize(pNewQueryInfo->exprsInfo), numOfCols,
pNewQueryInfo->fieldsInfo.numOfOutput, pNewQueryInfo->pTableMetaInfo[0]->name);
}
//prepare the subqueries object failed, abort
......@@ -413,10 +416,10 @@ static void quitAllSubquery(SSqlObj* pSqlObj, SJoinSubquerySupporter* pSupporter
// update the query time range according to the join results on timestamp
static void updateQueryTimeRange(SQueryInfo* pQueryInfo, int64_t st, int64_t et) {
assert(pQueryInfo->stime <= st && pQueryInfo->etime >= et);
assert(pQueryInfo->window.skey <= st && pQueryInfo->window.ekey >= et);
pQueryInfo->stime = st;
pQueryInfo->etime = et;
pQueryInfo->window.skey = st;
pQueryInfo->window.ekey = et;
}
static void joinRetrieveCallback(void* param, TAOS_RES* tres, int numOfRows) {
......@@ -688,9 +691,9 @@ void tscSetupOutputColumnIndex(SSqlObj* pSql) {
}
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
pRes->pColumnIndex = calloc(1, sizeof(SColumnIndex) * pQueryInfo->fieldsInfo.numOfOutputCols);
pRes->pColumnIndex = calloc(1, sizeof(SColumnIndex) * pQueryInfo->fieldsInfo.numOfOutput);
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutputCols; ++i) {
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
SSqlExpr* pExpr = tscSqlExprGet(pQueryInfo, i);
int32_t tableIndexOfSub = -1;
......@@ -707,7 +710,8 @@ void tscSetupOutputColumnIndex(SSqlObj* pSql) {
SSqlCmd* pSubCmd = &pSql->pSubs[tableIndexOfSub]->cmd;
SQueryInfo* pSubQueryInfo = tscGetQueryInfoDetail(pSubCmd, 0);
for (int32_t k = 0; k < pSubQueryInfo->exprsInfo.numOfExprs; ++k) {
size_t numOfExprs = taosArrayGetSize(pSubQueryInfo->exprsInfo);
for (int32_t k = 0; k < numOfExprs; ++k) {
SSqlExpr* pSubExpr = tscSqlExprGet(pSubQueryInfo, k);
if (pExpr->functionId == pSubExpr->functionId && pExpr->colInfo.colId == pSubExpr->colInfo.colId) {
pRes->pColumnIndex[i] = (SColumnIndex){.tableIndex = tableIndexOfSub, .columnIndex = k};
......@@ -723,11 +727,6 @@ void tscJoinQueryCallback(void* param, TAOS_RES* tres, int code) {
// int32_t idx = pSql->cmd.vnodeIdx;
// SVnodeSidList *vnodeInfo = NULL;
// if (pTableMetaInfo->pMetricMeta != NULL) {
// vnodeInfo = tscGetVnodeSidList(pTableMetaInfo->pMetricMeta, idx - 1);
// }
SJoinSubquerySupporter* pSupporter = (SJoinSubquerySupporter*)param;
// if (atomic_add_fetch_32(pSupporter->numOfComplete, 1) >=
......@@ -850,11 +849,17 @@ int32_t tscLaunchJoinSubquery(SSqlObj *pSql, int16_t tableIndex, SJoinSubquerySu
SQueryInfo *pNewQueryInfo = tscGetQueryInfoDetail(&pNew->cmd, 0);
assert(pNewQueryInfo != NULL);
tscColumnBaseInfoUpdateTableIndex(&pNewQueryInfo->colList, 0);
tscColumnBaseInfoCopy(&pSupporter->colList, &pNewQueryInfo->colList, 0);
// update the table index
size_t num = taosArrayGetSize(pNewQueryInfo->colList);
for (int32_t i = 0; i < num; ++i) {
SColumn* pCol = taosArrayGetP(pNewQueryInfo->colList, i);
pCol->colIndex.tableIndex = 0;
}
tscSqlExprCopy(&pSupporter->exprsInfo, &pNewQueryInfo->exprsInfo, pSupporter->uid, false);
tscFieldInfoCopyAll(&pSupporter->fieldsInfo, &pNewQueryInfo->fieldsInfo);
tscColumnListCopy(pSupporter->colList, pNewQueryInfo->colList, 0);
pSupporter->exprsInfo = tscSqlExprCopy(pNewQueryInfo->exprsInfo, pSupporter->uid, false);
tscFieldInfoCopy(&pSupporter->fieldsInfo, &pNewQueryInfo->fieldsInfo);
tscTagCondCopy(&pSupporter->tagCond, &pNewQueryInfo->tagCond);
......@@ -867,8 +872,7 @@ int32_t tscLaunchJoinSubquery(SSqlObj *pSql, int16_t tableIndex, SJoinSubquerySu
memset(&pNewQueryInfo->groupbyExpr, 0, sizeof(SSqlGroupbyExpr));
// this data needs to be transfer to support struct
pNewQueryInfo->fieldsInfo.numOfOutputCols = 0;
pNewQueryInfo->exprsInfo.numOfExprs = 0;
pNewQueryInfo->fieldsInfo.numOfOutput = 0;
// set the ts,tags that involved in join, as the output column of intermediate result
tscClearSubqueryInfo(&pNew->cmd);
......@@ -888,27 +892,26 @@ int32_t tscLaunchJoinSubquery(SSqlObj *pSql, int16_t tableIndex, SJoinSubquerySu
pExpr->numOfParams = 1;
// add the filter tag column
for (int32_t i = 0; i < pSupporter->colList.numOfCols; ++i) {
SColumnBase *pColBase = &pSupporter->colList.pColList[i];
if (pColBase->numOfFilters > 0) { // copy to the pNew->cmd.colList if it is filtered.
tscColumnBaseCopy(&pNewQueryInfo->colList.pColList[pNewQueryInfo->colList.numOfCols], pColBase);
pNewQueryInfo->colList.numOfCols++;
size_t s = taosArrayGetSize(pSupporter->colList);
for (int32_t i = 0; i < s; ++i) {
SColumn *pCol = taosArrayGetP(pSupporter->colList, i);
if (pCol->numOfFilters > 0) { // copy to the pNew->cmd.colList if it is filtered.
SColumn* p = tscColumnClone(pCol);
taosArrayPush(pNewQueryInfo->colList, &p);
}
}
tscTrace("%p subquery:%p tableIndex:%d, vnodeIdx:%d, type:%d, transfer to ts_comp query to retrieve timestamps, "
"exprInfo:%d, colList:%d, fieldsInfo:%d, name:%s",
pSql, pNew, tableIndex, pTableMetaInfo->vgroupIndex, pNewQueryInfo->type,
pNewQueryInfo->exprsInfo.numOfExprs, pNewQueryInfo->colList.numOfCols,
pNewQueryInfo->fieldsInfo.numOfOutputCols, pNewQueryInfo->pTableMetaInfo[0]->name);
tscPrintSelectClause(pNew, 0);
size_t numOfCols = taosArrayGetSize(pNewQueryInfo->colList);
tscTrace("%p subquery:%p tableIndex:%d, vnodeIdx:%d, type:%d, transfer to ts_comp query to retrieve timestamps, "
"exprInfo:%d, colList:%d, fieldsInfo:%d, name:%s",
pSql, pNew, tableIndex, pTableMetaInfo->vgroupIndex, pNewQueryInfo->type,
pNewQueryInfo->exprsInfo.numOfExprs, pNewQueryInfo->colList.numOfCols,
pNewQueryInfo->fieldsInfo.numOfOutputCols, pNewQueryInfo->pTableMetaInfo[0]->name);
tscSqlExprNumOfExprs(pNewQueryInfo), numOfCols,
pNewQueryInfo->fieldsInfo.numOfOutput, pNewQueryInfo->pTableMetaInfo[0]->name);
tscPrintSelectClause(pNew, 0);
} else {
SQueryInfo *pNewQueryInfo = tscGetQueryInfoDetail(&pNew->cmd, 0);
pNewQueryInfo->type |= TSDB_QUERY_TYPE_SUBQUERY;
......@@ -1352,7 +1355,7 @@ static void tscRetrieveFromDnodeCallBack(void *param, TAOS_RES *tres, int numOfR
}
#ifdef _DEBUG_VIEW
printf("received data from vnode: %d rows\n", pRes->numOfRows);
printf("received data from vnode: %"PRIu64" rows\n", pRes->numOfRows);
SSrcColumnInfo colInfo[256] = {0};
tscGetSrcColumnInfo(colInfo, pQueryInfo);
......
src/client/src/tscUtil.c
浏览文件 @ 863bda56
此差异已折叠。
src/dnode/inc/dnodeLog.h
浏览文件 @ 863bda56
......@@ -39,10 +39,6 @@ extern int32_t ddebugFlag;
#define dPrint(...) \
{ taosPrintLog("DND ", 255, __VA_ARGS__); }
#define dLError(...) taosLogError(__VA_ARGS__) dError(__VA_ARGS__)
#define dLWarn(...) taosLogWarn(__VA_ARGS__) dWarn(__VA_ARGS__)
#define dLPrint(...) taosLogPrint(__VA_ARGS__) dPrint(__VA_ARGS__)
#ifdef __cplusplus
}
#endif
......
src/dnode/src/dnodeModule.c
浏览文件 @ 863bda56
......@@ -90,7 +90,7 @@ void dnodeCleanUpModules() {
int32_t dnodeInitModules() {
dnodeAllocModules();
for (int32_t module = 0; module < TSDB_MOD_MAX; ++module) {
for (EModuleType module = 0; module < TSDB_MOD_MAX; ++module) {
if (tsModule[module].initFp) {
if ((*tsModule[module].initFp)() != 0) {
dError("failed to init module:%s", tsModule[module].name);
......@@ -103,7 +103,7 @@ int32_t dnodeInitModules() {
}
void dnodeStartModules() {
for (int32_t module = 1; module < TSDB_MOD_MAX; ++module) {
for (EModuleType module = 1; module < TSDB_MOD_MAX; ++module) {
if (tsModule[module].enable && tsModule[module].startFp) {
if ((*tsModule[module].startFp)() != 0) {
dError("failed to start module:%s", tsModule[module].name);
......
src/dnode/src/dnodeShell.c
浏览文件 @ 863bda56
......@@ -20,16 +20,17 @@
#include "taosmsg.h"
#include "trpc.h"
#include "tglobal.h"
#include "http.h"
#include "dnode.h"
#include "dnodeLog.h"
#include "dnodeRead.h"
#include "dnodeWrite.h"
#include "dnodeShell.h"
static void (*dnodeProcessShellMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *);
static void dnodeProcessMsgFromShell(SRpcMsg *pMsg);
static int dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char *secret, char *ckey);
static void *tsDnodeShellRpc = NULL;
static void (*dnodeProcessShellMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *);
static void dnodeProcessMsgFromShell(SRpcMsg *pMsg);
static int dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char *secret, char *ckey);
static void * tsDnodeShellRpc = NULL;
static int32_t tsDnodeQueryReqNum = 0;
static int32_t tsDnodeSubmitReqNum = 0;
......@@ -110,7 +111,7 @@ static int dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char
SDnodeStatisInfo dnodeGetStatisInfo() {
SDnodeStatisInfo info = {0};
if (dnodeGetRunStatus() == TSDB_DNODE_RUN_STATUS_RUNING) {
//info.httpReqNum = httpGetReqCount();
info.httpReqNum = httpGetReqCount();
info.queryReqNum = atomic_exchange_32(&tsDnodeQueryReqNum, 0);
info.submitReqNum = atomic_exchange_32(&tsDnodeSubmitReqNum, 0);
}
......
src/inc/monitor.h
浏览文件 @ 863bda56
......@@ -22,10 +22,38 @@ extern "C" {
#include <stdint.h>
typedef struct {
char * acctId;
int64_t currentPointsPerSecond;
int64_t maxPointsPerSecond;
int64_t totalTimeSeries;
int64_t maxTimeSeries;
int64_t totalStorage;
int64_t maxStorage;
int64_t totalQueryTime;
int64_t maxQueryTime;
int64_t totalInbound;
int64_t maxInbound;
int64_t totalOutbound;
int64_t maxOutbound;
int64_t totalDbs;
int64_t maxDbs;
int64_t totalUsers;
int64_t maxUsers;
int64_t totalStreams;
int64_t maxStreams;
int64_t totalConns;
int64_t maxConns;
int8_t accessState;
} SAcctMonitorObj;
int32_t monitorInitSystem();
int32_t monitorStartSystem();
void monitorStopSystem();
void monitorCleanUpSystem();
void monitorStopSystem();
void monitorCleanUpSystem();
void monitorSaveAcctLog(SAcctMonitorObj *pMonObj);
void monitorSaveLog(int32_t level, const char *const format, ...);
void monitorExecuteSQL(char *sql);
#ifdef __cplusplus
}
......
src/inc/taosmsg.h
浏览文件 @ 863bda56
......@@ -372,7 +372,7 @@ typedef struct SColIndex {
/* sql function msg, to describe the message to vnode about sql function
* operations in select clause */
typedef struct SSqlFuncExprMsg {
typedef struct SSqlFuncMsg {
int16_t functionId;
int16_t numOfParams;
......@@ -386,26 +386,26 @@ typedef struct SSqlFuncExprMsg {
char * pz;
} argValue;
} arg[3];
} SSqlFuncExprMsg;
} SSqlFuncMsg;
typedef struct SSqlBinaryExprInfo {
typedef struct SExprInfo {
struct tExprNode *pBinExpr; /* for binary expression */
int32_t numOfCols; /* binary expression involves the readed number of columns*/
SColIndex * pReqColumns; /* source column list */
} SSqlBinaryExprInfo;
} SExprInfo;
typedef struct SSqlFunctionExpr {
SSqlFuncExprMsg pBase;
SSqlBinaryExprInfo binExprInfo;
int16_t resBytes;
int16_t resType;
int16_t interResBytes;
} SSqlFunctionExpr;
typedef struct SArithExprInfo {
SSqlFuncMsg pBase;
SExprInfo binExprInfo;
int16_t bytes;
int16_t type;
int16_t interResBytes;
} SArithExprInfo;
typedef struct SColumnFilterInfo {
int16_t lowerRelOptr;
int16_t upperRelOptr;
int16_t filterOnBinary; /* denote if current column is binary */
int16_t filterstr; // denote if current column is char(binary/nchar)
union {
struct {
......@@ -469,7 +469,7 @@ typedef struct {
int64_t limit;
int64_t offset;
uint16_t queryType; // denote another query process
int16_t numOfOutputCols; // final output columns numbers
int16_t numOfOutput; // final output columns numbers
int16_t interpoType; // interpolate type
uint64_t defaultVal; // default value array list
......@@ -509,7 +509,6 @@ typedef struct {
int64_t pointsWritten;
uint8_t status;
uint8_t role;
uint8_t accessState;
uint8_t replica;
uint8_t reserved[5];
} SVnodeLoad;
......
src/inc/tsdb.h
浏览文件 @ 863bda56
......@@ -146,7 +146,7 @@ typedef struct STsdbQueryCond {
STimeWindow twindow;
int32_t order; // desc/asc order to iterate the data block
int32_t numOfCols;
SColumnInfoData *colList;
SColumnInfo *colList;
} STsdbQueryCond;
typedef struct SBlockInfo {
......
src/mnode/inc/mgmtLog.h
浏览文件 @ 863bda56
......@@ -21,6 +21,7 @@ extern "C" {
#endif
#include "tlog.h"
#include "monitor.h"
extern int32_t mdebugFlag;
extern int32_t sdbDebugFlag;
......@@ -41,9 +42,9 @@ extern int32_t sdbDebugFlag;
#define mPrint(...) \
{ taosPrintLog("MND ", 255, __VA_ARGS__); }
#define mLError(...) mError(__VA_ARGS__)
#define mLWarn(...) mWarn(__VA_ARGS__)
#define mLPrint(...) mPrint(__VA_ARGS__)
#define mLError(...) monitorSaveLog(2, __VA_ARGS__); mError(__VA_ARGS__)
#define mLWarn(...) monitorSaveLog(1, __VA_ARGS__); mWarn(__VA_ARGS__)
#define mLPrint(...) monitorSaveLog(0, __VA_ARGS__); mPrint(__VA_ARGS__)
#define sdbError(...) \
if (sdbDebugFlag & DEBUG_ERROR) { \
......@@ -60,9 +61,9 @@ extern int32_t sdbDebugFlag;
#define sdbPrint(...) \
{ taosPrintLog("MND-SDB ", 255, __VA_ARGS__); }
#define sdbLError(...) sdbError(__VA_ARGS__)
#define sdbLWarn(...) sdbWarn(__VA_ARGS__)
#define sdbLPrint(...) sdbPrint(__VA_ARGS__)
#define sdbLError(...) monitorSaveLog(2, __VA_ARGS__); sdbError(__VA_ARGS__)
#define sdbLWarn(...) monitorSaveLog(1, __VA_ARGS__); sdbWarn(__VA_ARGS__)
#define sdbLPrint(...) monitorSaveLog(0, __VA_ARGS__); sdbPrint(__VA_ARGS__)
#ifdef __cplusplus
}
......
src/mnode/src/mgmtDb.c
浏览文件 @ 863bda56
......@@ -884,6 +884,8 @@ void mgmtDropAllDbs(SAcctObj *pAcct) {
SDbObj *pDb = NULL;
void * pNode = NULL;
mPrint("acct:%s, all dbs will be dropped from sdb", pAcct->user);
while (1) {
pNode = sdbFetchRow(tsDbSdb, pNode, (void **)&pDb);
if (pDb == NULL) break;
......@@ -902,5 +904,5 @@ void mgmtDropAllDbs(SAcctObj *pAcct) {
mgmtDecDbRef(pDb);
}
mTrace("acct:%s, all dbs is is dropped from sdb", pAcct->user, numOfDbs);
mPrint("acct:%s, all dbs:%d is dropped from sdb", pAcct->user, numOfDbs);
}
src/mnode/src/mgmtDnode.c
浏览文件 @ 863bda56
......@@ -260,7 +260,7 @@ void mgmtProcessCfgDnodeMsg(SQueuedMsg *pMsg) {
}
uint32_t dnodeIp = inet_addr(pCmCfgDnode->ip);
if (strcmp(pMsg->pUser->pAcct->user, "root") != 0) {
if (strcmp(pMsg->pUser->user, "root") != 0) {
rpcRsp.code = TSDB_CODE_NO_RIGHTS;
} else {
SRpcIpSet ipSet = mgmtGetIpSetFromIp(dnodeIp);
......@@ -469,7 +469,7 @@ static void mgmtProcessCreateDnodeMsg(SQueuedMsg *pMsg) {
SCMCreateDnodeMsg *pCreate = pMsg->pCont;
if (strcmp(pMsg->pUser->pAcct->user, "root") != 0) {
if (strcmp(pMsg->pUser->user, "root") != 0) {
rpcRsp.code = TSDB_CODE_NO_RIGHTS;
} else {
uint32_t ip = inet_addr(pCreate->ip);
......@@ -489,7 +489,7 @@ static void mgmtProcessDropDnodeMsg(SQueuedMsg *pMsg) {
SRpcMsg rpcRsp = {.handle = pMsg->thandle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
SCMDropDnodeMsg *pDrop = pMsg->pCont;
if (strcmp(pMsg->pUser->pAcct->user, "root") != 0) {
if (strcmp(pMsg->pUser->user, "root") != 0) {
rpcRsp.code = TSDB_CODE_NO_RIGHTS;
} else {
uint32_t ip = inet_addr(pDrop->ip);
......
src/mnode/src/mgmtSdb.c
浏览文件 @ 863bda56
......@@ -400,6 +400,8 @@ static int32_t sdbInsertHash(SSdbTable *pTable, SSdbOper *pOper) {
if (pTable->keyType == SDB_KEY_AUTO) {
pTable->autoIndex = MAX(pTable->autoIndex, *((uint32_t *)pOper->pObj));
} else {
pTable->autoIndex++;
}
pthread_mutex_unlock(&pTable->mutex);
......
src/mnode/src/mgmtTable.c
浏览文件 @ 863bda56
......@@ -364,7 +364,7 @@ static void mgmtAddTableIntoStable(SSuperTableObj *pStable, SChildTableObj *pCta
bool find = false;
int32_t pos = 0;
for (int pos = 0; pos < pStable->vgLen; ++pos) {
for (pos = 0; pos < pStable->vgLen; ++pos) {
if (pStable->vgList[pos] == 0) break;
if (pStable->vgList[pos] == pCtable->vgId) {
find = true;
......@@ -730,6 +730,9 @@ static void mgmtProcessCreateSuperTableMsg(SQueuedMsg *pMsg) {
SSchema *tschema = pStable->schema;
tschema[col].colId = pStable->nextColId++;
tschema[col].bytes = htons(tschema[col].bytes);
// todo 1. check the length of each column; 2. check the total length of all columns
assert(tschema[col].type >= TSDB_DATA_TYPE_BOOL && tschema[col].type <= TSDB_DATA_TYPE_NCHAR);
}
SSdbOper oper = {
......@@ -1109,12 +1112,11 @@ void mgmtDropAllSuperTables(SDbObj *pDropDb) {
int32_t dbNameLen = strlen(pDropDb->name);
SSuperTableObj *pTable = NULL;
mPrint("db:%s, all super tables will be dropped from sdb", pDropDb->name);
while (1) {
mgmtDecTableRef(pTable);
pNode = sdbFetchRow(tsSuperTableSdb, pNode, (void **)&pTable);
if (pTable == NULL) {
break;
}
if (pTable == NULL) break;
if (strncmp(pDropDb->name, pTable->info.tableId, dbNameLen) == 0) {
SSdbOper oper = {
......@@ -1125,16 +1127,18 @@ void mgmtDropAllSuperTables(SDbObj *pDropDb) {
sdbDeleteRow(&oper);
pNode = pLastNode;
numOfTables ++;
continue;
}
mgmtDecTableRef(pTable);
}
mTrace("db:%s, all super tables:%d is dropped from sdb", pDropDb->name, numOfTables);
mPrint("db:%s, all super tables:%d is dropped from sdb", pDropDb->name, numOfTables);
}
static int32_t mgmtSetSchemaFromSuperTable(SSchema *pSchema, SSuperTableObj *pTable) {
int32_t numOfCols = pTable->numOfColumns + pTable->numOfTags;
for (int32_t i = 0; i < numOfCols; ++i) {
strcpy(pSchema->name, pTable->schema[i].name);
strncpy(pSchema->name, pTable->schema[i].name, TSDB_TABLE_ID_LEN);
pSchema->type = pTable->schema[i].type;
pSchema->bytes = htons(pTable->schema[i].bytes);
pSchema->colId = htons(pTable->schema[i].colId);
......@@ -1154,7 +1158,7 @@ static void mgmtGetSuperTableMeta(SQueuedMsg *pMsg) {
pMeta->numOfColumns = htons((int16_t)pTable->numOfColumns);
pMeta->tableType = pTable->info.type;
pMeta->contLen = sizeof(STableMetaMsg) + mgmtSetSchemaFromSuperTable(pMeta->schema, pTable);
strcpy(pMeta->tableId, pTable->info.tableId);
strncpy(pMeta->tableId, pTable->info.tableId, TSDB_TABLE_ID_LEN);
SRpcMsg rpcRsp = {
.handle = pMsg->thandle,
......@@ -1678,12 +1682,11 @@ void mgmtDropAllChildTables(SDbObj *pDropDb) {
int32_t dbNameLen = strlen(pDropDb->name);
SChildTableObj *pTable = NULL;
mPrint("db:%s, all child tables will be dropped from sdb", pDropDb->name);
while (1) {
mgmtDecTableRef(pTable);
pNode = sdbFetchRow(tsChildTableSdb, pNode, (void **)&pTable);
if (pTable == NULL) {
break;
}
if (pTable == NULL) break;
if (strncmp(pDropDb->name, pTable->info.tableId, dbNameLen) == 0) {
SSdbOper oper = {
......@@ -1694,11 +1697,11 @@ void mgmtDropAllChildTables(SDbObj *pDropDb) {
sdbDeleteRow(&oper);
pNode = pLastNode;
numOfTables++;
continue;
}
mgmtDecTableRef(pTable);
}
mTrace("db:%s, all child tables:%d is dropped from sdb", pDropDb->name, numOfTables);
mPrint("db:%s, all child tables:%d is dropped from sdb", pDropDb->name, numOfTables);
}
static void mgmtDropAllChildTablesInStable(SSuperTableObj *pStable) {
......@@ -1707,12 +1710,11 @@ static void mgmtDropAllChildTablesInStable(SSuperTableObj *pStable) {
int32_t numOfTables = 0;
SChildTableObj *pTable = NULL;
mPrint("stable:%s, all child tables will dropped from sdb", pStable->info.tableId, numOfTables);
while (1) {
mgmtDecTableRef(pTable);
pNode = sdbFetchRow(tsChildTableSdb, pNode, (void **)&pTable);
if (pTable == NULL) {
break;
}
if (pTable == NULL) break;
if (pTable->superTable == pStable) {
SSdbOper oper = {
......@@ -1723,11 +1725,12 @@ static void mgmtDropAllChildTablesInStable(SSuperTableObj *pStable) {
sdbDeleteRow(&oper);
pNode = pLastNode;
numOfTables++;
continue;
}
mgmtDecTableRef(pTable);
}
mTrace("stable:%s, all child tables:%d is dropped from sdb", pStable->info.tableId, numOfTables);
mPrint("stable:%s, all child tables:%d is dropped from sdb", pStable->info.tableId, numOfTables);
}
static SChildTableObj* mgmtGetTableByPos(uint32_t dnodeId, int32_t vnode, int32_t sid) {
......
src/mnode/src/mgmtUser.c
浏览文件 @ 863bda56
......@@ -353,6 +353,7 @@ static void mgmtProcessCreateUserMsg(SQueuedMsg *pMsg) {
mLPrint("user:%s, is created by %s", pCreate->user, pOperUser->user);
}
} else {
mError("user:%s, no rights to create user", pOperUser->user);
code = TSDB_CODE_NO_RIGHTS;
}
......@@ -398,6 +399,7 @@ static void mgmtProcessAlterUserMsg(SQueuedMsg *pMsg) {
code = mgmtUpdateUser(pUser);
mLPrint("user:%s, password is altered by %s, result:%s", pUser->user, pOperUser->user, tstrerror(code));
} else {
mError("user:%s, no rights to ater user", pOperUser->user);
code = TSDB_CODE_NO_RIGHTS;
}
......@@ -440,11 +442,13 @@ static void mgmtProcessAlterUserMsg(SQueuedMsg *pMsg) {
code = mgmtUpdateUser(pUser);
mLPrint("user:%s, privilege is altered by %s, result:%s", pUser->user, pOperUser->user, tstrerror(code));
} else {
mError("user:%s, no rights to ater user", pOperUser->user);
code = TSDB_CODE_NO_RIGHTS;
}
mgmtSendSimpleResp(pMsg->thandle, code);
} else {
mError("user:%s, no rights to ater user", pOperUser->user);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_NO_RIGHTS);
}
......
src/mnode/src/mgmtVgroup.c
浏览文件 @ 863bda56
......@@ -744,12 +744,13 @@ static void mgmtProcessVnodeCfgMsg(SRpcMsg *rpcMsg) {
void mgmtDropAllVgroups(SDbObj *pDropDb) {
void *pNode = NULL;
void *pLastNode = NULL;
int32_t numOfTables = 0;
int32_t numOfVgroups = 0;
int32_t dbNameLen = strlen(pDropDb->name);
SVgObj *pVgroup = NULL;
mPrint("db:%s, all vgroups will be dropped from sdb", pDropDb->name);
while (1) {
mgmtDecVgroupRef(pVgroup);
pNode = sdbFetchRow(tsVgroupSdb, pNode, (void **)&pVgroup);
if (pVgroup == NULL) break;
......@@ -761,12 +762,14 @@ void mgmtDropAllVgroups(SDbObj *pDropDb) {
};
sdbDeleteRow(&oper);
pNode = pLastNode;
numOfTables++;
continue;
numOfVgroups++;
}
mgmtSendDropVgroupMsg(pVgroup, NULL);
mgmtDecVgroupRef(pVgroup);
}
mTrace("db:%s, all vgroups is dropped from sdb", pDropDb->name, numOfTables);
mPrint("db:%s, all vgroups:%d is dropped from sdb", pDropDb->name, numOfVgroups);
}
void mgmtAlterVgroup(SVgObj *pVgroup, void *ahandle) {
......
src/plugins/http/inc/httpLog.h
浏览文件 @ 863bda56
......@@ -39,9 +39,4 @@ extern int32_t httpDebugFlag;
#define httpPrint(...) \
{ taosPrintLog("HTP ", 255, __VA_ARGS__); }
#define httpLError(...) taosLogError(__VA_ARGS__) httpError(__VA_ARGS__)
#define httpLWarn(...) taosLogWarn(__VA_ARGS__) httpWarn(__VA_ARGS__)
#define httpLPrint(...) taosLogPrint(__VA_ARGS__) httpPrint(__VA_ARGS__)
#endif
src/plugins/monitor/inc/monitorSystem.h 已删除 100644 → 0
浏览文件 @ 512572e8
此差异已折叠。
src/query/inc/queryExecutor.h
浏览文件 @ 863bda56
......@@ -90,7 +90,7 @@ typedef struct SColumnFilterElem {
} SColumnFilterElem;
typedef struct SSingleColumnFilterInfo {
SColumnInfoData info;
SColumnInfo info;
int32_t numOfFilters;
SColumnFilterElem* pFilters;
void* pData;
......@@ -122,21 +122,20 @@ typedef struct SQuery {
int64_t slidingTime; // sliding time for sliding window query
char slidingTimeUnit; // interval data type, used for daytime revise
int8_t precision;
int16_t numOfOutputCols;
int16_t numOfOutput;
int16_t interpoType;
int16_t checkBuffer; // check if the buffer is full during scan each block
SLimitVal limit;
int32_t rowSize;
SSqlGroupbyExpr* pGroupbyExpr;
SSqlFunctionExpr* pSelectExpr;
SColumnInfoData* colList;
SArithExprInfo* pSelectExpr;
SColumnInfo* colList;
int32_t numOfFilterCols;
int64_t* defaultVal;
TSKEY lastKey;
uint32_t status; // query status
SResultRec rec;
int32_t pos;
int64_t pointsOffset; // the number of points offset to save read data
SData** sdata;
SSingleColumnFilterInfo* pFilterInfo;
} SQuery;
......
src/query/inc/queryUtil.h
浏览文件 @ 863bda56
此差异已折叠。
src/query/inc/tsqlfunction.h
浏览文件 @ 863bda56
此差异已折叠。
src/query/src/queryExecutor.c
浏览文件 @ 863bda56
此差异已折叠。
src/query/src/queryFilterFunc.c 0 → 100644
浏览文件 @ 863bda56
此差异已折叠。
src/query/src/queryUtil.c
浏览文件 @ 863bda56
此差异已折叠。
src/tsdb/inc/tsdbMain.h
浏览文件 @ 863bda56
此差异已折叠。
src/tsdb/src/tsdbRWHelper.c
浏览文件 @ 863bda56
此差异已折叠。
src/tsdb/src/tsdbRead.c
浏览文件 @ 863bda56
此差异已折叠。
src/util/inc/tarray.h
浏览文件 @ 863bda56
此差异已折叠。
src/util/src/talgo.c
浏览文件 @ 863bda56
此差异已折叠。
src/util/src/tarray.c
浏览文件 @ 863bda56
此差异已折叠。
src/vnode/inc/vnodeLog.h
浏览文件 @ 863bda56
此差异已折叠。
src/vnode/src/vnodeMain.c
浏览文件 @ 863bda56
此差异已折叠。
tests/script/sh/deploy.sh
浏览文件 @ 863bda56
此差异已折叠。
tests/script/unique/account/monitor.sim 已删除 100644 → 0
浏览文件 @ 512572e8
此差异已折叠。
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或