Skip to content

T
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
提交 3e455e37 编写于 作者: L Liu Jicong
浏览文件
操作

Merge branch 'feature/shm' into feature/tq

上级 d3e62358 44faa52a
展开全部 隐藏空白更改
内联 并排
Showing with 1809 addition and 1015 deletion
include/common/taosdef.h
浏览文件 @ 3e455e37
......@@ -78,8 +78,6 @@ typedef enum {
extern char *qtypeStr[];
#define TSDB_PORT_DNODEDNODE 5
#define TSDB_PORT_SYNC 10
#define TSDB_PORT_HTTP 11
#ifdef __cplusplus
......
include/common/tmsg.h
浏览文件 @ 3e455e37
......@@ -184,6 +184,13 @@ typedef struct SField {
int32_t bytes;
} SField;
typedef struct SRetention {
int32_t freq;
int32_t keep;
int8_t freqUnit;
int8_t keepUnit;
} SRetention;
#pragma pack(push, 1)
// null-terminated string instead of char array to avoid too many memory consumption in case of more than 1M tableMeta
......@@ -272,11 +279,14 @@ typedef struct {
float xFilesFactor;
int32_t aggregationMethod;
int32_t delay;
int32_t ttl;
int32_t numOfColumns;
int32_t numOfTags;
int32_t numOfSmas;
int32_t commentLen;
SArray* pColumns;
SArray* pTags;
SArray* pColumns; // array of SField
SArray* pTags; // array of SField
SArray* pSmas; // array of SField
char* comment;
} SMCreateStbReq;
......@@ -467,7 +477,8 @@ typedef struct {
int32_t tz; // query client timezone
char intervalUnit;
char slidingUnit;
char offsetUnit;
char offsetUnit; // TODO Remove it, the offset is the number of precision tickle, and it must be a immutable duration.
int8_t precision;
int64_t interval;
int64_t sliding;
int64_t offset;
......@@ -503,10 +514,13 @@ typedef struct {
int8_t cacheLastRow;
int8_t ignoreExist;
int8_t streamMode;
int32_t numOfRetensions;
SArray* pRetensions; // SRetention
} SCreateDbReq;
int32_t tSerializeSCreateDbReq(void* buf, int32_t bufLen, SCreateDbReq* pReq);
int32_t tDeserializeSCreateDbReq(void* buf, int32_t bufLen, SCreateDbReq* pReq);
void tFreeSCreateDbReq(SCreateDbReq* pReq);
typedef struct {
char db[TSDB_DB_FNAME_LEN];
......@@ -751,11 +765,13 @@ typedef struct {
int8_t selfIndex;
int8_t streamMode;
SReplica replicas[TSDB_MAX_REPLICA];
int32_t numOfRetensions;
SArray* pRetensions; // SRetention
} SCreateVnodeReq, SAlterVnodeReq;
int32_t tSerializeSCreateVnodeReq(void* buf, int32_t bufLen, SCreateVnodeReq* pReq);
int32_t tDeserializeSCreateVnodeReq(void* buf, int32_t bufLen, SCreateVnodeReq* pReq);
int32_t tFreeSCreateVnodeReq(SCreateVnodeReq* pReq);
typedef struct {
int32_t vgId;
......
include/libs/function/function.h
浏览文件 @ 3e455e37
......@@ -327,7 +327,7 @@ bool taosFillHasMoreResults(struct SFillInfo* pFillInfo);
struct SFillInfo* taosCreateFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_t capacity, int32_t numOfCols,
int64_t slidingTime, int8_t slidingUnit, int8_t precision, int32_t fillType,
struct SFillColInfo* pFillCol, void* handle);
struct SFillColInfo* pFillCol, const char* id);
void* taosDestroyFillInfo(struct SFillInfo *pFillInfo);
int64_t taosFillResultDataBlock(struct SFillInfo* pFillInfo, void** output, int32_t capacity);
......
include/libs/nodes/nodes.h
浏览文件 @ 3e455e37
......@@ -135,6 +135,7 @@ typedef enum ENodeType {
QUERY_NODE_PHYSICAL_PLAN_EXCHANGE,
QUERY_NODE_PHYSICAL_PLAN_SORT,
QUERY_NODE_PHYSICAL_PLAN_INTERVAL,
QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW,
QUERY_NODE_PHYSICAL_PLAN_DISPATCH,
QUERY_NODE_PHYSICAL_PLAN_INSERT,
QUERY_NODE_PHYSICAL_SUBPLAN,
......@@ -169,6 +170,7 @@ void nodesDestroyNode(SNodeptr pNode);
SNodeList* nodesMakeList();
int32_t nodesListAppend(SNodeList* pList, SNodeptr pNode);
int32_t nodesListStrictAppend(SNodeList* pList, SNodeptr pNode);
int32_t nodesListMakeAppend(SNodeList** pList, SNodeptr pNode);
int32_t nodesListAppendList(SNodeList* pTarget, SNodeList* pSrc);
int32_t nodesListStrictAppendList(SNodeList* pTarget, SNodeList* pSrc);
SListCell* nodesListErase(SNodeList* pList, SListCell* pCell);
......
include/libs/nodes/plannodes.h
浏览文件 @ 3e455e37
......@@ -95,6 +95,7 @@ typedef struct SWindowLogicNode {
int8_t intervalUnit;
int8_t slidingUnit;
SFillNode* pFill;
int64_t sessionGap;
} SWindowLogicNode;
typedef enum ESubplanType {
......@@ -110,7 +111,7 @@ typedef struct SSubplanId {
int32_t subplanId;
} SSubplanId;
typedef struct SSubLogicPlan {
typedef struct SLogicSubplan {
ENodeType type;
SSubplanId id;
SNodeList* pChildren;
......@@ -120,7 +121,7 @@ typedef struct SSubLogicPlan {
SVgroupsInfo* pVgroupList;
int32_t level;
int32_t splitFlag;
} SSubLogicPlan;
} SLogicSubplan;
typedef struct SQueryLogicPlan {
ENodeType type;
......@@ -213,10 +214,14 @@ typedef struct SExchangePhysiNode {
SNodeList* pSrcEndPoints; // element is SDownstreamSource, scheduler fill by calling qSetSuplanExecutionNode
} SExchangePhysiNode;
typedef struct SIntervalPhysiNode {
typedef struct SWinodwPhysiNode {
SPhysiNode node;
SNodeList* pExprs; // these are expression list of parameter expression of function
SNodeList* pFuncs;
} SWinodwPhysiNode;
typedef struct SIntervalPhysiNode {
SWinodwPhysiNode window;
int64_t interval;
int64_t offset;
int64_t sliding;
......@@ -225,6 +230,11 @@ typedef struct SIntervalPhysiNode {
SFillNode* pFill;
} SIntervalPhysiNode;
typedef struct SSessionWinodwPhysiNode {
SWinodwPhysiNode window;
int64_t gap;
} SSessionWinodwPhysiNode;
typedef struct SDataSinkNode {
ENodeType type;
SDataBlockDescNode* pInputDataBlockDesc;
......
include/libs/nodes/querynodes.h
浏览文件 @ 3e455e37
......@@ -191,8 +191,8 @@ typedef struct SStateWindowNode {
typedef struct SSessionWindowNode {
ENodeType type; // QUERY_NODE_SESSION_WINDOW
int64_t gap; // gap between two session window(in microseconds)
SNode* pCol;
SNode* pGap; // gap between two session window(in microseconds)
} SSessionWindowNode;
typedef struct SIntervalWindowNode {
......
source/common/src/tmsg.c
浏览文件 @ 3e455e37
......@@ -287,6 +287,7 @@ int32_t tSerializeSVCreateTbReq(void **buf, SVCreateTbReq *pReq) {
int32_t tlen = 0;
tlen += taosEncodeFixedI64(buf, pReq->ver);
tlen += taosEncodeString(buf, pReq->dbFName);
tlen += taosEncodeString(buf, pReq->name);
tlen += taosEncodeFixedU32(buf, pReq->ttl);
tlen += taosEncodeFixedU32(buf, pReq->keep);
......@@ -360,6 +361,7 @@ int32_t tSerializeSVCreateTbReq(void **buf, SVCreateTbReq *pReq) {
void *tDeserializeSVCreateTbReq(void *buf, SVCreateTbReq *pReq) {
buf = taosDecodeFixedI64(buf, &(pReq->ver));
buf = taosDecodeString(buf, &(pReq->dbFName));
buf = taosDecodeString(buf, &(pReq->name));
buf = taosDecodeFixedU32(buf, &(pReq->ttl));
buf = taosDecodeFixedU32(buf, &(pReq->keep));
......@@ -478,7 +480,7 @@ void *tDeserializeSVCreateTbBatchReq(void *buf, SVCreateTbBatchReq *pReq) {
buf = taosDecodeFixedU32(buf, &nsize);
pReq->pArray = taosArrayInit(nsize, sizeof(SVCreateTbReq));
for (size_t i = 0; i < nsize; i++) {
SVCreateTbReq req;
SVCreateTbReq req = {0};
buf = tDeserializeSVCreateTbReq(buf, &req);
taosArrayPush(pReq->pArray, &req);
}
......@@ -511,8 +513,10 @@ int32_t tSerializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pReq
if (tEncodeFloat(&encoder, pReq->xFilesFactor) < 0) return -1;
if (tEncodeI32(&encoder, pReq->aggregationMethod) < 0) return -1;
if (tEncodeI32(&encoder, pReq->delay) < 0) return -1;
if (tEncodeI32(&encoder, pReq->ttl) < 0) return -1;
if (tEncodeI32(&encoder, pReq->numOfColumns) < 0) return -1;
if (tEncodeI32(&encoder, pReq->numOfTags) < 0) return -1;
if (tEncodeI32(&encoder, pReq->numOfSmas) < 0) return -1;
if (tEncodeI32(&encoder, pReq->commentLen) < 0) return -1;
for (int32_t i = 0; i < pReq->numOfColumns; ++i) {
......@@ -529,7 +533,16 @@ int32_t tSerializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pReq
if (tEncodeCStr(&encoder, pField->name) < 0) return -1;
}
if (tEncodeBinary(&encoder, pReq->comment, pReq->commentLen) < 0) return -1;
for (int32_t i = 0; i < pReq->numOfSmas; ++i) {
SField *pField = taosArrayGet(pReq->pSmas, i);
if (tEncodeI8(&encoder, pField->type) < 0) return -1;
if (tEncodeI32(&encoder, pField->bytes) < 0) return -1;
if (tEncodeCStr(&encoder, pField->name) < 0) return -1;
}
if (pReq->commentLen > 0) {
if (tEncodeBinary(&encoder, pReq->comment, pReq->commentLen) < 0) return -1;
}
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
......@@ -547,13 +560,16 @@ int32_t tDeserializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pR
if (tDecodeFloat(&decoder, &pReq->xFilesFactor) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->aggregationMethod) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->delay) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->ttl) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->numOfColumns) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->numOfTags) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->numOfSmas) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->commentLen) < 0) return -1;
pReq->pColumns = taosArrayInit(pReq->numOfColumns, sizeof(SField));
pReq->pTags = taosArrayInit(pReq->numOfTags, sizeof(SField));
if (pReq->pColumns == NULL || pReq->pTags == NULL) {
pReq->pSmas = taosArrayInit(pReq->numOfSmas, sizeof(SField));
if (pReq->pColumns == NULL || pReq->pTags == NULL || pReq->pSmas == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
......@@ -580,13 +596,23 @@ int32_t tDeserializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pR
}
}
for (int32_t i = 0; i < pReq->numOfSmas; ++i) {
SField field = {0};
if (tDecodeI8(&decoder, &field.type) < 0) return -1;
if (tDecodeI32(&decoder, &field.bytes) < 0) return -1;
if (tDecodeCStrTo(&decoder, field.name) < 0) return -1;
if (taosArrayPush(pReq->pSmas, &field) == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
}
if (pReq->commentLen > 0) {
pReq->comment = malloc(pReq->commentLen);
if (pReq->comment == NULL) return -1;
if (tDecodeCStrTo(&decoder, pReq->comment) < 0) return -1;
}
if (tDecodeCStrTo(&decoder, pReq->comment) < 0) return -1;
tEndDecode(&decoder);
tCoderClear(&decoder);
......@@ -596,8 +622,11 @@ int32_t tDeserializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pR
void tFreeSMCreateStbReq(SMCreateStbReq *pReq) {
taosArrayDestroy(pReq->pColumns);
taosArrayDestroy(pReq->pTags);
taosArrayDestroy(pReq->pSmas);
tfree(pReq->comment);
pReq->pColumns = NULL;
pReq->pTags = NULL;
pReq->pSmas = NULL;
}
int32_t tSerializeSMDropStbReq(void *buf, int32_t bufLen, SMDropStbReq *pReq) {
......@@ -1515,6 +1544,14 @@ int32_t tSerializeSCreateDbReq(void *buf, int32_t bufLen, SCreateDbReq *pReq) {
if (tEncodeI8(&encoder, pReq->cacheLastRow) < 0) return -1;
if (tEncodeI8(&encoder, pReq->ignoreExist) < 0) return -1;
if (tEncodeI8(&encoder, pReq->streamMode) < 0) return -1;
if (tEncodeI32(&encoder, pReq->numOfRetensions) < 0) return -1;
for (int32_t i = 0; i < pReq->numOfRetensions; ++i) {
SRetention *pRetension = taosArrayGet(pReq->pRetensions, i);
if (tEncodeI32(&encoder, pRetension->freq) < 0) return -1;
if (tEncodeI32(&encoder, pRetension->keep) < 0) return -1;
if (tEncodeI8(&encoder, pRetension->freqUnit) < 0) return -1;
if (tEncodeI8(&encoder, pRetension->keepUnit) < 0) return -1;
}
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
......@@ -1548,12 +1585,36 @@ int32_t tDeserializeSCreateDbReq(void *buf, int32_t bufLen, SCreateDbReq *pReq)
if (tDecodeI8(&decoder, &pReq->cacheLastRow) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->ignoreExist) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->streamMode) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->numOfRetensions) < 0) return -1;
pReq->pRetensions = taosArrayInit(pReq->numOfRetensions, sizeof(SRetention));
if (pReq->pRetensions == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
for (int32_t i = 0; i < pReq->numOfRetensions; ++i) {
SRetention rentension = {0};
if (tDecodeI32(&decoder, &rentension.freq) < 0) return -1;
if (tDecodeI32(&decoder, &rentension.keep) < 0) return -1;
if (tDecodeI8(&decoder, &rentension.freqUnit) < 0) return -1;
if (tDecodeI8(&decoder, &rentension.keepUnit) < 0) return -1;
if (taosArrayPush(pReq->pRetensions, &rentension) == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
}
tEndDecode(&decoder);
tCoderClear(&decoder);
return 0;
}
void tFreeSCreateDbReq(SCreateDbReq *pReq) {
taosArrayDestroy(pReq->pRetensions);
pReq->pRetensions = NULL;
}
int32_t tSerializeSAlterDbReq(void *buf, int32_t bufLen, SAlterDbReq *pReq) {
SCoder encoder = {0};
tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER);
......@@ -2431,6 +2492,14 @@ int32_t tSerializeSCreateVnodeReq(void *buf, int32_t bufLen, SCreateVnodeReq *pR
SReplica *pReplica = &pReq->replicas[i];
if (tEncodeSReplica(&encoder, pReplica) < 0) return -1;
}
if (tEncodeI32(&encoder, pReq->numOfRetensions) < 0) return -1;
for (int32_t i = 0; i < pReq->numOfRetensions; ++i) {
SRetention *pRetension = taosArrayGet(pReq->pRetensions, i);
if (tEncodeI32(&encoder, pRetension->freq) < 0) return -1;
if (tEncodeI32(&encoder, pRetension->keep) < 0) return -1;
if (tEncodeI8(&encoder, pRetension->freqUnit) < 0) return -1;
if (tEncodeI8(&encoder, pRetension->keepUnit) < 0) return -1;
}
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
......@@ -2475,11 +2544,35 @@ int32_t tDeserializeSCreateVnodeReq(void *buf, int32_t bufLen, SCreateVnodeReq *
if (tDecodeSReplica(&decoder, pReplica) < 0) return -1;
}
if (tDecodeI32(&decoder, &pReq->numOfRetensions) < 0) return -1;
pReq->pRetensions = taosArrayInit(pReq->numOfRetensions, sizeof(SRetention));
if (pReq->pRetensions == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
for (int32_t i = 0; i < pReq->numOfRetensions; ++i) {
SRetention rentension = {0};
if (tDecodeI32(&decoder, &rentension.freq) < 0) return -1;
if (tDecodeI32(&decoder, &rentension.keep) < 0) return -1;
if (tDecodeI8(&decoder, &rentension.freqUnit) < 0) return -1;
if (tDecodeI8(&decoder, &rentension.keepUnit) < 0) return -1;
if (taosArrayPush(pReq->pRetensions, &rentension) == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
}
tEndDecode(&decoder);
tCoderClear(&decoder);
return 0;
}
int32_t tFreeSCreateVnodeReq(SCreateVnodeReq *pReq) {
taosArrayDestroy(pReq->pRetensions);
pReq->pRetensions = NULL;
}
int32_t tSerializeSDropVnodeReq(void *buf, int32_t bufLen, SDropVnodeReq *pReq) {
SCoder encoder = {0};
tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER);
......
source/dnode/mgmt/test/vnode/vnode.cpp
浏览文件 @ 3e455e37
......@@ -158,6 +158,7 @@ TEST_F(DndTestVnode, 03_Create_Stb) {
for (int i = 0; i < 1; ++i) {
SVCreateTbReq req = {0};
req.ver = 0;
req.dbFName = (char*)"1.db1";
req.name = (char*)"stb1";
req.ttl = 0;
req.keep = 0;
......@@ -229,6 +230,7 @@ TEST_F(DndTestVnode, 04_Alter_Stb) {
for (int i = 0; i < 1; ++i) {
SVCreateTbReq req = {0};
req.ver = 0;
req.dbFName = (char*)"1.db1";
req.name = (char*)"stb1";
req.ttl = 0;
req.keep = 0;
......
source/dnode/mnode/impl/inc/mndDef.h
浏览文件 @ 3e455e37
......@@ -267,6 +267,8 @@ typedef struct {
int8_t update;
int8_t cacheLastRow;
int8_t streamMode;
int32_t numOfRetensions;
SArray* pRetensions;
} SDbCfg;
typedef struct {
......@@ -344,11 +346,14 @@ typedef struct {
float xFilesFactor;
int32_t aggregationMethod;
int32_t delay;
int32_t ttl;
int32_t numOfColumns;
int32_t numOfTags;
int32_t numOfSmas;
int32_t commentLen;
SSchema* pColumns;
SSchema* pTags;
SSchema* pSmas;
char* comment;
SRWLatch lock;
} SStbObj;
......
source/dnode/mnode/impl/src/mndDb.c
浏览文件 @ 3e455e37
......@@ -100,6 +100,15 @@ static SSdbRaw *mndDbActionEncode(SDbObj *pDb) {
SDB_SET_INT8(pRaw, dataPos, pDb->cfg.quorum, DB_ENCODE_OVER)
SDB_SET_INT8(pRaw, dataPos, pDb->cfg.update, DB_ENCODE_OVER)
SDB_SET_INT8(pRaw, dataPos, pDb->cfg.cacheLastRow, DB_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pDb->cfg.numOfRetensions, DB_ENCODE_OVER)
for (int32_t i = 0; i < pDb->cfg.numOfRetensions; ++i) {
SRetention *pRetension = taosArrayGet(pDb->cfg.pRetensions, i);
SDB_SET_INT32(pRaw, dataPos, pRetension->freq, DB_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pRetension->keep, DB_ENCODE_OVER)
SDB_SET_INT8(pRaw, dataPos, pRetension->freqUnit, DB_ENCODE_OVER)
SDB_SET_INT8(pRaw, dataPos, pRetension->keepUnit, DB_ENCODE_OVER)
}
SDB_SET_RESERVE(pRaw, dataPos, TSDB_DB_RESERVE_SIZE, DB_ENCODE_OVER)
SDB_SET_DATALEN(pRaw, dataPos, DB_ENCODE_OVER)
......@@ -161,6 +170,22 @@ static SSdbRow *mndDbActionDecode(SSdbRaw *pRaw) {
SDB_GET_INT8(pRaw, dataPos, &pDb->cfg.quorum, DB_DECODE_OVER)
SDB_GET_INT8(pRaw, dataPos, &pDb->cfg.update, DB_DECODE_OVER)
SDB_GET_INT8(pRaw, dataPos, &pDb->cfg.cacheLastRow, DB_DECODE_OVER)
SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.numOfRetensions, DB_DECODE_OVER)
if (pDb->cfg.numOfRetensions > 0) {
pDb->cfg.pRetensions = taosArrayInit(pDb->cfg.numOfRetensions, sizeof(SRetention));
if (pDb->cfg.pRetensions == NULL) goto DB_DECODE_OVER;
for (int32_t i = 0; i < pDb->cfg.numOfRetensions; ++i) {
SRetention retension = {0};
SDB_GET_INT32(pRaw, dataPos, &retension.freq, DB_DECODE_OVER)
SDB_GET_INT32(pRaw, dataPos, &retension.keep, DB_DECODE_OVER)
SDB_GET_INT8(pRaw, dataPos, &retension.freqUnit, DB_DECODE_OVER)
SDB_GET_INT8(pRaw, dataPos, &retension.keepUnit, DB_DECODE_OVER)
if (taosArrayPush(pDb->cfg.pRetensions, &retension) == NULL) {
goto DB_DECODE_OVER;
}
}
}
SDB_GET_RESERVE(pRaw, dataPos, TSDB_DB_RESERVE_SIZE, DB_DECODE_OVER)
terrno = 0;
......@@ -183,6 +208,7 @@ static int32_t mndDbActionInsert(SSdb *pSdb, SDbObj *pDb) {
static int32_t mndDbActionDelete(SSdb *pSdb, SDbObj *pDb) {
mTrace("db:%s, perform delete action, row:%p", pDb->name, pDb);
taosArrayDestroy(pDb->cfg.pRetensions);
return 0;
}
......@@ -417,6 +443,10 @@ static int32_t mndCreateDb(SMnode *pMnode, SNodeMsg *pReq, SCreateDbReq *pCreate
.streamMode = pCreate->streamMode,
};
dbObj.cfg.numOfRetensions = pCreate->numOfRetensions;
dbObj.cfg.pRetensions = pCreate->pRetensions;
pCreate = NULL;
mndSetDefaultDbCfg(&dbObj.cfg);
if (mndCheckDbName(dbObj.name, pUser) != 0) {
......@@ -505,6 +535,7 @@ CREATE_DB_OVER:
mndReleaseDb(pMnode, pDb);
mndReleaseUser(pMnode, pUser);
tFreeSCreateDbReq(&createReq);
return code;
}
......
source/dnode/mnode/impl/src/mndStb.c
浏览文件 @ 3e455e37
......@@ -72,7 +72,8 @@ void mndCleanupStb(SMnode *pMnode) {}
SSdbRaw *mndStbActionEncode(SStbObj *pStb) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
int32_t size = sizeof(SStbObj) + (pStb->numOfColumns + pStb->numOfTags) * sizeof(SSchema) + TSDB_STB_RESERVE_SIZE;
int32_t size = sizeof(SStbObj) + (pStb->numOfColumns + pStb->numOfTags + pStb->numOfSmas) * sizeof(SSchema) +
TSDB_STB_RESERVE_SIZE;
SSdbRaw *pRaw = sdbAllocRaw(SDB_STB, TSDB_STB_VER_NUMBER, size);
if (pRaw == NULL) goto STB_ENCODE_OVER;
......@@ -88,8 +89,10 @@ SSdbRaw *mndStbActionEncode(SStbObj *pStb) {
SDB_SET_INT32(pRaw, dataPos, (int32_t)(pStb->xFilesFactor * 10000), STB_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pStb->aggregationMethod, STB_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pStb->delay, STB_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pStb->ttl, STB_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pStb->numOfColumns, STB_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pStb->numOfTags, STB_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pStb->numOfSmas, STB_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pStb->commentLen, STB_ENCODE_OVER)
for (int32_t i = 0; i < pStb->numOfColumns; ++i) {
......@@ -108,7 +111,17 @@ SSdbRaw *mndStbActionEncode(SStbObj *pStb) {
SDB_SET_BINARY(pRaw, dataPos, pSchema->name, TSDB_COL_NAME_LEN, STB_ENCODE_OVER)
}
SDB_SET_BINARY(pRaw, dataPos, pStb->comment, pStb->commentLen, STB_ENCODE_OVER)
for (int32_t i = 0; i < pStb->numOfSmas; ++i) {
SSchema *pSchema = &pStb->pSmas[i];
SDB_SET_INT8(pRaw, dataPos, pSchema->type, STB_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pSchema->colId, STB_ENCODE_OVER)
SDB_SET_INT32(pRaw, dataPos, pSchema->bytes, STB_ENCODE_OVER)
SDB_SET_BINARY(pRaw, dataPos, pSchema->name, TSDB_COL_NAME_LEN, STB_ENCODE_OVER)
}
if (pStb->commentLen > 0) {
SDB_SET_BINARY(pRaw, dataPos, pStb->comment, pStb->commentLen, STB_ENCODE_OVER)
}
SDB_SET_RESERVE(pRaw, dataPos, TSDB_STB_RESERVE_SIZE, STB_ENCODE_OVER)
SDB_SET_DATALEN(pRaw, dataPos, STB_ENCODE_OVER)
......@@ -156,13 +169,16 @@ static SSdbRow *mndStbActionDecode(SSdbRaw *pRaw) {
pStb->xFilesFactor = xFilesFactor / 10000.0f;
SDB_GET_INT32(pRaw, dataPos, &pStb->aggregationMethod, STB_DECODE_OVER)
SDB_GET_INT32(pRaw, dataPos, &pStb->delay, STB_DECODE_OVER)
SDB_GET_INT32(pRaw, dataPos, &pStb->ttl, STB_DECODE_OVER)
SDB_GET_INT32(pRaw, dataPos, &pStb->numOfColumns, STB_DECODE_OVER)
SDB_GET_INT32(pRaw, dataPos, &pStb->numOfTags, STB_DECODE_OVER)
SDB_GET_INT32(pRaw, dataPos, &pStb->numOfSmas, STB_DECODE_OVER)
SDB_GET_INT32(pRaw, dataPos, &pStb->commentLen, STB_DECODE_OVER)
pStb->pColumns = calloc(pStb->numOfColumns, sizeof(SSchema));
pStb->pTags = calloc(pStb->numOfTags, sizeof(SSchema));
if (pStb->pColumns == NULL || pStb->pTags == NULL) {
pStb->pSmas = calloc(pStb->numOfSmas, sizeof(SSchema));
if (pStb->pColumns == NULL || pStb->pTags == NULL || pStb->pSmas == NULL) {
goto STB_DECODE_OVER;
}
......@@ -182,6 +198,14 @@ static SSdbRow *mndStbActionDecode(SSdbRaw *pRaw) {
SDB_GET_BINARY(pRaw, dataPos, pSchema->name, TSDB_COL_NAME_LEN, STB_DECODE_OVER)
}
for (int32_t i = 0; i < pStb->numOfSmas; ++i) {
SSchema *pSchema = &pStb->pSmas[i];
SDB_GET_INT8(pRaw, dataPos, &pSchema->type, STB_DECODE_OVER)
SDB_GET_INT32(pRaw, dataPos, &pSchema->colId, STB_DECODE_OVER)
SDB_GET_INT32(pRaw, dataPos, &pSchema->bytes, STB_DECODE_OVER)
SDB_GET_BINARY(pRaw, dataPos, pSchema->name, TSDB_COL_NAME_LEN, STB_DECODE_OVER)
}
if (pStb->commentLen > 0) {
pStb->comment = calloc(pStb->commentLen, 1);
if (pStb->comment == NULL) goto STB_DECODE_OVER;
......@@ -247,6 +271,30 @@ static int32_t mndStbActionUpdate(SSdb *pSdb, SStbObj *pOld, SStbObj *pNew) {
}
}
if (pOld->numOfSmas < pNew->numOfSmas) {
void *pSmas = malloc(pNew->numOfSmas * sizeof(SSchema));
if (pSmas != NULL) {
free(pOld->pSmas);
pOld->pSmas = pSmas;
} else {
terrno = TSDB_CODE_OUT_OF_MEMORY;
mTrace("stb:%s, failed to perform update action since %s", pOld->name, terrstr());
taosWUnLockLatch(&pOld->lock);
}
}
if (pOld->commentLen < pNew->commentLen) {
void *comment = malloc(pNew->commentLen);
if (comment != NULL) {
free(pOld->comment);
pOld->comment = comment;
} else {
terrno = TSDB_CODE_OUT_OF_MEMORY;
mTrace("stb:%s, failed to perform update action since %s", pOld->name, terrstr());
taosWUnLockLatch(&pOld->lock);
}
}
pOld->updateTime = pNew->updateTime;
pOld->version = pNew->version;
pOld->nextColId = pNew->nextColId;
......@@ -254,7 +302,9 @@ static int32_t mndStbActionUpdate(SSdb *pSdb, SStbObj *pOld, SStbObj *pNew) {
pOld->numOfTags = pNew->numOfTags;
memcpy(pOld->pColumns, pNew->pColumns, pOld->numOfColumns * sizeof(SSchema));
memcpy(pOld->pTags, pNew->pTags, pOld->numOfTags * sizeof(SSchema));
memcpy(pOld->comment, pNew->comment, TSDB_STB_COMMENT_LEN);
if (pNew->commentLen != 0) {
memcpy(pOld->comment, pNew->comment, TSDB_STB_COMMENT_LEN);
}
taosWUnLockLatch(&pOld->lock);
return 0;
}
......@@ -286,9 +336,12 @@ static SDbObj *mndAcquireDbByStb(SMnode *pMnode, const char *stbName) {
static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb, int32_t *pContLen) {
SName name = {0};
tNameFromString(&name, pStb->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
char dbFName[TSDB_DB_FNAME_LEN] = {0};
tNameGetFullDbName(&name, dbFName);
SVCreateTbReq req = {0};
req.ver = 0;
req.dbFName = dbFName;
req.name = (char *)tNameGetTableName(&name);
req.ttl = 0;
req.keep = 0;
......@@ -505,6 +558,16 @@ static int32_t mndSetCreateStbUndoActions(SMnode *pMnode, STrans *pTrans, SDbObj
return 0;
}
static SSchema *mndFindStbColumns(const SStbObj *pStb, const char *colName) {
for (int32_t col = 0; col < pStb->numOfColumns; col++) {
SSchema *pSchema = &pStb->pColumns[col];
if (strcasecmp(pStb->pColumns[col].name, colName) == 0) {
return pSchema;
}
}
return NULL;
}
static int32_t mndCreateStb(SMnode *pMnode, SNodeMsg *pReq, SMCreateStbReq *pCreate, SDbObj *pDb) {
SStbObj stbObj = {0};
memcpy(stbObj.name, pCreate->name, TSDB_TABLE_FNAME_LEN);
......@@ -515,19 +578,24 @@ static int32_t mndCreateStb(SMnode *pMnode, SNodeMsg *pReq, SMCreateStbReq *pCre
stbObj.dbUid = pDb->uid;
stbObj.version = 1;
stbObj.nextColId = 1;
stbObj.ttl = pCreate->ttl;
stbObj.numOfColumns = pCreate->numOfColumns;
stbObj.numOfTags = pCreate->numOfTags;
stbObj.numOfSmas = pCreate->numOfSmas;
stbObj.commentLen = pCreate->commentLen;
stbObj.comment = calloc(stbObj.commentLen, 1);
if (stbObj.comment == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
if (stbObj.commentLen > 0) {
stbObj.comment = calloc(stbObj.commentLen, 1);
if (stbObj.comment == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
memcpy(stbObj.comment, pCreate->comment, stbObj.commentLen);
}
memcpy(stbObj.comment, pCreate->comment, stbObj.commentLen);
stbObj.pColumns = malloc(stbObj.numOfColumns * sizeof(SSchema));
stbObj.pTags = malloc(stbObj.numOfTags * sizeof(SSchema));
if (stbObj.pColumns == NULL || stbObj.pTags == NULL) {
stbObj.pSmas = malloc(stbObj.numOfSmas * sizeof(SSchema));
if (stbObj.pColumns == NULL || stbObj.pTags == NULL || stbObj.pSmas == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
......@@ -552,6 +620,18 @@ static int32_t mndCreateStb(SMnode *pMnode, SNodeMsg *pReq, SMCreateStbReq *pCre
stbObj.nextColId++;
}
for (int32_t i = 0; i < stbObj.numOfSmas; ++i) {
SField *pField = taosArrayGet(pCreate->pSmas, i);
SSchema *pSchema = &stbObj.pSmas[i];
SSchema *pColSchema = mndFindStbColumns(&stbObj, pField->name);
if (pColSchema == NULL) {
mError("stb:%s, sma:%s not found in columns", stbObj.name, pSchema->name);
terrno = TSDB_CODE_MND_INVALID_STB_OPTION;
return -1;
}
memcpy(pSchema, pColSchema, sizeof(SSchema));
}
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_STB, &pReq->rpcMsg);
if (pTrans == NULL) goto CREATE_STB_OVER;
......@@ -1536,19 +1616,11 @@ static int32_t mndRetrieveStb(SNodeMsg *pReq, SShowObj *pShow, char *data, int32
if (pDb == NULL) return 0;
}
tstrncpy(prefix, pShow->db, TSDB_DB_FNAME_LEN);
strcat(prefix, TS_PATH_DELIMITER);
int32_t prefixLen = (int32_t)strlen(prefix);
while (numOfRows < rows) {
pShow->pIter = sdbFetch(pSdb, SDB_STB, pShow->pIter, (void **)&pStb);
if (pShow->pIter == NULL) break;
if (pDb != NULL && pStb->dbUid != pDb->uid) {
if (strncmp(pStb->db, pDb->name, prefixLen) == 0) {
mError("Inconsistent table data, name:%s, db:%s, dbUid:%" PRIu64, pStb->name, pDb->name, pDb->uid);
}
sdbRelease(pSdb, pStb);
continue;
}
......
source/dnode/mnode/impl/src/mndVgroup.c
浏览文件 @ 3e455e37
......@@ -218,6 +218,8 @@ void *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVg
createReq.hashBegin = pVgroup->hashBegin;
createReq.hashEnd = pVgroup->hashEnd;
createReq.hashMethod = pDb->hashMethod;
createReq.numOfRetensions = pDb->cfg.numOfRetensions;
createReq.pRetensions = pDb->cfg.pRetensions;
for (int32_t v = 0; v < pVgroup->replica; ++v) {
SReplica *pReplica = &createReq.replicas[v];
......
source/dnode/mnode/impl/test/sma/sma.cpp
浏览文件 @ 3e455e37
......@@ -26,9 +26,12 @@ class MndTestSma : public ::testing::Test {
void* BuildDropDbReq(const char* dbname, int32_t* pContLen);
void* BuildCreateStbReq(const char* stbname, int32_t* pContLen);
void* BuildDropStbReq(const char* stbname, int32_t* pContLen);
void* BuildCreateSmaReq(const char* smaname, const char* stbname, int8_t igExists, const char* expr,
const char* tagsFilter, const char* sql, const char* ast, int32_t* pContLen);
void* BuildDropSmaReq(const char* smaname, int8_t igNotExists, int32_t* pContLen);
void* BuildCreateBSmaStbReq(const char* stbname, int32_t* pContLen);
void* BuildCreateTSmaReq(const char* smaname, const char* stbname, int8_t igExists, const char* expr,
const char* tagsFilter, const char* sql, const char* ast, int32_t* pContLen);
void* BuildDropTSmaReq(const char* smaname, int8_t igNotExists, int32_t* pContLen);
void PushField(SArray* pArray, int32_t bytes, int8_t type, const char* name);
};
Testbase MndTestSma::test;
......@@ -76,6 +79,14 @@ void* MndTestSma::BuildDropDbReq(const char* dbname, int32_t* pContLen) {
return pReq;
}
void MndTestSma::PushField(SArray* pArray, int32_t bytes, int8_t type, const char* name) {
SField field = {0};
field.bytes = bytes;
field.type = type;
strcpy(field.name, name);
taosArrayPush(pArray, &field);
}
void* MndTestSma::BuildCreateStbReq(const char* stbname, int32_t* pContLen) {
SMCreateStbReq createReq = {0};
createReq.numOfColumns = 3;
......@@ -85,37 +96,35 @@ void* MndTestSma::BuildCreateStbReq(const char* stbname, int32_t* pContLen) {
createReq.pTags = taosArrayInit(createReq.numOfTags, sizeof(SField));
strcpy(createReq.name, stbname);
{
SField field = {0};
field.bytes = 8;
field.type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(field.name, "ts");
taosArrayPush(createReq.pColumns, &field);
}
PushField(createReq.pColumns, 8, TSDB_DATA_TYPE_TIMESTAMP, "ts");
PushField(createReq.pColumns, 2, TSDB_DATA_TYPE_TINYINT, "col1");
PushField(createReq.pColumns, 8, TSDB_DATA_TYPE_BIGINT, "col2");
PushField(createReq.pTags, 2, TSDB_DATA_TYPE_TINYINT, "tag1");
{
SField field = {0};
field.bytes = 2;
field.type = TSDB_DATA_TYPE_TINYINT;
strcpy(field.name, "col1");
taosArrayPush(createReq.pColumns, &field);
}
int32_t tlen = tSerializeSMCreateStbReq(NULL, 0, &createReq);
void* pHead = rpcMallocCont(tlen);
tSerializeSMCreateStbReq(pHead, tlen, &createReq);
tFreeSMCreateStbReq(&createReq);
*pContLen = tlen;
return pHead;
}
{
SField field = {0};
field.bytes = 8;
field.type = TSDB_DATA_TYPE_BIGINT;
strcpy(field.name, "col2");
taosArrayPush(createReq.pColumns, &field);
}
void* MndTestSma::BuildCreateBSmaStbReq(const char* stbname, int32_t* pContLen) {
SMCreateStbReq createReq = {0};
createReq.numOfColumns = 3;
createReq.numOfTags = 1;
createReq.numOfSmas = 1;
createReq.igExists = 0;
createReq.pColumns = taosArrayInit(createReq.numOfColumns, sizeof(SField));
createReq.pTags = taosArrayInit(createReq.numOfTags, sizeof(SField));
createReq.pSmas = taosArrayInit(createReq.numOfSmas, sizeof(SField));
strcpy(createReq.name, stbname);
{
SField field = {0};
field.bytes = 2;
field.type = TSDB_DATA_TYPE_TINYINT;
strcpy(field.name, "tag1");
taosArrayPush(createReq.pTags, &field);
}
PushField(createReq.pColumns, 8, TSDB_DATA_TYPE_TIMESTAMP, "ts");
PushField(createReq.pColumns, 2, TSDB_DATA_TYPE_TINYINT, "col1");
PushField(createReq.pColumns, 8, TSDB_DATA_TYPE_BIGINT, "col2");
PushField(createReq.pTags, 2, TSDB_DATA_TYPE_TINYINT, "tag1");
PushField(createReq.pSmas, 2, TSDB_DATA_TYPE_TINYINT, "col1");
int32_t tlen = tSerializeSMCreateStbReq(NULL, 0, &createReq);
void* pHead = rpcMallocCont(tlen);
......@@ -137,8 +146,8 @@ void* MndTestSma::BuildDropStbReq(const char* stbname, int32_t* pContLen) {
return pReq;
}
void* MndTestSma::BuildCreateSmaReq(const char* smaname, const char* stbname, int8_t igExists, const char* expr,
const char* tagsFilter, const char* sql, const char* ast, int32_t* pContLen) {
void* MndTestSma::BuildCreateTSmaReq(const char* smaname, const char* stbname, int8_t igExists, const char* expr,
const char* tagsFilter, const char* sql, const char* ast, int32_t* pContLen) {
SMCreateSmaReq createReq = {0};
strcpy(createReq.name, smaname);
strcpy(createReq.stb, stbname);
......@@ -166,7 +175,7 @@ void* MndTestSma::BuildCreateSmaReq(const char* smaname, const char* stbname, in
return pHead;
}
void* MndTestSma::BuildDropSmaReq(const char* smaname, int8_t igNotExists, int32_t* pContLen) {
void* MndTestSma::BuildDropTSmaReq(const char* smaname, int8_t igNotExists, int32_t* pContLen) {
SMDropSmaReq dropsmaReq = {0};
dropsmaReq.igNotExists = igNotExists;
strcpy(dropsmaReq.name, smaname);
......@@ -180,6 +189,7 @@ void* MndTestSma::BuildDropSmaReq(const char* smaname, int8_t igNotExists, int32
}
TEST_F(MndTestSma, 01_Create_Show_Meta_Drop_Restart_Stb) {
#if 0
const char* dbname = "1.d1";
const char* stbname = "1.d1.stb";
const char* smaname = "1.d1.sma";
......@@ -201,9 +211,9 @@ TEST_F(MndTestSma, 01_Create_Show_Meta_Drop_Restart_Stb) {
test.SendShowRetrieveReq();
EXPECT_EQ(test.GetShowRows(), 1);
}
#if 0
{
pReq = BuildCreateSmaReq(smaname, stbname, 0, "expr", "tagsFilter", "sql", "ast", &contLen);
pReq = BuildCreateTSmaReq(smaname, stbname, 0, "expr", "tagsFilter", "sql", "ast", &contLen);
pRsp = test.SendReq(TDMT_MND_CREATE_SMA, pReq, contLen);
ASSERT_EQ(pRsp->code, 0);
test.SendShowMetaReq(TSDB_MGMT_TABLE_INDEX, dbname);
......@@ -226,7 +236,7 @@ TEST_F(MndTestSma, 01_Create_Show_Meta_Drop_Restart_Stb) {
}
{
pReq = BuildDropSmaReq(smaname, 0, &contLen);
pReq = BuildDropTSmaReq(smaname, 0, &contLen);
pRsp = test.SendReq(TDMT_MND_DROP_SMA, pReq, contLen);
ASSERT_EQ(pRsp->code, 0);
test.SendShowMetaReq(TSDB_MGMT_TABLE_INDEX, dbname);
......@@ -235,3 +245,50 @@ TEST_F(MndTestSma, 01_Create_Show_Meta_Drop_Restart_Stb) {
}
#endif
}
TEST_F(MndTestSma, 02_Create_Show_Meta_Drop_Restart_BSma) {
const char* dbname = "1.d1";
const char* stbname = "1.d1.bsmastb";
int32_t contLen = 0;
void* pReq;
SRpcMsg* pRsp;
{
pReq = BuildCreateDbReq(dbname, &contLen);
pRsp = test.SendReq(TDMT_MND_CREATE_DB, pReq, contLen);
ASSERT_EQ(pRsp->code, 0);
}
{
pReq = BuildCreateBSmaStbReq(stbname, &contLen);
pRsp = test.SendReq(TDMT_MND_CREATE_STB, pReq, contLen);
ASSERT_EQ(pRsp->code, 0);
test.SendShowMetaReq(TSDB_MGMT_TABLE_STB, dbname);
test.SendShowRetrieveReq();
EXPECT_EQ(test.GetShowRows(), 1);
// CheckBinary("bsmastb", TSDB_TABLE_NAME_LEN);
}
test.Restart();
{
pReq = BuildCreateBSmaStbReq(stbname, &contLen);
pRsp = test.SendReq(TDMT_MND_CREATE_STB, pReq, contLen);
ASSERT_EQ(pRsp->code, TSDB_CODE_MND_STB_ALREADY_EXIST);
}
{
pReq = BuildDropStbReq(stbname, &contLen);
pRsp = test.SendReq(TDMT_MND_DROP_STB, pReq, contLen);
ASSERT_EQ(pRsp->code, 0);
test.SendShowMetaReq(TSDB_MGMT_TABLE_STB, dbname);
test.SendShowRetrieveReq();
EXPECT_EQ(test.GetShowRows(), 0);
}
{
pReq = BuildDropStbReq(stbname, &contLen);
pRsp = test.SendReq(TDMT_MND_DROP_STB, pReq, contLen);
ASSERT_EQ(pRsp->code, TSDB_CODE_MND_STB_NOT_EXIST);
}
}
source/dnode/vnode/src/tsdb/tsdbRead.c
浏览文件 @ 3e455e37
......@@ -1377,7 +1377,6 @@ static int doBinarySearchKey(char* pValue, int num, TSKEY key, int order) {
}
static int32_t doCopyRowsFromFileBlock(STsdbReadHandle* pTsdbReadHandle, int32_t capacity, int32_t numOfRows, int32_t start, int32_t end) {
char* pData = NULL;
int32_t step = ASCENDING_TRAVERSE(pTsdbReadHandle->order)? 1 : -1;
SDataCols* pCols = pTsdbReadHandle->rhelper.pDCols[0];
......@@ -1454,14 +1453,12 @@ static int32_t doCopyRowsFromFileBlock(STsdbReadHandle* pTsdbReadHandle, int32_t
return numOfRows + num;
}
// TODO fix bug for reverse copy data
// TODO handle the null data
// TODO fix bug for reverse copy data problem
// Note: row1 always has high priority
static void mergeTwoRowFromMem(STsdbReadHandle* pTsdbReadHandle, int32_t capacity, int32_t numOfRows, STSRow* row1,
STSRow* row2, int32_t numOfCols, uint64_t uid, STSchema* pSchema1, STSchema* pSchema2,
bool forceSetNull) {
#if 1
char* pData = NULL;
STSchema* pSchema;
STSRow* row;
int16_t colId;
......@@ -1503,12 +1500,6 @@ static void mergeTwoRowFromMem(STsdbReadHandle* pTsdbReadHandle, int32_t capacit
while(i < numOfCols && (j < numOfColsOfRow1 || k < numOfColsOfRow2)) {
SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
pData = (char*)pColInfo->pData + numOfRows * pColInfo->info.bytes;
} else {
pData = (char*)pColInfo->pData + (capacity - numOfRows - 1) * pColInfo->info.bytes;
}
int32_t colIdOfRow1;
if(j >= numOfColsOfRow1) {
colIdOfRow1 = INT32_MAX;
......@@ -1571,43 +1562,11 @@ static void mergeTwoRowFromMem(STsdbReadHandle* pTsdbReadHandle, int32_t capacit
if (colId == pColInfo->info.colId) {
if (tdValTypeIsNorm(sVal.valType)) {
switch (pColInfo->info.type) {
case TSDB_DATA_TYPE_BINARY:
case TSDB_DATA_TYPE_NCHAR:
memcpy(pData, sVal.val, varDataTLen(sVal.val));
break;
case TSDB_DATA_TYPE_BOOL:
case TSDB_DATA_TYPE_TINYINT:
case TSDB_DATA_TYPE_UTINYINT:
*(uint8_t *)pData = *(uint8_t *)sVal.val;
break;
case TSDB_DATA_TYPE_SMALLINT:
case TSDB_DATA_TYPE_USMALLINT:
*(uint16_t *)pData = *(uint16_t *)sVal.val;
break;
case TSDB_DATA_TYPE_INT:
case TSDB_DATA_TYPE_UINT:
*(uint32_t *)pData = *(uint32_t *)sVal.val;
break;
case TSDB_DATA_TYPE_BIGINT:
case TSDB_DATA_TYPE_UBIGINT:
*(uint64_t *)pData = *(uint64_t *)sVal.val;
break;
case TSDB_DATA_TYPE_FLOAT:
SET_FLOAT_PTR(pData, sVal.val);
break;
case TSDB_DATA_TYPE_DOUBLE:
SET_DOUBLE_PTR(pData, sVal.val);
break;
case TSDB_DATA_TYPE_TIMESTAMP:
*(TSKEY*)pData = *(TSKEY*)sVal.val;
break;
default:
memcpy(pData, sVal.val, pColInfo->info.bytes);
}
colDataAppend(pColInfo, numOfRows, sVal.val, false);
} else if (forceSetNull) {
colDataAppend(pColInfo, numOfRows, NULL, true);
}
i++;
if(row == row1) {
......
source/libs/executor/inc/executorimpl.h
浏览文件 @ 3e455e37
......@@ -76,11 +76,12 @@ typedef struct SResultRowCell {
* If the number of generated results is greater than this value,
* query query will be halt and return results to client immediate.
*/
typedef struct SRspResultInfo {
int64_t total; // total generated result size in rows
int32_t capacity; // capacity of current result output buffer
int32_t threshold; // result size threshold in rows.
} SRspResultInfo;
typedef struct SResultInfo { // TODO refactor
int64_t totalRows; // total generated result size in rows
int64_t totalBytes; // total results in bytes.
int32_t capacity; // capacity of current result output buffer
int32_t threshold; // result size threshold in rows.
} SResultInfo;
typedef struct SColumnFilterElem {
int16_t bytes; // column length
......@@ -160,8 +161,8 @@ typedef struct STaskCostInfo {
typedef struct SOperatorCostInfo {
uint64_t openCost;
uint64_t execCost;
uint64_t totalRows;
uint64_t totalBytes;
// uint64_t totalRows;
// uint64_t totalBytes;
} SOperatorCostInfo;
typedef struct {
......@@ -301,7 +302,7 @@ typedef struct STaskRuntimeEnv {
int64_t currentOffset; // dynamic offset value
STableQueryInfo* current;
SRspResultInfo resultInfo;
SResultInfo resultInfo;
SHashObj* pTableRetrieveTsMap;
struct SUdfInfo* pUdfInfo;
} STaskRuntimeEnv;
......@@ -324,7 +325,7 @@ typedef struct SOperatorInfo {
STaskRuntimeEnv* pRuntimeEnv; // todo remove it
SExecTaskInfo* pTaskInfo;
SOperatorCostInfo cost;
SResultInfo resultInfo;
struct SOperatorInfo** pDownstream; // downstram pointer list
int32_t numOfDownstream; // number of downstream. The value is always ONE expect for join operator
__optr_fn_t getNextFn;
......@@ -539,6 +540,8 @@ typedef struct SFillOperatorInfo {
void** p;
SSDataBlock* existNewGroupBlock;
bool multigroupResult;
SInterval intervalInfo;
int32_t capacity;
} SFillOperatorInfo;
typedef struct SGroupKeys {
......@@ -649,21 +652,20 @@ SOperatorInfo* createOrderOperatorInfo(SOperatorInfo* downstream, SExprInfo* pEx
SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t numOfDownstream, SExprInfo* pExprInfo, int32_t num, SArray* pOrderVal, SArray* pGroupInfo, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createSysTableScanOperatorInfo(void* pSysTableReadHandle, SSDataBlock* pResBlock, const SName* pName,
SNode* pCondition, SEpSet epset, SArray* colList, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createLimitOperatorInfo(SOperatorInfo* downstream, int32_t numOfDownstream, SLimit* pLimit, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createLimitOperatorInfo(SOperatorInfo* downstream, SLimit* pLimit, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, SInterval* pInterval,
const STableGroupInfo* pTableGroupInfo, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock,
SArray* pGroupColList, SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols, SInterval* pInterval, SSDataBlock* pResBlock,
int32_t fillType, char* fillVal, bool multigroupResult, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createTableSeqScanOperatorInfo(void* pTsdbReadHandle, STaskRuntimeEnv* pRuntimeEnv);
SOperatorInfo* createAllTimeIntervalOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream,
SExprInfo* pExpr, int32_t numOfOutput);
SOperatorInfo* createFillOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr,
int32_t numOfOutput, bool multigroupResult);
SOperatorInfo* createMultiTableTimeIntervalOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream,
SExprInfo* pExpr, int32_t numOfOutput);
SOperatorInfo* createAllMultiTableTimeIntervalOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream,
......
source/libs/function/inc/tfill.h
浏览文件 @ 3e455e37
......@@ -61,7 +61,7 @@ typedef struct SFillInfo {
SFillColInfo* pFillCol; // column info for fill operations
SFillTagColInfo* pTags; // tags value for filling gap
void* handle; // for debug purpose
const char* id;
} SFillInfo;
int64_t getNumOfResultsAfterFillGap(SFillInfo* pFillInfo, int64_t ekey, int32_t maxNumOfRows);
......
source/libs/function/src/tfill.c
浏览文件 @ 3e455e37
......@@ -342,7 +342,7 @@ static int32_t taosNumOfRemainRows(SFillInfo* pFillInfo) {
struct SFillInfo* taosCreateFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_t capacity, int32_t numOfCols,
int64_t slidingTime, int8_t slidingUnit, int8_t precision, int32_t fillType,
struct SFillColInfo* pCol, void* handle) {
struct SFillColInfo* pCol, const char* id) {
if (fillType == TSDB_FILL_NONE) {
return NULL;
}
......@@ -357,7 +357,7 @@ struct SFillInfo* taosCreateFillInfo(int32_t order, TSKEY skey, int32_t numOfTag
pFillInfo->numOfCols = numOfCols;
pFillInfo->precision = precision;
pFillInfo->alloc = capacity;
pFillInfo->handle = handle;
pFillInfo->id = id;
pFillInfo->interval.interval = slidingTime;
pFillInfo->interval.intervalUnit = slidingUnit;
......
source/libs/nodes/src/nodesCloneFuncs.c
浏览文件 @ 3e455e37
......@@ -272,11 +272,14 @@ static SNode* logicWindowCopy(const SWindowLogicNode* pSrc, SWindowLogicNode* pD
COPY_SCALAR_FIELD(interval);
COPY_SCALAR_FIELD(offset);
COPY_SCALAR_FIELD(sliding);
COPY_SCALAR_FIELD(intervalUnit);
COPY_SCALAR_FIELD(slidingUnit);
CLONE_NODE_FIELD(pFill);
COPY_SCALAR_FIELD(sessionGap);
return (SNode*)pDst;
}
static SNode* logicSubplanCopy(const SSubLogicPlan* pSrc, SSubLogicPlan* pDst) {
static SNode* logicSubplanCopy(const SLogicSubplan* pSrc, SLogicSubplan* pDst) {
CLONE_NODE_FIELD(pNode);
COPY_SCALAR_FIELD(subplanType);
return (SNode*)pDst;
......@@ -358,7 +361,7 @@ SNodeptr nodesCloneNode(const SNodeptr pNode) {
case QUERY_NODE_LOGIC_PLAN_WINDOW:
return logicWindowCopy((const SWindowLogicNode*)pNode, (SWindowLogicNode*)pDst);
case QUERY_NODE_LOGIC_SUBPLAN:
return logicSubplanCopy((const SSubLogicPlan*)pNode, (SSubLogicPlan*)pDst);
return logicSubplanCopy((const SLogicSubplan*)pNode, (SLogicSubplan*)pDst);
default:
break;
}
......
source/libs/nodes/src/nodesCodeFuncs.c
浏览文件 @ 3e455e37
......@@ -70,6 +70,14 @@ const char* nodesNodeName(ENodeType type) {
return "SlotDesc";
case QUERY_NODE_COLUMN_DEF:
return "ColumnDef";
case QUERY_NODE_DOWNSTREAM_SOURCE:
return "DownstreamSource";
case QUERY_NODE_DATABASE_OPTIONS:
return "DatabaseOptions";
case QUERY_NODE_TABLE_OPTIONS:
return "TableOptions";
case QUERY_NODE_INDEX_OPTIONS:
return "IndexOptions";
case QUERY_NODE_SET_OPERATOR:
return "SetOperator";
case QUERY_NODE_SELECT_STMT:
......@@ -78,16 +86,76 @@ const char* nodesNodeName(ENodeType type) {
return "VnodeModifStmt";
case QUERY_NODE_CREATE_DATABASE_STMT:
return "CreateDatabaseStmt";
case QUERY_NODE_DROP_DATABASE_STMT:
return "DropDatabaseStmt";
case QUERY_NODE_ALTER_DATABASE_STMT:
return "AlterDatabaseStmt";
case QUERY_NODE_CREATE_TABLE_STMT:
return "CreateTableStmt";
case QUERY_NODE_CREATE_SUBTABLE_CLAUSE:
return "CreateSubtableClause";
case QUERY_NODE_CREATE_MULTI_TABLE_STMT:
return "CreateMultiTableStmt";
case QUERY_NODE_DROP_TABLE_CLAUSE:
return "DropTableClause";
case QUERY_NODE_DROP_TABLE_STMT:
return "DropTableStmt";
case QUERY_NODE_DROP_SUPER_TABLE_STMT:
return "DropSuperTableStmt";
case QUERY_NODE_ALTER_TABLE_STMT:
return "AlterTableStmt";
case QUERY_NODE_CREATE_USER_STMT:
return "CreateUserStmt";
case QUERY_NODE_ALTER_USER_STMT:
return "AlterUserStmt";
case QUERY_NODE_DROP_USER_STMT:
return "DropUserStmt";
case QUERY_NODE_USE_DATABASE_STMT:
return "UseDatabaseStmt";
case QUERY_NODE_CREATE_DNODE_STMT:
return "CreateDnodeStmt";
case QUERY_NODE_DROP_DNODE_STMT:
return "DropDnodeStmt";
case QUERY_NODE_ALTER_DNODE_STMT:
return "AlterDnodeStmt";
case QUERY_NODE_CREATE_INDEX_STMT:
return "CreateIndexStmt";
case QUERY_NODE_DROP_INDEX_STMT:
return "DropIndexStmt";
case QUERY_NODE_CREATE_QNODE_STMT:
return "CreateQnodeStmt";
case QUERY_NODE_DROP_QNODE_STMT:
return "DropQnodeStmt";
case QUERY_NODE_CREATE_TOPIC_STMT:
return "CreateTopicStmt";
case QUERY_NODE_DROP_TOPIC_STMT:
return "DropTopicStmt";
case QUERY_NODE_ALTER_LOCAL_STMT:
return "AlterLocalStmt";
case QUERY_NODE_SHOW_DATABASES_STMT:
return "ShowDatabaseStmt";
case QUERY_NODE_SHOW_TABLES_STMT:
return "ShowTablesStmt";
case QUERY_NODE_CREATE_TOPIC_STMT:
return "CreateTopicStmt";
case QUERY_NODE_SHOW_STABLES_STMT:
return "ShowStablesStmt";
case QUERY_NODE_SHOW_USERS_STMT:
return "ShowUsersStmt";
case QUERY_NODE_SHOW_DNODES_STMT:
return "ShowDnodesStmt";
case QUERY_NODE_SHOW_VGROUPS_STMT:
return "ShowVgroupsStmt";
case QUERY_NODE_SHOW_MNODES_STMT:
return "ShowMnodesStmt";
case QUERY_NODE_SHOW_MODULES_STMT:
return "ShowModulesStmt";
case QUERY_NODE_SHOW_QNODES_STMT:
return "ShowQnodesStmt";
case QUERY_NODE_SHOW_FUNCTIONS_STMT:
return "ShowFunctionsStmt";
case QUERY_NODE_SHOW_INDEXES_STMT:
return "ShowIndexesStmt";
case QUERY_NODE_SHOW_STREAMS_STMT:
return "ShowStreamsStmt";
case QUERY_NODE_LOGIC_PLAN_SCAN:
return "LogicScan";
case QUERY_NODE_LOGIC_PLAN_JOIN:
......@@ -98,6 +166,10 @@ const char* nodesNodeName(ENodeType type) {
return "LogicProject";
case QUERY_NODE_LOGIC_PLAN_VNODE_MODIF:
return "LogicVnodeModif";
case QUERY_NODE_LOGIC_PLAN_EXCHANGE:
return "LogicExchange";
case QUERY_NODE_LOGIC_PLAN_WINDOW:
return "LogicWindow";
case QUERY_NODE_LOGIC_SUBPLAN:
return "LogicSubplan";
case QUERY_NODE_LOGIC_PLAN:
......@@ -124,6 +196,8 @@ const char* nodesNodeName(ENodeType type) {
return "PhysiSort";
case QUERY_NODE_PHYSICAL_PLAN_INTERVAL:
return "PhysiInterval";
case QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW:
return "PhysiSessionWindow";
case QUERY_NODE_PHYSICAL_PLAN_DISPATCH:
return "PhysiDispatch";
case QUERY_NODE_PHYSICAL_PLAN_INSERT:
......@@ -846,8 +920,37 @@ static int32_t jsonToPhysiExchangeNode(const SJson* pJson, void* pObj) {
return code;
}
static const char* jkIntervalPhysiPlanExprs = "Exprs";
static const char* jkIntervalPhysiPlanFuncs = "Funcs";
static const char* jkWindowPhysiPlanExprs = "Exprs";
static const char* jkWindowPhysiPlanFuncs = "Funcs";
static int32_t physiWindowNodeToJson(const void* pObj, SJson* pJson) {
const SWinodwPhysiNode* pNode = (const SWinodwPhysiNode*)pObj;
int32_t code = physicPlanNodeToJson(pObj, pJson);
if (TSDB_CODE_SUCCESS == code) {
code = nodeListToJson(pJson, jkWindowPhysiPlanExprs, pNode->pExprs);
}
if (TSDB_CODE_SUCCESS == code) {
code = nodeListToJson(pJson, jkWindowPhysiPlanFuncs, pNode->pFuncs);
}
return code;
}
static int32_t jsonToPhysiWindowNode(const SJson* pJson, void* pObj) {
SWinodwPhysiNode* pNode = (SWinodwPhysiNode*)pObj;
int32_t code = jsonToPhysicPlanNode(pJson, pObj);
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeList(pJson, jkWindowPhysiPlanExprs, &pNode->pExprs);
}
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeList(pJson, jkWindowPhysiPlanFuncs, &pNode->pFuncs);
}
return code;
}
static const char* jkIntervalPhysiPlanInterval = "Interval";
static const char* jkIntervalPhysiPlanOffset = "Offset";
static const char* jkIntervalPhysiPlanSliding = "Sliding";
......@@ -858,13 +961,7 @@ static const char* jkIntervalPhysiPlanFill = "Fill";
static int32_t physiIntervalNodeToJson(const void* pObj, SJson* pJson) {
const SIntervalPhysiNode* pNode = (const SIntervalPhysiNode*)pObj;
int32_t code = physicPlanNodeToJson(pObj, pJson);
if (TSDB_CODE_SUCCESS == code) {
code = nodeListToJson(pJson, jkIntervalPhysiPlanExprs, pNode->pExprs);
}
if (TSDB_CODE_SUCCESS == code) {
code = nodeListToJson(pJson, jkIntervalPhysiPlanFuncs, pNode->pFuncs);
}
int32_t code = physiWindowNodeToJson(pObj, pJson);
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkIntervalPhysiPlanInterval, pNode->interval);
}
......@@ -890,13 +987,7 @@ static int32_t physiIntervalNodeToJson(const void* pObj, SJson* pJson) {
static int32_t jsonToPhysiIntervalNode(const SJson* pJson, void* pObj) {
SIntervalPhysiNode* pNode = (SIntervalPhysiNode*)pObj;
int32_t code = jsonToPhysicPlanNode(pJson, pObj);
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeList(pJson, jkIntervalPhysiPlanExprs, &pNode->pExprs);
}
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeList(pJson, jkIntervalPhysiPlanFuncs, &pNode->pFuncs);
}
int32_t code = jsonToPhysiWindowNode(pJson, pObj);
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBigIntValue(pJson, jkIntervalPhysiPlanInterval, &pNode->interval);
}
......@@ -919,6 +1010,30 @@ static int32_t jsonToPhysiIntervalNode(const SJson* pJson, void* pObj) {
return code;
}
static const char* jkSessionWindowPhysiPlanGap = "Gap";
static int32_t physiSessionWindowNodeToJson(const void* pObj, SJson* pJson) {
const SSessionWinodwPhysiNode* pNode = (const SSessionWinodwPhysiNode*)pObj;
int32_t code = physiWindowNodeToJson(pObj, pJson);
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkSessionWindowPhysiPlanGap, pNode->gap);
}
return code;
}
static int32_t jsonToPhysiSessionWindowNode(const SJson* pJson, void* pObj) {
SSessionWinodwPhysiNode* pNode = (SSessionWinodwPhysiNode*)pObj;
int32_t code = jsonToPhysiWindowNode(pJson, pObj);
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetNumberValue(pJson, jkSessionWindowPhysiPlanGap, pNode->gap);
}
return code;
}
static const char* jkDataSinkInputDataBlockDesc = "InputDataBlockDesc";
static int32_t physicDataSinkNodeToJson(const void* pObj, SJson* pJson) {
......@@ -2066,6 +2181,8 @@ static int32_t specificNodeToJson(const void* pObj, SJson* pJson) {
break;
case QUERY_NODE_PHYSICAL_PLAN_INTERVAL:
return physiIntervalNodeToJson(pObj, pJson);
case QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW:
return physiSessionWindowNodeToJson(pObj, pJson);
case QUERY_NODE_PHYSICAL_PLAN_DISPATCH:
return physiDispatchNodeToJson(pObj, pJson);
case QUERY_NODE_PHYSICAL_PLAN_INSERT:
......@@ -2075,7 +2192,7 @@ static int32_t specificNodeToJson(const void* pObj, SJson* pJson) {
case QUERY_NODE_PHYSICAL_PLAN:
return planToJson(pObj, pJson);
default:
assert(0);
// assert(0);
break;
}
nodesWarn("specificNodeToJson unknown node = %s", nodesNodeName(nodeType(pObj)));
......@@ -2149,14 +2266,16 @@ static int32_t jsonToSpecificNode(const SJson* pJson, void* pObj) {
return jsonToPhysiAggNode(pJson, pObj);
case QUERY_NODE_PHYSICAL_PLAN_EXCHANGE:
return jsonToPhysiExchangeNode(pJson, pObj);
case QUERY_NODE_PHYSICAL_PLAN_INTERVAL:
return jsonToPhysiIntervalNode(pJson, pObj);
case QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW:
return jsonToPhysiSessionWindowNode(pJson, pObj);
case QUERY_NODE_PHYSICAL_PLAN_DISPATCH:
return jsonToPhysiDispatchNode(pJson, pObj);
case QUERY_NODE_PHYSICAL_SUBPLAN:
return jsonToSubplan(pJson, pObj);
case QUERY_NODE_PHYSICAL_PLAN:
return jsonToPlan(pJson, pObj);
case QUERY_NODE_PHYSICAL_PLAN_INTERVAL:
return jsonToPhysiIntervalNode(pJson, pObj);
default:
assert(0);
break;
......
source/libs/nodes/src/nodesTraverseFuncs.c
浏览文件 @ 3e455e37
......@@ -79,9 +79,14 @@ static EDealRes walkNode(SNode* pNode, ETraversalOrder order, FNodeWalker walker
case QUERY_NODE_STATE_WINDOW:
res = walkNode(((SStateWindowNode*)pNode)->pCol, order, walker, pContext);
break;
case QUERY_NODE_SESSION_WINDOW:
res = walkNode(((SSessionWindowNode*)pNode)->pCol, order, walker, pContext);
case QUERY_NODE_SESSION_WINDOW: {
SSessionWindowNode* pSession = (SSessionWindowNode*)pNode;
res = walkNode(pSession->pCol, order, walker, pContext);
if (DEAL_RES_ERROR != res) {
res = walkNode(pSession->pGap, order, walker, pContext);
}
break;
}
case QUERY_NODE_INTERVAL_WINDOW: {
SIntervalWindowNode* pInterval = (SIntervalWindowNode*)pNode;
res = walkNode(pInterval->pInterval, order, walker, pContext);
......
source/libs/nodes/src/nodesUtilFuncs.c
浏览文件 @ 3e455e37
......@@ -160,7 +160,7 @@ SNodeptr nodesMakeNode(ENodeType type) {
case QUERY_NODE_LOGIC_PLAN_WINDOW:
return makeNode(type, sizeof(SWindowLogicNode));
case QUERY_NODE_LOGIC_SUBPLAN:
return makeNode(type, sizeof(SSubLogicPlan));
return makeNode(type, sizeof(SLogicSubplan));
case QUERY_NODE_LOGIC_PLAN:
return makeNode(type, sizeof(SQueryLogicPlan));
case QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN:
......@@ -185,6 +185,8 @@ SNodeptr nodesMakeNode(ENodeType type) {
return makeNode(type, sizeof(SNode));
case QUERY_NODE_PHYSICAL_PLAN_INTERVAL:
return makeNode(type, sizeof(SIntervalPhysiNode));
case QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW:
return makeNode(type, sizeof(SSessionWinodwPhysiNode));
case QUERY_NODE_PHYSICAL_PLAN_DISPATCH:
return makeNode(type, sizeof(SDataDispatcherNode));
case QUERY_NODE_PHYSICAL_PLAN_INSERT:
......@@ -332,6 +334,7 @@ int32_t nodesListAppend(SNodeList* pList, SNodeptr pNode) {
int32_t nodesListStrictAppend(SNodeList* pList, SNodeptr pNode) {
if (NULL == pNode) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return TSDB_CODE_OUT_OF_MEMORY;
}
int32_t code = nodesListAppend(pList, pNode);
......@@ -341,6 +344,17 @@ int32_t nodesListStrictAppend(SNodeList* pList, SNodeptr pNode) {
return code;
}
int32_t nodesListMakeAppend(SNodeList** pList, SNodeptr pNode) {
if (NULL == *pList) {
*pList = nodesMakeList();
if (NULL == *pList) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return TSDB_CODE_OUT_OF_MEMORY;
}
}
return nodesListAppend(*pList, pNode);
}
int32_t nodesListAppendList(SNodeList* pTarget, SNodeList* pSrc) {
if (NULL == pTarget || NULL == pSrc) {
return TSDB_CODE_SUCCESS;
......
source/libs/parser/inc/parAst.h
浏览文件 @ 3e455e37
......@@ -100,7 +100,7 @@ SNode* createTempTableNode(SAstCreateContext* pCxt, SNode* pSubquery, const STok
SNode* createJoinTableNode(SAstCreateContext* pCxt, EJoinType type, SNode* pLeft, SNode* pRight, SNode* pJoinCond);
SNode* createLimitNode(SAstCreateContext* pCxt, const SToken* pLimit, const SToken* pOffset);
SNode* createOrderByExprNode(SAstCreateContext* pCxt, SNode* pExpr, EOrder order, ENullOrder nullOrder);
SNode* createSessionWindowNode(SAstCreateContext* pCxt, SNode* pCol, const SToken* pVal);
SNode* createSessionWindowNode(SAstCreateContext* pCxt, SNode* pCol, SNode* pGap);
SNode* createStateWindowNode(SAstCreateContext* pCxt, SNode* pCol);
SNode* createIntervalWindowNode(SAstCreateContext* pCxt, SNode* pInterval, SNode* pOffset, SNode* pSliding, SNode* pFill);
SNode* createFillNode(SAstCreateContext* pCxt, EFillMode mode, SNode* pValues);
......
source/libs/parser/inc/sql.y
浏览文件 @ 3e455e37
......@@ -624,7 +624,7 @@ partition_by_clause_opt(A) ::= PARTITION BY expression_list(B).
twindow_clause_opt(A) ::= . { A = NULL; }
twindow_clause_opt(A) ::=
SESSION NK_LP column_reference(B) NK_COMMA NK_INTEGER(C) NK_RP. { A = createSessionWindowNode(pCxt, releaseRawExprNode(pCxt, B), &C); }
SESSION NK_LP column_reference(B) NK_COMMA duration_literal(C) NK_RP. { A = createSessionWindowNode(pCxt, releaseRawExprNode(pCxt, B), releaseRawExprNode(pCxt, C)); }
twindow_clause_opt(A) ::= STATE_WINDOW NK_LP column_reference(B) NK_RP. { A = createStateWindowNode(pCxt, releaseRawExprNode(pCxt, B)); }
twindow_clause_opt(A) ::=
INTERVAL NK_LP duration_literal(B) NK_RP sliding_opt(C) fill_opt(D). { A = createIntervalWindowNode(pCxt, releaseRawExprNode(pCxt, B), NULL, C, D); }
......
source/libs/parser/src/parAstCreater.c
浏览文件 @ 3e455e37
......@@ -679,11 +679,11 @@ SNode* createOrderByExprNode(SAstCreateContext* pCxt, SNode* pExpr, EOrder order
return (SNode*)orderByExpr;
}
SNode* createSessionWindowNode(SAstCreateContext* pCxt, SNode* pCol, const SToken* pVal) {
SNode* createSessionWindowNode(SAstCreateContext* pCxt, SNode* pCol, SNode* pGap) {
SSessionWindowNode* session = (SSessionWindowNode*)nodesMakeNode(QUERY_NODE_SESSION_WINDOW);
CHECK_OUT_OF_MEM(session);
session->pCol = pCol;
// session->gap = getInteger(pVal);
session->pGap = pGap;
return (SNode*)session;
}
......
source/libs/parser/src/parTranslater.c
浏览文件 @ 3e455e37
......@@ -1870,14 +1870,21 @@ static void toSchema(const SColumnDefNode* pCol, int32_t colId, SSchema* pSchema
}
static void destroyCreateTbReq(SVCreateTbReq* pReq) {
tfree(pReq->dbFName);
tfree(pReq->name);
tfree(pReq->ntbCfg.pSchema);
}
static int32_t buildNormalTableBatchReq(
const char* pDbName, const char* pTableName, const SNodeList* pColumns, const SVgroupInfo* pVgroupInfo, SVgroupTablesBatch* pBatch) {
static int32_t buildNormalTableBatchReq(int32_t acctId, const char* pDbName, const char* pTableName,
const SNodeList* pColumns, const SVgroupInfo* pVgroupInfo, SVgroupTablesBatch* pBatch) {
char dbFName[TSDB_DB_FNAME_LEN] = {0};
SName name = { .type = TSDB_DB_NAME_T, .acctId = acctId };
strcpy(name.dbname, pDbName);
tNameGetFullDbName(&name, dbFName);
SVCreateTbReq req = {0};
req.type = TD_NORMAL_TABLE;
req.dbFName = strdup(dbFName);
req.name = strdup(pTableName);
req.ntbCfg.nCols = LIST_LENGTH(pColumns);
req.ntbCfg.pSchema = calloc(req.ntbCfg.nCols, sizeof(SSchema));
......@@ -1904,7 +1911,7 @@ static int32_t buildNormalTableBatchReq(
return TSDB_CODE_SUCCESS;
}
static int32_t serializeVgroupTablesBatch(int32_t acctId, SVgroupTablesBatch* pTbBatch, SArray* pBufArray) {
static int32_t serializeVgroupTablesBatch(SVgroupTablesBatch* pTbBatch, SArray* pBufArray) {
int tlen = sizeof(SMsgHead) + tSerializeSVCreateTbBatchReq(NULL, &(pTbBatch->req));
void* buf = malloc(tlen);
if (NULL == buf) {
......@@ -1932,6 +1939,7 @@ static void destroyCreateTbReqBatch(SVgroupTablesBatch* pTbBatch) {
size_t size = taosArrayGetSize(pTbBatch->req.pArray);
for(int32_t i = 0; i < size; ++i) {
SVCreateTbReq* pTableReq = taosArrayGet(pTbBatch->req.pArray, i);
tfree(pTableReq->dbFName);
tfree(pTableReq->name);
if (pTableReq->type == TSDB_NORMAL_TABLE) {
......@@ -1973,9 +1981,9 @@ static int32_t buildCreateTableDataBlock(int32_t acctId, const SCreateTableStmt*
}
SVgroupTablesBatch tbatch = {0};
int32_t code = buildNormalTableBatchReq(pStmt->dbName, pStmt->tableName, pStmt->pCols, pInfo, &tbatch);
int32_t code = buildNormalTableBatchReq(acctId, pStmt->dbName, pStmt->tableName, pStmt->pCols, pInfo, &tbatch);
if (TSDB_CODE_SUCCESS == code) {
code = serializeVgroupTablesBatch(acctId, &tbatch, *pBufArray);
code = serializeVgroupTablesBatch(&tbatch, *pBufArray);
}
destroyCreateTbReqBatch(&tbatch);
......@@ -2004,9 +2012,16 @@ static int32_t rewriteCreateTable(STranslateContext* pCxt, SQuery* pQuery) {
return code;
}
static void addCreateTbReqIntoVgroup(SHashObj* pVgroupHashmap, const char* pDbName, const char* pTableName, SKVRow row, uint64_t suid, SVgroupInfo* pVgInfo) {
static void addCreateTbReqIntoVgroup(int32_t acctId, SHashObj* pVgroupHashmap,
const char* pDbName, const char* pTableName, SKVRow row, uint64_t suid, SVgroupInfo* pVgInfo) {
char dbFName[TSDB_DB_FNAME_LEN] = {0};
SName name = { .type = TSDB_DB_NAME_T, .acctId = acctId };
strcpy(name.dbname, pDbName);
tNameGetFullDbName(&name, dbFName);
struct SVCreateTbReq req = {0};
req.type = TD_CHILD_TABLE;
req.dbFName = strdup(dbFName);
req.name = strdup(pTableName);
req.ctbCfg.suid = suid;
req.ctbCfg.pTag = row;
......@@ -2159,7 +2174,7 @@ static int32_t rewriteCreateSubTable(STranslateContext* pCxt, SCreateSubTableCla
code = getTableHashVgroup(pCxt, pStmt->dbName, pStmt->tableName, &info);
}
if (TSDB_CODE_SUCCESS == code) {
addCreateTbReqIntoVgroup(pVgroupHashmap, pStmt->dbName, pStmt->tableName, row, pSuperTableMeta->uid, &info);
addCreateTbReqIntoVgroup(pCxt->pParseCxt->acctId, pVgroupHashmap, pStmt->dbName, pStmt->tableName, row, pSuperTableMeta->uid, &info);
}
tfree(pSuperTableMeta);
......@@ -2181,7 +2196,7 @@ static SArray* serializeVgroupsTablesBatch(int32_t acctId, SHashObj* pVgroupHash
break;
}
serializeVgroupTablesBatch(acctId, pTbBatch, pBufArray);
serializeVgroupTablesBatch(pTbBatch, pBufArray);
destroyCreateTbReqBatch(pTbBatch);
} while (true);
......
source/libs/parser/src/parUtil.c
浏览文件 @ 3e455e37
......@@ -71,6 +71,7 @@ int32_t generateSyntaxErrMsg(SMsgBuf* pBuf, int32_t errCode, ...) {
va_start(vArgList, errCode);
vsnprintf(pBuf->buf, pBuf->len, getSyntaxErrFormat(errCode), vArgList);
va_end(vArgList);
terrno = errCode;
return errCode;
}
......
source/libs/parser/src/sql.c
浏览文件 @ 3e455e37
此差异已折叠。
source/libs/parser/test/parserAstTest.cpp
浏览文件 @ 3e455e37
......@@ -44,6 +44,9 @@ protected:
query_ = nullptr;
bool res = runImpl(parseCode, translateCode);
qDestroyQuery(query_);
if (!res) {
dump();
}
return res;
}
......@@ -53,26 +56,40 @@ private:
bool runImpl(int32_t parseCode, int32_t translateCode) {
int32_t code = doParse(&cxt_, &query_);
if (code != TSDB_CODE_SUCCESS) {
cout << "sql:[" << cxt_.pSql << "] parser code:" << tstrerror(code) << ", msg:" << errMagBuf_ << endl;
return (TSDB_CODE_SUCCESS != parseCode);
parseErrStr_ = string("code:") + tstrerror(code) + string(", msg:") + errMagBuf_;
return (terrno == parseCode);
}
if (TSDB_CODE_SUCCESS != parseCode) {
return false;
}
string parserStr = toString(query_->pRoot);
parsedAstStr_ = toString(query_->pRoot);
code = doTranslate(&cxt_, query_);
if (code != TSDB_CODE_SUCCESS) {
cout << "sql:[" << cxt_.pSql << "] translate code:" << code << ", " << translateCode << ", msg:" << errMagBuf_ << endl;
return (code == translateCode);
translateErrStr_ = string("code:") + tstrerror(code) + string(", msg:") + errMagBuf_;
return (terrno == translateCode);
}
cout << "input sql : [" << cxt_.pSql << "]" << endl;
cout << "parser output: " << endl;
cout << parserStr << endl;
cout << "translate output: " << endl;
cout << toString(query_->pRoot) << endl;
translatedAstStr_ = toString(query_->pRoot);
return (TSDB_CODE_SUCCESS == translateCode);
}
void dump() {
cout << "input sql : [" << cxt_.pSql << "]" << endl;
if (!parseErrStr_.empty()) {
cout << "parse error: " << parseErrStr_ << endl;
}
if (!parsedAstStr_.empty()) {
cout << "parse output: " << endl;
cout << parsedAstStr_ << endl;
}
if (!translateErrStr_.empty()) {
cout << "translate error: " << translateErrStr_ << endl;
}
if (!translatedAstStr_.empty()) {
cout << "translate output: " << endl;
cout << translatedAstStr_ << endl;
}
}
string toString(const SNode* pRoot, bool format = false) {
char* pStr = NULL;
int32_t len = 0;
......@@ -91,6 +108,10 @@ private:
memset(errMagBuf_, 0, max_err_len);
cxt_.pMsg = errMagBuf_;
cxt_.msgLen = max_err_len;
parseErrStr_.clear();
parsedAstStr_.clear();
translateErrStr_.clear();
translatedAstStr_.clear();
}
string acctId_;
......@@ -99,8 +120,50 @@ private:
string sqlBuf_;
SParseContext cxt_;
SQuery* query_;
string parseErrStr_;
string parsedAstStr_;
string translateErrStr_;
string translatedAstStr_;
};
TEST_F(ParserTest, createAccount) {
setDatabase("root", "test");
bind("create account ac_wxy pass '123456'");
ASSERT_TRUE(run(TSDB_CODE_PAR_EXPRIE_STATEMENT));
}
TEST_F(ParserTest, alterAccount) {
setDatabase("root", "test");
bind("alter account ac_wxy pass '123456'");
ASSERT_TRUE(run(TSDB_CODE_PAR_EXPRIE_STATEMENT));
}
TEST_F(ParserTest, createUser) {
setDatabase("root", "test");
bind("create user wxy pass '123456'");
ASSERT_TRUE(run());
}
TEST_F(ParserTest, alterUser) {
setDatabase("root", "test");
bind("alter user wxy pass '123456'");
ASSERT_TRUE(run());
bind("alter user wxy privilege 'write'");
ASSERT_TRUE(run());
}
TEST_F(ParserTest, dropUser) {
setDatabase("root", "test");
bind("drop user wxy");
ASSERT_TRUE(run());
}
TEST_F(ParserTest, selectSimple) {
setDatabase("root", "test");
......@@ -295,12 +358,6 @@ TEST_F(ParserTest, selectSemanticError) {
ASSERT_TRUE(run(TSDB_CODE_SUCCESS, TSDB_CODE_PAR_NOT_SELECTED_EXPRESSION));
}
TEST_F(ParserTest, createUser) {
setDatabase("root", "test");
bind("create user wxy pass '123456'");
ASSERT_TRUE(run());
}
TEST_F(ParserTest, showUsers) {
setDatabase("root", "test");
......@@ -309,12 +366,7 @@ TEST_F(ParserTest, showUsers) {
ASSERT_TRUE(run());
}
TEST_F(ParserTest, alterAccount) {
setDatabase("root", "test");
bind("alter account ac_wxy pass '123456'");
ASSERT_TRUE(run(TSDB_CODE_PAR_EXPRIE_STATEMENT));
}
TEST_F(ParserTest, createDnode) {
setDatabase("root", "test");
......
source/libs/planner/inc/planInt.h
浏览文件 @ 3e455e37
......@@ -56,9 +56,10 @@ extern "C" {
#define planTrace(param, ...) qTrace("PLAN: " param, __VA_ARGS__)
int32_t createLogicPlan(SPlanContext* pCxt, SLogicNode** pLogicNode);
int32_t optimize(SPlanContext* pCxt, SLogicNode* pLogicNode);
int32_t applySplitRule(SSubLogicPlan* pSubplan);
int32_t createPhysiPlan(SPlanContext* pCxt, SLogicNode* pLogicNode, SQueryPlan** pPlan, SArray* pExecNodeList);
int32_t optimizeLogicPlan(SPlanContext* pCxt, SLogicNode* pLogicNode);
int32_t splitLogicPlan(SPlanContext* pCxt, SLogicNode* pLogicNode, SLogicSubplan** pLogicSubplan);
int32_t scaleOutLogicPlan(SPlanContext* pCxt, SLogicSubplan* pLogicSubplan, SQueryLogicPlan** pLogicPlan);
int32_t createPhysiPlan(SPlanContext* pCxt, SQueryLogicPlan* pLogicPlan, SQueryPlan** pPlan, SArray* pExecNodeList);
#ifdef __cplusplus
}
......
source/libs/planner/src/planLogicCreater.c
浏览文件 @ 3e455e37
......@@ -411,6 +411,38 @@ static int32_t createAggLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect,
return code;
}
static int32_t createWindowLogicNodeFinalize(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SWindowLogicNode* pWindow, SLogicNode** pLogicNode) {
int32_t code = nodesCollectFuncs(pSelect, fmIsAggFunc, &pWindow->pFuncs);
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExpr(pWindow->pFuncs, pSelect, SQL_CLAUSE_WINDOW);
}
if (TSDB_CODE_SUCCESS == code) {
code = createColumnByRewriteExps(pCxt, pWindow->pFuncs, &pWindow->node.pTargets);
}
if (TSDB_CODE_SUCCESS == code) {
*pLogicNode = (SLogicNode*)pWindow;
} else {
nodesDestroyNode(pWindow);
}
return code;
}
static int32_t createWindowLogicNodeBySession(SLogicPlanContext* pCxt, SSessionWindowNode* pSession, SSelectStmt* pSelect, SLogicNode** pLogicNode) {
SWindowLogicNode* pWindow = nodesMakeNode(QUERY_NODE_LOGIC_PLAN_WINDOW);
if (NULL == pWindow) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pWindow->winType = WINDOW_TYPE_SESSION;
pWindow->sessionGap = ((SValueNode*)pSession->pGap)->datum.i;
return createWindowLogicNodeFinalize(pCxt, pSelect, pWindow, pLogicNode);
}
static int32_t createWindowLogicNodeByInterval(SLogicPlanContext* pCxt, SIntervalWindowNode* pInterval, SSelectStmt* pSelect, SLogicNode** pLogicNode) {
SWindowLogicNode* pWindow = nodesMakeNode(QUERY_NODE_LOGIC_PLAN_WINDOW);
if (NULL == pWindow) {
......@@ -424,34 +456,15 @@ static int32_t createWindowLogicNodeByInterval(SLogicPlanContext* pCxt, SInterva
pWindow->sliding = (NULL != pInterval->pSliding ? ((SValueNode*)pInterval->pSliding)->datum.i : pWindow->interval);
pWindow->slidingUnit = (NULL != pInterval->pSliding ? ((SValueNode*)pInterval->pSliding)->unit : pWindow->intervalUnit);
int32_t code = TSDB_CODE_SUCCESS;
if (NULL != pInterval->pFill) {
pWindow->pFill = nodesCloneNode(pInterval->pFill);
if (NULL == pWindow->pFill) {
code = TSDB_CODE_OUT_OF_MEMORY;
nodesDestroyNode(pWindow);
return TSDB_CODE_OUT_OF_MEMORY;
}
}
if (TSDB_CODE_SUCCESS == code) {
code = nodesCollectFuncs(pSelect, fmIsAggFunc, &pWindow->pFuncs);
}
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExpr(pWindow->pFuncs, pSelect, SQL_CLAUSE_WINDOW);
}
if (TSDB_CODE_SUCCESS == code) {
code = createColumnByRewriteExps(pCxt, pWindow->pFuncs, &pWindow->node.pTargets);
}
if (TSDB_CODE_SUCCESS == code) {
*pLogicNode = (SLogicNode*)pWindow;
} else {
nodesDestroyNode(pWindow);
}
return code;
return createWindowLogicNodeFinalize(pCxt, pSelect, pWindow, pLogicNode);
}
static int32_t createWindowLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SLogicNode** pLogicNode) {
......@@ -460,8 +473,10 @@ static int32_t createWindowLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSele
}
switch (nodeType(pSelect->pWindow)) {
case QUERY_NODE_SESSION_WINDOW:
return createWindowLogicNodeBySession(pCxt, (SSessionWindowNode*)pSelect->pWindow, pSelect, pLogicNode);
case QUERY_NODE_INTERVAL_WINDOW:
return createWindowLogicNodeByInterval(pCxt, (SIntervalWindowNode*)pSelect->pWindow, pSelect, pLogicNode);
return createWindowLogicNodeByInterval(pCxt, (SIntervalWindowNode*)pSelect->pWindow, pSelect, pLogicNode);
default:
break;
}
......
source/libs/planner/src/planOptimizer.c 0 → 100644
浏览文件 @ 3e455e37
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "planInt.h"
int32_t optimizeLogicPlan(SPlanContext* pCxt, SLogicNode* pLogicNode) {
return TSDB_CODE_SUCCESS;
}
source/libs/planner/src/planScaleOut.c 0 → 100644
浏览文件 @ 3e455e37
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "planInt.h"
typedef struct SScaleOutContext {
SPlanContext* pPlanCxt;
int32_t subplanId;
} SScaleOutContext;
static SLogicSubplan* singleCloneSubLogicPlan(SScaleOutContext* pCxt, SLogicSubplan* pSrc, int32_t level) {
SLogicSubplan* pDst = nodesMakeNode(QUERY_NODE_LOGIC_SUBPLAN);
if (NULL == pDst) {
return NULL;
}
pDst->pNode = nodesCloneNode(pSrc->pNode);
if (NULL == pDst->pNode) {
nodesDestroyNode(pDst);
return NULL;
}
pDst->subplanType = pSrc->subplanType;
pDst->level = level;
pDst->id.queryId = pSrc->id.queryId;
pDst->id.groupId = pSrc->id.groupId;
pDst->id.subplanId = pCxt->subplanId++;
return pDst;
}
static int32_t scaleOutForModify(SScaleOutContext* pCxt, SLogicSubplan* pSubplan, int32_t level, SNodeList* pGroup) {
SVnodeModifLogicNode* pNode = (SVnodeModifLogicNode*)pSubplan->pNode;
size_t numOfVgroups = taosArrayGetSize(pNode->pDataBlocks);
for (int32_t i = 0; i < numOfVgroups; ++i) {
SLogicSubplan* pNewSubplan = singleCloneSubLogicPlan(pCxt, pSubplan, level);
if (NULL == pNewSubplan) {
return TSDB_CODE_OUT_OF_MEMORY;
}
((SVnodeModifLogicNode*)pNewSubplan->pNode)->pVgDataBlocks = (SVgDataBlocks*)taosArrayGetP(pNode->pDataBlocks, i);
if (TSDB_CODE_SUCCESS != nodesListStrictAppend(pGroup, pNewSubplan)) {
return TSDB_CODE_OUT_OF_MEMORY;
}
}
return TSDB_CODE_SUCCESS;
}
static int32_t scaleOutForMerge(SScaleOutContext* pCxt, SLogicSubplan* pSubplan, int32_t level, SNodeList* pGroup) {
return nodesListStrictAppend(pGroup, singleCloneSubLogicPlan(pCxt, pSubplan, level));
}
static int32_t doSetScanVgroup(SLogicNode* pNode, const SVgroupInfo* pVgroup, bool* pFound) {
if (QUERY_NODE_LOGIC_PLAN_SCAN == nodeType(pNode)) {
SScanLogicNode* pScan = (SScanLogicNode*)pNode;
pScan->pVgroupList = calloc(1, sizeof(SVgroupsInfo) + sizeof(SVgroupInfo));
if (NULL == pScan->pVgroupList) {
return TSDB_CODE_OUT_OF_MEMORY;
}
memcpy(pScan->pVgroupList->vgroups, pVgroup, sizeof(SVgroupInfo));
*pFound = true;
return TSDB_CODE_SUCCESS;
}
SNode* pChild = NULL;
FOREACH(pChild, pNode->pChildren) {
int32_t code = doSetScanVgroup((SLogicNode*)pChild, pVgroup, pFound);
if (TSDB_CODE_SUCCESS != code || *pFound) {
return code;
}
}
return TSDB_CODE_SUCCESS;
}
static int32_t setScanVgroup(SLogicNode* pNode, const SVgroupInfo* pVgroup) {
bool found = false;
return doSetScanVgroup(pNode, pVgroup, &found);
}
static int32_t scaleOutForScan(SScaleOutContext* pCxt, SLogicSubplan* pSubplan, int32_t level, SNodeList* pGroup) {
if (pSubplan->pVgroupList) {
int32_t code = TSDB_CODE_SUCCESS;
for (int32_t i = 0; i < pSubplan->pVgroupList->numOfVgroups; ++i) {
SLogicSubplan* pNewSubplan = singleCloneSubLogicPlan(pCxt, pSubplan, level);
if (NULL == pNewSubplan) {
return TSDB_CODE_OUT_OF_MEMORY;
}
code = setScanVgroup(pNewSubplan->pNode, pSubplan->pVgroupList->vgroups + i);
if (TSDB_CODE_SUCCESS == code) {
code = nodesListStrictAppend(pGroup, pNewSubplan);
}
if (TSDB_CODE_SUCCESS != code) {
break;
}
}
return code;
} else {
return scaleOutForMerge(pCxt, pSubplan, level, pGroup);
}
}
static int32_t pushHierarchicalPlan(SNodeList* pParentsGroup, SNodeList* pCurrentGroup) {
int32_t code = TSDB_CODE_SUCCESS;
bool topLevel = (0 == LIST_LENGTH(pParentsGroup));
SNode* pChild = NULL;
FOREACH(pChild, pCurrentGroup) {
if (topLevel) {
code = nodesListAppend(pParentsGroup, pChild);
} else {
SNode* pParent = NULL;
FOREACH(pParent, pParentsGroup) {
code = nodesListMakeAppend(&(((SLogicSubplan*)pParent)->pChildren), pChild);
if (TSDB_CODE_SUCCESS == code) {
code = nodesListMakeAppend(&(((SLogicSubplan*)pChild)->pParents), pParent);
}
}
}
if (TSDB_CODE_SUCCESS != code) {
break;
}
}
return code;
}
static int32_t doScaleOut(SScaleOutContext* pCxt, SLogicSubplan* pSubplan, int32_t* pLevel, SNodeList* pParentsGroup) {
SNodeList* pCurrentGroup = nodesMakeList();
if (NULL == pCurrentGroup) {
return TSDB_CODE_OUT_OF_MEMORY;
}
int32_t code = TSDB_CODE_SUCCESS;
switch (pSubplan->subplanType) {
case SUBPLAN_TYPE_MERGE:
code = scaleOutForMerge(pCxt, pSubplan, *pLevel, pCurrentGroup);
break;
case SUBPLAN_TYPE_SCAN:
code = scaleOutForScan(pCxt, pSubplan, *pLevel, pCurrentGroup);
break;
case SUBPLAN_TYPE_MODIFY:
code = scaleOutForModify(pCxt, pSubplan, *pLevel, pCurrentGroup);
break;
default:
break;
}
if (TSDB_CODE_SUCCESS == code) {
code = pushHierarchicalPlan(pParentsGroup, pCurrentGroup);
++(*pLevel);
}
if (TSDB_CODE_SUCCESS == code) {
SNode* pChild;
FOREACH(pChild, pSubplan->pChildren) {
code = doScaleOut(pCxt, (SLogicSubplan*)pChild, pLevel, pCurrentGroup);
if (TSDB_CODE_SUCCESS != code) {
break;
}
}
}
if (TSDB_CODE_SUCCESS != code) {
nodesDestroyList(pCurrentGroup);
}
return code;
}
static SQueryLogicPlan* makeQueryLogicPlan() {
SQueryLogicPlan* pLogicPlan = (SQueryLogicPlan*)nodesMakeNode(QUERY_NODE_LOGIC_PLAN);
if (NULL == pLogicPlan) {
return NULL;
}
pLogicPlan->pTopSubplans = nodesMakeList();
if (NULL == pLogicPlan->pTopSubplans) {
nodesDestroyNode(pLogicPlan);
return NULL;
}
return pLogicPlan;
}
int32_t scaleOutLogicPlan(SPlanContext* pCxt, SLogicSubplan* pLogicSubplan, SQueryLogicPlan** pLogicPlan) {
SQueryLogicPlan* pPlan = makeQueryLogicPlan();
if (NULL == pPlan) {
return TSDB_CODE_OUT_OF_MEMORY;
}
SScaleOutContext cxt = { .pPlanCxt = pCxt, .subplanId = 1 };
int32_t code = doScaleOut(&cxt, pLogicSubplan, &(pPlan->totalLevel), pPlan->pTopSubplans);
if (TSDB_CODE_SUCCESS == code) {
*pLogicPlan = pPlan;
} else {
nodesDestroyNode(pPlan);
}
return code;
}
source/libs/planner/src/planSpliter.c
浏览文件 @ 3e455e37
......@@ -29,7 +29,7 @@ typedef struct SSplitContext {
void* pInfo;
} SSplitContext;
typedef int32_t (*FMatch)(SSplitContext* pCxt, SSubLogicPlan* pSubplan);
typedef int32_t (*FMatch)(SSplitContext* pCxt, SLogicSubplan* pSubplan);
typedef int32_t (*FSplit)(SSplitContext* pCxt);
typedef struct SSplitRule {
......@@ -40,7 +40,7 @@ typedef struct SSplitRule {
typedef struct SStsInfo {
SScanLogicNode* pScan;
SSubLogicPlan* pSubplan;
SLogicSubplan* pSubplan;
} SStsInfo;
static SLogicNode* stsMatchByNode(SLogicNode* pNode) {
......@@ -58,7 +58,7 @@ static SLogicNode* stsMatchByNode(SLogicNode* pNode) {
return NULL;
}
static int32_t stsMatch(SSplitContext* pCxt, SSubLogicPlan* pSubplan) {
static int32_t stsMatch(SSplitContext* pCxt, SLogicSubplan* pSubplan) {
if (SPLIT_FLAG_TEST_MASK(pSubplan->splitFlag, SPLIT_FLAG_STS)) {
return TSDB_CODE_SUCCESS;
}
......@@ -74,7 +74,7 @@ static int32_t stsMatch(SSplitContext* pCxt, SSubLogicPlan* pSubplan) {
}
SNode* pChild;
FOREACH(pChild, pSubplan->pChildren) {
int32_t code = stsMatch(pCxt, (SSubLogicPlan*)pChild);
int32_t code = stsMatch(pCxt, (SLogicSubplan*)pChild);
if (TSDB_CODE_SUCCESS != code || pCxt->match) {
return code;
}
......@@ -82,8 +82,8 @@ static int32_t stsMatch(SSplitContext* pCxt, SSubLogicPlan* pSubplan) {
return TSDB_CODE_SUCCESS;
}
static SSubLogicPlan* stsCreateScanSubplan(SSplitContext* pCxt, SScanLogicNode* pScan) {
SSubLogicPlan* pSubplan = nodesMakeNode(QUERY_NODE_LOGIC_SUBPLAN);
static SLogicSubplan* stsCreateScanSubplan(SSplitContext* pCxt, SScanLogicNode* pScan) {
SLogicSubplan* pSubplan = nodesMakeNode(QUERY_NODE_LOGIC_SUBPLAN);
if (NULL == pSubplan) {
return NULL;
}
......@@ -95,7 +95,7 @@ static SSubLogicPlan* stsCreateScanSubplan(SSplitContext* pCxt, SScanLogicNode*
return pSubplan;
}
static int32_t stsCreateExchangeNode(SSplitContext* pCxt, SSubLogicPlan* pSubplan, SScanLogicNode* pScan) {
static int32_t stsCreateExchangeNode(SSplitContext* pCxt, SLogicSubplan* pSubplan, SScanLogicNode* pScan) {
SExchangeLogicNode* pExchange = nodesMakeNode(QUERY_NODE_LOGIC_PLAN_EXCHANGE);
if (NULL == pExchange) {
return TSDB_CODE_OUT_OF_MEMORY;
......@@ -145,7 +145,7 @@ static const SSplitRule splitRuleSet[] = {
static const int32_t splitRuleNum = (sizeof(splitRuleSet) / sizeof(SSplitRule));
int32_t applySplitRule(SSubLogicPlan* pSubplan) {
static int32_t applySplitRule(SLogicSubplan* pSubplan) {
SSplitContext cxt = { .errCode = TSDB_CODE_SUCCESS, .groupId = pSubplan->id.groupId + 1, .match = false, .pInfo = NULL };
bool split = false;
do {
......@@ -164,3 +164,45 @@ int32_t applySplitRule(SSubLogicPlan* pSubplan) {
} while (split);
return TSDB_CODE_SUCCESS;
}
static void doSetLogicNodeParent(SLogicNode* pNode, SLogicNode* pParent) {
pNode->pParent = pParent;
SNode* pChild;
FOREACH(pChild, pNode->pChildren) {
doSetLogicNodeParent((SLogicNode*)pChild, pNode);
}
}
static void setLogicNodeParent(SLogicNode* pNode) {
doSetLogicNodeParent(pNode, NULL);
}
int32_t splitLogicPlan(SPlanContext* pCxt, SLogicNode* pLogicNode, SLogicSubplan** pLogicSubplan) {
SLogicSubplan* pSubplan = (SLogicSubplan*)nodesMakeNode(QUERY_NODE_LOGIC_SUBPLAN);
if (NULL == pSubplan) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pSubplan->pNode = nodesCloneNode(pLogicNode);
if (NULL == pSubplan->pNode) {
nodesDestroyNode(pSubplan);
return TSDB_CODE_OUT_OF_MEMORY;
}
if (QUERY_NODE_LOGIC_PLAN_VNODE_MODIF == nodeType(pLogicNode)) {
pSubplan->subplanType = SUBPLAN_TYPE_MODIFY;
TSWAP(((SVnodeModifLogicNode*)pLogicNode)->pDataBlocks, ((SVnodeModifLogicNode*)pSubplan->pNode)->pDataBlocks, SArray*);
} else {
pSubplan->subplanType = SUBPLAN_TYPE_SCAN;
}
pSubplan->id.queryId = pCxt->queryId;
setLogicNodeParent(pSubplan->pNode);
int32_t code = applySplitRule(pSubplan);
if (TSDB_CODE_SUCCESS == code) {
*pLogicSubplan = pSubplan;
} else {
nodesDestroyNode(pSubplan);
}
return code;
}
\ No newline at end of file
source/libs/planner/src/planner.c
浏览文件 @ 3e455e37
......@@ -17,20 +17,29 @@
#include "planInt.h"
int32_t optimize(SPlanContext* pCxt, SLogicNode* pLogicNode) {
return TSDB_CODE_SUCCESS;
}
int32_t qCreateQueryPlan(SPlanContext* pCxt, SQueryPlan** pPlan, SArray* pExecNodeList) {
SLogicNode* pLogicNode = NULL;
SLogicSubplan* pLogicSubplan = NULL;
SQueryLogicPlan* pLogicPlan = NULL;
int32_t code = createLogicPlan(pCxt, &pLogicNode);
if (TSDB_CODE_SUCCESS == code) {
code = optimize(pCxt, pLogicNode);
code = optimizeLogicPlan(pCxt, pLogicNode);
}
if (TSDB_CODE_SUCCESS == code) {
code = splitLogicPlan(pCxt, pLogicNode, &pLogicSubplan);
}
if (TSDB_CODE_SUCCESS == code) {
code = scaleOutLogicPlan(pCxt, pLogicSubplan, &pLogicPlan);
}
if (TSDB_CODE_SUCCESS == code) {
code = createPhysiPlan(pCxt, pLogicNode, pPlan, pExecNodeList);
code = createPhysiPlan(pCxt, pLogicPlan, pPlan, pExecNodeList);
}
nodesDestroyNode(pLogicNode);
nodesDestroyNode(pLogicSubplan);
nodesDestroyNode(pLogicPlan);
terrno = code;
return code;
}
......
source/libs/planner/test/plannerTest.cpp
浏览文件 @ 3e455e37
......@@ -26,11 +26,6 @@ using namespace testing;
class PlannerTest : public Test {
protected:
enum TestTarget {
TEST_LOGIC_PLAN,
TEST_PHYSICAL_PLAN
};
void setDatabase(const string& acctId, const string& db) {
acctId_ = acctId;
db_ = db;
......@@ -46,7 +41,7 @@ protected:
cxt_.pSql = sqlBuf_.c_str();
}
bool run(TestTarget target = TEST_PHYSICAL_PLAN) {
bool run() {
int32_t code = qParseQuerySql(&cxt_, &query_);
if (code != TSDB_CODE_SUCCESS) {
......@@ -56,12 +51,12 @@ protected:
const string syntaxTreeStr = toString(query_->pRoot, false);
SLogicNode* pLogicPlan = nullptr;
SLogicNode* pLogicNode = nullptr;
SPlanContext cxt = { .queryId = 1, .acctId = 0 };
setPlanContext(query_, &cxt);
code = createLogicPlan(&cxt, &pLogicPlan);
code = createLogicPlan(&cxt, &pLogicNode);
if (code != TSDB_CODE_SUCCESS) {
cout << "sql:[" << cxt_.pSql << "] logic plan code:" << code << ", strerror:" << tstrerror(code) << endl;
cout << "sql:[" << cxt_.pSql << "] createLogicPlan code:" << code << ", strerror:" << tstrerror(code) << endl;
return false;
}
......@@ -69,24 +64,37 @@ protected:
cout << "syntax test : " << endl;
cout << syntaxTreeStr << endl;
cout << "unformatted logic plan : " << endl;
cout << toString((const SNode*)pLogicPlan, false) << endl;
if (TEST_PHYSICAL_PLAN == target) {
SQueryPlan* pPlan = nullptr;
code = createPhysiPlan(&cxt, pLogicPlan, &pPlan, NULL);
if (code != TSDB_CODE_SUCCESS) {
cout << "sql:[" << cxt_.pSql << "] physical plan code:" << code << ", strerror:" << tstrerror(code) << endl;
return false;
}
cout << "unformatted physical plan : " << endl;
cout << toString((const SNode*)pPlan, false) << endl;
SNode* pNode;
FOREACH(pNode, pPlan->pSubplans) {
SNode* pSubplan;
FOREACH(pSubplan, ((SNodeListNode*)pNode)->pNodeList) {
cout << "unformatted physical subplan : " << endl;
cout << toString(pSubplan, false) << endl;
}
cout << toString((const SNode*)pLogicNode, false) << endl;
SLogicSubplan* pLogicSubplan = nullptr;
code = splitLogicPlan(&cxt, pLogicNode, &pLogicSubplan);
if (code != TSDB_CODE_SUCCESS) {
cout << "sql:[" << cxt_.pSql << "] splitLogicPlan code:" << code << ", strerror:" << tstrerror(code) << endl;
return false;
}
SQueryLogicPlan* pLogicPlan = NULL;
code = scaleOutLogicPlan(&cxt, pLogicSubplan, &pLogicPlan);
if (code != TSDB_CODE_SUCCESS) {
cout << "sql:[" << cxt_.pSql << "] createPhysiPlan code:" << code << ", strerror:" << tstrerror(code) << endl;
return false;
}
SQueryPlan* pPlan = nullptr;
code = createPhysiPlan(&cxt, pLogicPlan, &pPlan, NULL);
if (code != TSDB_CODE_SUCCESS) {
cout << "sql:[" << cxt_.pSql << "] createPhysiPlan code:" << code << ", strerror:" << tstrerror(code) << endl;
return false;
}
cout << "unformatted physical plan : " << endl;
cout << toString((const SNode*)pPlan, false) << endl;
SNode* pNode;
FOREACH(pNode, pPlan->pSubplans) {
SNode* pSubplan;
FOREACH(pSubplan, ((SNodeListNode*)pNode)->pNodeList) {
cout << "unformatted physical subplan : " << endl;
cout << toString(pSubplan, false) << endl;
}
}
......@@ -120,14 +128,6 @@ private:
cout << "sql:[" << cxt_.pSql << "] toString code:" << code << ", strerror:" << tstrerror(code) << endl;
return string();
}
SNode* pNode;
code = nodesStringToNode(pStr, &pNode);
if (code != TSDB_CODE_SUCCESS) {
tfree(pStr);
cout << "sql:[" << cxt_.pSql << "] toObject code:" << code << ", strerror:" << tstrerror(code) << endl;
return string();
}
nodesDestroyNode(pNode);
string str(pStr);
tfree(pStr);
return str;
......@@ -185,6 +185,13 @@ TEST_F(PlannerTest, interval) {
ASSERT_TRUE(run());
}
TEST_F(PlannerTest, sessionWindow) {
setDatabase("root", "test");
bind("SELECT count(*) FROM t1 session(ts, 10s)");
ASSERT_TRUE(run());
}
TEST_F(PlannerTest, showTables) {
setDatabase("root", "test");
......
tests/script/jenkins/basic.txt
浏览文件 @ 3e455e37
......@@ -15,10 +15,13 @@
# ---- insert
./test.sh -f tsim/insert/basic0.sim
./test.sh -f tsim/insert/basic1.sim
./test.sh -f tsim/insert/backquote.sim
./test.sh -f tsim/insert/null.sim
# ---- query
./test.sh -f tsim/query/interval.sim
#./test.sh -f tsim/query/interval-offset.sim
# ---- table
./test.sh -f tsim/table/basic1.sim
......
tests/script/tsim/db/basic7.sim
浏览文件 @ 3e455e37
......@@ -14,6 +14,7 @@ sql insert into tb1 values (now, 1);
sql show stables
if $rows != 1 then
print $rows
return -1
endi
......
tests/script/tsim/insert/backquote.sim
浏览文件 @ 3e455e37
......@@ -36,8 +36,9 @@ while $dbCnt < 2
$dbCnt = $dbCnt + 1
print =============== create super table, include all type
sql create table `stable` (`timestamp` timestamp, `int` int, `binary` binary(16), `nchar` nchar(16)) tags (`float` float, `Binary` binary(16), `Nchar` nchar(16))
sql create table `Stable` (`timestamp` timestamp, `int` int, `Binary` binary(32), `Nchar` nchar(32)) tags (`float` float, `binary` binary(16), `nchar` nchar(16))
print notes: after nchar show ok, modify binary to nchar
sql create table `stable` (`timestamp` timestamp, `int` int, `binary` binary(16), `nchar` binary(16)) tags (`float` float, `Binary` binary(16), `Nchar` nchar(16))
sql create table `Stable` (`timestamp` timestamp, `int` int, `Binary` binary(32), `Nchar` binary(32)) tags (`float` float, `binary` binary(16), `nchar` nchar(16))
sql show stables
print rows: $rows
......@@ -47,10 +48,14 @@ while $dbCnt < 2
return -1
endi
if $data00 != Stable then
return -1
if $data00 != stable then
return -1
endi
endi
if $data10 != stable then
return -1
if $data10 != Stable then
if $data10 != stable then
return -1
endi
endi
print =============== create child table
......@@ -145,26 +150,26 @@ while $dbCnt < 2
return -1
endi
print =============== query data from st, but not support select * from super table, waiting fix
sql select count(*) from `stable`
print rows: $rows
print $data00 $data01 $data02 $data03
if $rows != 1 then
return -1
endi
if $data00 != 4 then
return -1
endi
sql select count(*) from `Stable`
print rows: $rows
print $data00 $data01 $data02 $data03
if $rows != 1 then
return -1
endi
if $data00 != 4 then
return -1
endi
#sql select * from st
#print =============== query data from st, but not support select * from super table, waiting fix
#sql select count(*) from `stable`
#print rows: $rows
#print $data00 $data01 $data02 $data03
#if $rows != 1 then
# return -1
#endi
#if $data00 != 4 then
# return -1
#endi
#sql select count(*) from `Stable`
#print rows: $rows
#print $data00 $data01 $data02 $data03
#if $rows != 1 then
# return -1
#endi
#if $data00 != 4 then
# return -1
#endi
#sql select * from `stable`
#if $rows != 4 then
# return -1
#endi
......@@ -178,7 +183,7 @@ system sh/exec.sh -n dnode1 -s start
$loop_cnt = 0
check_dnode_ready:
$loop_cnt = $loop_cnt + 1
sleep 100
sleep 200
if $loop_cnt == 10 then
print ====> dnode not ready!
return -1
......@@ -228,10 +233,14 @@ while $dbCnt < 2
return -1
endi
if $data00 != Stable then
return -1
if $data00 != stable then
return -1
endi
endi
if $data10 != stable then
return -1
if $data10 != Stable then
if $data10 != stable then
return -1
endi
endi
sql show tables
......@@ -313,26 +322,26 @@ while $dbCnt < 2
return -1
endi
print =============== query data from st, but not support select * from super table, waiting fix
sql select count(*) from `stable`
print rows: $rows
print $data00 $data01 $data02 $data03
if $rows != 1 then
return -1
endi
if $data00 != 4 then
return -1
endi
sql select count(*) from `Stable`
print rows: $rows
print $data00 $data01 $data02 $data03
if $rows != 1 then
return -1
endi
if $data00 != 4 then
return -1
endi
#sql select * from st
#print =============== query data from st, but not support select * from super table, waiting fix
#sql select count(*) from `stable`
#print rows: $rows
#print $data00 $data01 $data02 $data03
#if $rows != 1 then
# return -1
#endi
#if $data00 != 4 then
# return -1
#endi
#sql select count(*) from `Stable`
#print rows: $rows
#print $data00 $data01 $data02 $data03
#if $rows != 1 then
# return -1
#endi
#if $data00 != 4 then
# return -1
#endi
#sql select * from `stable`
#if $rows != 4 then
# return -1
#endi
......
tests/script/tsim/insert/basic1.sim
浏览文件 @ 3e455e37
......@@ -55,7 +55,8 @@ if $rows != 4 then
return -1
endi
if $data01 != true then
if $data01 != 1 then
print expect 1, actual: $data01
return -1
endi
......@@ -80,7 +81,7 @@ system sh/exec.sh -n dnode1 -s start
$loop_cnt = 0
check_dnode_ready:
$loop_cnt = $loop_cnt + 1
sleep 100
sleep 200
if $loop_cnt == 10 then
print ====> dnode not ready!
return -1
......@@ -105,7 +106,7 @@ if $rows != 4 then
return -1
endi
if $data01 != true then
if $data01 != 1 then
return -1
endi
......
tests/script/tsim/testCaseSuite.sim
浏览文件 @ 3e455e37
......@@ -9,9 +9,9 @@ run tsim/db/error1.sim
run tsim/dnode/basic1.sim
run tsim/insert/basic0.sim
#run tsim/insert/basic1.sim # TD-14246
#run tsim/insert/backquote.sim # TD-14261
#run tsim/insert/null.sim
run tsim/insert/basic1.sim
run tsim/insert/backquote.sim
run tsim/insert/null.sim
run tsim/query/interval.sim
#run tsim/query/interval-offset.sim # TD-14266
......
tools/shell/src/shellMain.c
浏览文件 @ 3e455e37
......@@ -58,7 +58,7 @@ static struct argp_option options[] = {
{"check", 'k', "CHECK", 0, "Check tables."},
{"database", 'd', "DATABASE", 0, "Database to use when connecting to the server."},
{"timezone", 'z', "TIMEZONE", 0, "Time zone of the shell, default is local."},
{"netrole", 'n', "NETROLE", 0, "Net role when network connectivity test, default is startup, options: client|server|rpc|startup|sync|speen|fqdn."},
{"netrole", 'n', "NETROLE", 0, "Net role when network connectivity test, default is startup, options: client|server|rpc|startup|sync|speed|fqdn."},
{"pktlen", 'l', "PKTLEN", 0, "Packet length used for net test, default is 1000 bytes."},
{"pktnum", 'N', "PKTNUM", 0, "Packet numbers used for net test, default is 100."},
// Shuduo: 3.0 does not support UDP any more
......
tools/shell/src/tnettest.c
浏览文件 @ 3e455e37
......@@ -406,7 +406,7 @@ static int32_t taosNetCheckRpc(const char* serverFqdn, uint16_t port, uint16_t p
reqMsg.code = 0;
reqMsg.handle = NULL; // rpc handle returned to app
reqMsg.ahandle = NULL; // app handle set by client
strcpy(reqMsg.pCont, "nettest");
strcpy(reqMsg.pCont, "dnode-nettest");
rpcSendRecv(pRpcConn, &epSet, &reqMsg, &rspMsg);
......@@ -442,7 +442,7 @@ static void taosNetTestStartup(char *host, int32_t port) {
SStartupReq *pStep = malloc(sizeof(SStartupReq));
while (1) {
int32_t code = taosNetCheckRpc(host, port + TSDB_PORT_DNODEDNODE, 20, 0, pStep);
int32_t code = taosNetCheckRpc(host, port, 20, 0, pStep);
if (code > 0) {
code = taosNetParseStartup(pStep);
}
......@@ -499,48 +499,46 @@ static void taosNetCheckSync(char *host, int32_t port) {
}
static void taosNetTestRpc(char *host, int32_t startPort, int32_t pkgLen) {
int32_t endPort = startPort + TSDB_PORT_SYNC;
char spi = 0;
uInfo("check rpc, host:%s startPort:%d endPort:%d pkgLen:%d\n", host, startPort, endPort, pkgLen);
uInfo("check rpc, host:%s Port:%d pkgLen:%d\n", host, startPort, pkgLen);
for (uint16_t port = startPort; port < endPort; port++) {
int32_t sendpkgLen;
if (pkgLen <= tsRpcMaxUdpSize) {
uint16_t port = startPort;
int32_t sendpkgLen;
if (pkgLen <= tsRpcMaxUdpSize) {
sendpkgLen = tsRpcMaxUdpSize + 1000;
} else {
} else {
sendpkgLen = pkgLen;
}
}
tsRpcForceTcp = 1;
int32_t ret = taosNetCheckRpc(host, port, sendpkgLen, spi, NULL);
if (ret < 0) {
tsRpcForceTcp = 1;
int32_t ret = taosNetCheckRpc(host, port, sendpkgLen, spi, NULL);
if (ret < 0) {
printf("failed to test TCP port:%d\n", port);
} else {
} else {
printf("successed to test TCP port:%d\n", port);
}
}
if (pkgLen >= tsRpcMaxUdpSize) {
if (pkgLen >= tsRpcMaxUdpSize) {
sendpkgLen = tsRpcMaxUdpSize - 1000;
} else {
} else {
sendpkgLen = pkgLen;
}
tsRpcForceTcp = 0;
ret = taosNetCheckRpc(host, port, pkgLen, spi, NULL);
if (ret < 0) {
}
/*
tsRpcForceTcp = 0;
ret = taosNetCheckRpc(host, port, pkgLen, spi, NULL);
if (ret < 0) {
printf("failed to test UDP port:%d\n", port);
} else {
} else {
printf("successed to test UDP port:%d\n", port);
}
}
*/
taosNetCheckSync(host, startPort + TSDB_PORT_SYNC);
taosNetCheckSync(host, startPort);
}
static void taosNetTestClient(char *host, int32_t startPort, int32_t pkgLen) {
int32_t endPort = startPort + 11;
uInfo("work as client, host:%s startPort:%d endPort:%d pkgLen:%d\n", host, startPort, endPort, pkgLen);
uInfo("work as client, host:%s Port:%d pkgLen:%d\n", host, startPort, pkgLen);
uint32_t serverIp = taosGetIpv4FromFqdn(host);
if (serverIp == 0xFFFFFFFF) {
......@@ -549,15 +547,14 @@ static void taosNetTestClient(char *host, int32_t startPort, int32_t pkgLen) {
}
uInfo("server ip:%s is resolved from host:%s", taosIpStr(serverIp), host);
taosNetCheckPort(serverIp, startPort, endPort, pkgLen);
taosNetCheckPort(serverIp, startPort, startPort, pkgLen);
}
static void taosNetTestServer(char *host, int32_t startPort, int32_t pkgLen) {
int32_t endPort = startPort + 11;
uInfo("work as server, host:%s startPort:%d endPort:%d pkgLen:%d\n", host, startPort, endPort, pkgLen);
uInfo("work as server, host:%s Port:%d pkgLen:%d\n", host, startPort, pkgLen);
int32_t port = startPort;
int32_t num = endPort - startPort + 1;
int32_t num = 1;
if (num < 0) num = 1;
TdThread *pids = malloc(2 * num * sizeof(TdThread));
......
taos-tools @ 8145dd17
比较 f36b07f7...8145dd17
Subproject commit f36b07f710d661dca88fdd70e73b5e3e16a960e0
Subproject commit 8145dd1713ab9e7652ea621ca7e6895fd0b21d65
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或