Skip to content

T
TDengine

taosdata / TDengine
接近 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
提交 7a6f94a8 编写于 作者: C Cary Xu
浏览文件
操作

Merge branch '3.0' into feature/TD-14481-3.0

上级 6da9abd8 c6782e96
隐藏空白更改
内联 并排
Showing with 950 addition and 586 deletion
include/common/tmsg.h
浏览文件 @ 7a6f94a8
......@@ -2427,6 +2427,7 @@ typedef struct {
static FORCE_INLINE void tDestroyTSma(STSma* pSma) {
if (pSma) {
taosMemoryFreeClear(pSma->dstTbName);
taosMemoryFreeClear(pSma->expr);
taosMemoryFreeClear(pSma->tagsFilter);
}
......@@ -2455,7 +2456,7 @@ int32_t tEncodeSVCreateTSmaReq(SEncoder* pCoder, const SVCreateTSmaReq* pReq);
int32_t tDecodeSVCreateTSmaReq(SDecoder* pCoder, SVCreateTSmaReq* pReq);
int32_t tEncodeTSma(SEncoder* pCoder, const STSma* pSma);
int32_t tDecodeTSma(SDecoder* pCoder, STSma* pSma);
int32_t tDecodeTSma(SDecoder* pCoder, STSma* pSma, bool deepCopy);
static int32_t tEncodeTSmaWrapper(SEncoder* pEncoder, const STSmaWrapper* pReq) {
if (tEncodeI32(pEncoder, pReq->number) < 0) return -1;
......@@ -2465,10 +2466,10 @@ static int32_t tEncodeTSmaWrapper(SEncoder* pEncoder, const STSmaWrapper* pReq)
return 0;
}
static int32_t tDecodeTSmaWrapper(SDecoder* pDecoder, STSmaWrapper* pReq) {
static int32_t tDecodeTSmaWrapper(SDecoder* pDecoder, STSmaWrapper* pReq, bool deepCopy) {
if (tDecodeI32(pDecoder, &pReq->number) < 0) return -1;
for (int32_t i = 0; i < pReq->number; ++i) {
tDecodeTSma(pDecoder, pReq->tSma + i);
tDecodeTSma(pDecoder, pReq->tSma + i, deepCopy);
}
return 0;
}
......
include/common/tmsgdef.h
浏览文件 @ 7a6f94a8
......@@ -97,7 +97,6 @@ enum {
TD_DEF_MSG_TYPE(TDMT_MND_CREATE_MNODE, "create-mnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_ALTER_MNODE, "alter-mnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_DROP_MNODE, "drop-mnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_SET_STANDBY, "set-mnode-standby", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_CREATE_QNODE, "create-qnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_ALTER_QNODE, "alter-qnode", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_DROP_QNODE, "drop-qnode", NULL, NULL)
......@@ -239,6 +238,8 @@ enum {
TD_DEF_MSG_TYPE(TDMT_SYNC_SNAPSHOT_SEND, "sync-snapshot-send", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_SYNC_SNAPSHOT_RSP, "sync-snapshot-rsp", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_SYNC_LEADER_TRANSFER, "sync-leader-transfer", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_SYNC_SET_MNODE_STANDBY, "set-mnode-standby", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_SYNC_SET_VNODE_STANDBY, "set-vnode-standby", NULL, NULL)
#if defined(TD_MSG_NUMBER_)
TDMT_MAX
......
include/dnode/mnode/mnode.h
浏览文件 @ 7a6f94a8
......@@ -59,7 +59,13 @@ void mndClose(SMnode *pMnode);
* @param pMnode The mnode object.
*/
int32_t mndStart(SMnode *pMnode);
void mndStop(SMnode *pMnode);
/**
* @brief Stop mnode
*
* @param pMnode The mnode object.
*/
void mndStop(SMnode *pMnode);
/**
* @brief Get mnode monitor info.
......@@ -71,17 +77,25 @@ void mndStop(SMnode *pMnode);
* @return int32_t 0 for success, -1 for failure.
*/
int32_t mndGetMonitorInfo(SMnode *pMnode, SMonClusterInfo *pCluster, SMonVgroupInfo *pVgroup, SMonGrantInfo *pGrant);
/**
* @brief Get mnode loads for status msg.
*
* @param pMnode The mnode object.
* @param pLoad
* @return int32_t 0 for success, -1 for failure.
*/
int32_t mndGetLoad(SMnode *pMnode, SMnodeLoad *pLoad);
/**
* @brief Process the read, write, sync request.
* @brief Process the rpc, sync request.
*
* @param pMsg The request msg.
* @return int32_t 0 for success, -1 for failure.
*/
int32_t mndProcessRpcMsg(SRpcMsg *pMsg);
int32_t mndProcessSyncMsg(SRpcMsg *pMsg);
int32_t mndPreprocessQueryMsg(SMnode * pMnode, SRpcMsg * pMsg);
int32_t mndPreProcessMsg(SRpcMsg *pMsg);
/**
* @brief Generate machine code
......
include/libs/function/functionMgt.h
浏览文件 @ 7a6f94a8
......@@ -144,6 +144,8 @@ typedef enum EFunctionType {
FUNCTION_TYPE_LAST_MERGE,
FUNCTION_TYPE_AVG_PARTIAL,
FUNCTION_TYPE_AVG_MERGE,
FUNCTION_TYPE_STDDEV_PARTIAL,
FUNCTION_TYPE_STDDEV_MERGE,
// user defined funcion
FUNCTION_TYPE_UDF = 10000
......
include/libs/parser/parser.h
浏览文件 @ 7a6f94a8
......@@ -65,6 +65,7 @@ void qDestroyQuery(SQuery* pQueryNode);
int32_t qExtractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema);
int32_t qSetSTableIdForRsma(SNode* pStmt, int64_t uid);
void qCleanupKeywordsTable();
int32_t qBuildStmtOutput(SQuery* pQuery, SHashObj* pVgHash, SHashObj* pBlockHash);
int32_t qResetStmtDataBlock(void* block, bool keepBuf);
......
include/util/taoserror.h
浏览文件 @ 7a6f94a8
......@@ -653,6 +653,7 @@ int32_t* taosGetErrno();
#define TSDB_CODE_PAR_INVALID_DELETE_WHERE TAOS_DEF_ERROR_CODE(0, 0x2655)
#define TSDB_CODE_PAR_INVALID_REDISTRIBUTE_VG TAOS_DEF_ERROR_CODE(0, 0x2656)
#define TSDB_CODE_PAR_FILL_NOT_ALLOWED_FUNC TAOS_DEF_ERROR_CODE(0, 0x2657)
#define TSDB_CODE_PAR_INVALID_WINDOW_PC TAOS_DEF_ERROR_CODE(0, 0x2658)
//planner
#define TSDB_CODE_PLAN_INTERNAL_ERROR TAOS_DEF_ERROR_CODE(0, 0x2700)
......@@ -684,12 +685,15 @@ int32_t* taosGetErrno();
#define TSDB_CODE_SML_INVALID_DB_CONF TAOS_DEF_ERROR_CODE(0, 0x3003)
//tsma
#define TSDB_CODE_TSMA_ALREADY_EXIST TAOS_DEF_ERROR_CODE(0, 0x3100)
#define TSDB_CODE_TSMA_NO_INDEX_IN_META TAOS_DEF_ERROR_CODE(0, 0x3101)
#define TSDB_CODE_TSMA_INVALID_ENV TAOS_DEF_ERROR_CODE(0, 0x3102)
#define TSDB_CODE_TSMA_INVALID_STAT TAOS_DEF_ERROR_CODE(0, 0x3103)
#define TSDB_CODE_TSMA_NO_INDEX_IN_CACHE TAOS_DEF_ERROR_CODE(0, 0x3104)
#define TSDB_CODE_TSMA_RM_SKEY_IN_HASH TAOS_DEF_ERROR_CODE(0, 0x3105)
#define TSDB_CODE_TSMA_INIT_FAILED TAOS_DEF_ERROR_CODE(0, 0x3100)
#define TSDB_CODE_TSMA_ALREADY_EXIST TAOS_DEF_ERROR_CODE(0, 0x3101)
#define TSDB_CODE_TSMA_NO_INDEX_IN_META TAOS_DEF_ERROR_CODE(0, 0x3102)
#define TSDB_CODE_TSMA_INVALID_ENV TAOS_DEF_ERROR_CODE(0, 0x3103)
#define TSDB_CODE_TSMA_INVALID_STAT TAOS_DEF_ERROR_CODE(0, 0x3104)
#define TSDB_CODE_TSMA_INVALID_PTR TAOS_DEF_ERROR_CODE(0, 0x3105)
#define TSDB_CODE_TSMA_INVALID_PARA TAOS_DEF_ERROR_CODE(0, 0x3106)
#define TSDB_CODE_TSMA_NO_INDEX_IN_CACHE TAOS_DEF_ERROR_CODE(0, 0x3107)
//rsma
#define TSDB_CODE_RSMA_INVALID_ENV TAOS_DEF_ERROR_CODE(0, 0x3150)
......
source/client/src/clientMain.c
浏览文件 @ 7a6f94a8
......@@ -25,6 +25,7 @@
#include "tref.h"
#include "trpc.h"
#include "version.h"
#include "functionMgt.h"
#define TSC_VAR_NOT_RELEASE 1
#define TSC_VAR_RELEASED 0
......@@ -61,6 +62,9 @@ void taos_cleanup(void) {
cleanupTaskQueue();
fmFuncMgtDestroy();
qCleanupKeywordsTable();
id = clientConnRefPool;
clientConnRefPool = -1;
taosCloseRef(id);
......
source/common/src/tmsg.c
浏览文件 @ 7a6f94a8
......@@ -2653,7 +2653,7 @@ int32_t tSerializeSSTbHbRsp(void *buf, int32_t bufLen, SSTbHbRsp *pRsp) {
}
int32_t numOfIndex = taosArrayGetSize(pRsp->pIndexRsp);
if (tEncodeI32(&encoder, numOfIndex) < 0) return -1;
if (tEncodeI32(&encoder, numOfIndex) < 0) return -1;
for (int32_t i = 0; i < numOfIndex; ++i) {
STableIndexRsp *pIndexRsp = taosArrayGet(pRsp->pIndexRsp, i);
if (tEncodeCStr(&encoder, pIndexRsp->tbName) < 0) return -1;
......@@ -2738,7 +2738,7 @@ int32_t tDeserializeSSTbHbRsp(void *buf, int32_t bufLen, SSTbHbRsp *pRsp) {
}
taosArrayPush(pRsp->pIndexRsp, &tableIndexRsp);
}
tEndDecode(&decoder);
tDecoderClear(&decoder);
......@@ -4000,7 +4000,7 @@ int32_t tEncodeTSma(SEncoder *pCoder, const STSma *pSma) {
return 0;
}
int32_t tDecodeTSma(SDecoder *pCoder, STSma *pSma) {
int32_t tDecodeTSma(SDecoder *pCoder, STSma *pSma, bool deepCopy) {
if (tDecodeI8(pCoder, &pSma->version) < 0) return -1;
if (tDecodeI8(pCoder, &pSma->intervalUnit) < 0) return -1;
if (tDecodeI8(pCoder, &pSma->slidingUnit) < 0) return -1;
......@@ -4012,17 +4012,30 @@ int32_t tDecodeTSma(SDecoder *pCoder, STSma *pSma) {
if (tDecodeI64(pCoder, &pSma->indexUid) < 0) return -1;
if (tDecodeI64(pCoder, &pSma->tableUid) < 0) return -1;
if (tDecodeI64(pCoder, &pSma->dstTbUid) < 0) return -1;
if (tDecodeCStr(pCoder, &pSma->dstTbName) < 0) return -1;
if (deepCopy) {
if (tDecodeCStrAlloc(pCoder, &pSma->dstTbName) < 0) return -1;
} else {
if (tDecodeCStr(pCoder, &pSma->dstTbName) < 0) return -1;
}
if (tDecodeI64(pCoder, &pSma->interval) < 0) return -1;
if (tDecodeI64(pCoder, &pSma->offset) < 0) return -1;
if (tDecodeI64(pCoder, &pSma->sliding) < 0) return -1;
if (pSma->exprLen > 0) {
if (tDecodeCStr(pCoder, &pSma->expr) < 0) return -1;
if (deepCopy) {
if (tDecodeCStrAlloc(pCoder, &pSma->expr) < 0) return -1;
} else {
if (tDecodeCStr(pCoder, &pSma->expr) < 0) return -1;
}
} else {
pSma->expr = NULL;
}
if (pSma->tagsFilterLen > 0) {
if (tDecodeCStr(pCoder, &pSma->tagsFilter) < 0) return -1;
if (deepCopy) {
if (tDecodeCStrAlloc(pCoder, &pSma->tagsFilter) < 0) return -1;
} else {
if (tDecodeCStr(pCoder, &pSma->tagsFilter) < 0) return -1;
}
} else {
pSma->tagsFilter = NULL;
}
......@@ -4045,7 +4058,7 @@ int32_t tEncodeSVCreateTSmaReq(SEncoder *pCoder, const SVCreateTSmaReq *pReq) {
int32_t tDecodeSVCreateTSmaReq(SDecoder *pCoder, SVCreateTSmaReq *pReq) {
if (tStartDecode(pCoder) < 0) return -1;
tDecodeTSma(pCoder, pReq);
tDecodeTSma(pCoder, pReq, false);
tEndDecode(pCoder);
return 0;
......@@ -4879,4 +4892,3 @@ int32_t tDecodeSTqOffset(SDecoder *pDecoder, STqOffset *pOffset) {
if (tDecodeCStrTo(pDecoder, pOffset->subKey) < 0) return -1;
return 0;
}
source/dnode/mgmt/mgmt_mnode/src/mmHandle.c
浏览文件 @ 7a6f94a8
......@@ -155,7 +155,6 @@ SArray *mmGetMsgHandles() {
if (dmSetMgmtHandle(pArray, TDMT_MND_CREATE_MNODE, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_ALTER_MNODE, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_ALTER_MNODE_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_SET_STANDBY_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_DROP_MNODE, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_CREATE_QNODE, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_DROP_QNODE, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
......@@ -236,7 +235,10 @@ SArray *mmGetMsgHandles() {
if (dmSetMgmtHandle(pArray, TDMT_SYNC_APPEND_ENTRIES_REPLY, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_SYNC_SNAPSHOT_SEND, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_SYNC_SNAPSHOT_RSP, mmPutMsgToSyncQueue, 1) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_SET_STANDBY, mmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_SYNC_SET_MNODE_STANDBY, mmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_SYNC_SET_MNODE_STANDBY_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_SYNC_SET_VNODE_STANDBY_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
code = 0;
......
source/dnode/mgmt/mgmt_mnode/src/mmWorker.c
浏览文件 @ 7a6f94a8
......@@ -114,7 +114,8 @@ int32_t mmPutMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
}
int32_t mmPutMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
if (mndPreprocessQueryMsg(pMgmt->pMnode, pMsg) != 0) {
pMsg->info.node = pMgmt->pMnode;
if (mndPreProcessMsg(pMsg) != 0) {
dError("msg:%p, failed to pre-process in mnode since %s, type:%s", pMsg, terrstr(), TMSG_INFO(pMsg->msgType));
return -1;
}
......
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
浏览文件 @ 7a6f94a8
......@@ -374,6 +374,7 @@ SArray *vmGetMsgHandles() {
if (dmSetMgmtHandle(pArray, TDMT_SYNC_REQUEST_VOTE_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_SYNC_APPEND_ENTRIES, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_SYNC_APPEND_ENTRIES_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_SYNC_SET_VNODE_STANDBY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
code = 0;
......
source/dnode/mnode/impl/inc/mndInt.h
浏览文件 @ 7a6f94a8
......@@ -124,8 +124,6 @@ void mndReleaseRpcRef(SMnode *pMnode);
void mndSetRestore(SMnode *pMnode, bool restored);
void mndSetStop(SMnode *pMnode);
bool mndGetStop(SMnode *pMnode);
int32_t mndAcquireSyncRef(SMnode *pMnode);
void mndReleaseSyncRef(SMnode *pMnode);
#ifdef __cplusplus
}
......
source/dnode/mnode/impl/inc/mndSma.h
浏览文件 @ 7a6f94a8
......@@ -26,6 +26,8 @@ int32_t mndInitSma(SMnode *pMnode);
void mndCleanupSma(SMnode *pMnode);
SSmaObj *mndAcquireSma(SMnode *pMnode, char *smaName);
void mndReleaseSma(SMnode *pMnode, SSmaObj *pSma);
int32_t mndDropSmasByStb(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb);
int32_t mndDropSmasByDb(SMnode *pMnode, STrans *pTrans, SDbObj *pDb);
int32_t mndGetTableSma(SMnode *pMnode, char *tbFName, STableIndexRsp *rsp, bool *exist);
#ifdef __cplusplus
......
source/dnode/mnode/impl/src/mndDb.c
浏览文件 @ 7a6f94a8
......@@ -935,6 +935,7 @@ static int32_t mndDropDb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb) {
if (mndDropOffsetByDB(pMnode, pTrans, pDb) != 0) goto _OVER;
if (mndDropSubByDB(pMnode, pTrans, pDb) != 0) goto _OVER;
if (mndDropTopicByDB(pMnode, pTrans, pDb) != 0) goto _OVER;
if (mndDropSmasByDb(pMnode, pTrans, pDb) != 0) goto _OVER;
if (mndSetDropDbRedoActions(pMnode, pTrans, pDb) != 0) goto _OVER;
SUserObj *pUser = mndAcquireUser(pMnode, pDb->createUser);
......
source/dnode/mnode/impl/src/mndMain.c
浏览文件 @ 7a6f94a8
......@@ -393,11 +393,6 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
return TAOS_SYNC_OTHER_ERROR;
}
if (mndAcquireSyncRef(pMnode) != 0) {
mError("failed to process sync msg:%p type:%s since %s", pMsg, TMSG_INFO(pMsg->msgType), terrstr());
return TAOS_SYNC_OTHER_ERROR;
}
char logBuf[512] = {0};
char *syncNodeStr = sync2SimpleStr(pMgmt->sync);
snprintf(logBuf, sizeof(logBuf), "==vnodeProcessSyncReq== msgType:%d, syncNode: %s", pMsg->msgType, syncNodeStr);
......@@ -450,7 +445,7 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
SyncSnapshotRsp *pSyncMsg = syncSnapshotRspFromRpcMsg2(pMsg);
code = syncNodeOnSnapshotRspCb(pSyncNode, pSyncMsg);
syncSnapshotRspDestroy(pSyncMsg);
} else if (pMsg->msgType == TDMT_MND_SET_STANDBY) {
} else if (pMsg->msgType == TDMT_SYNC_SET_MNODE_STANDBY) {
code = syncSetStandby(pMgmt->sync);
SRpcMsg rsp = {.code = code, .info = pMsg->info};
tmsgSendRsp(&rsp);
......@@ -491,7 +486,7 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
SyncAppendEntriesReply *pSyncMsg = syncAppendEntriesReplyFromRpcMsg2(pMsg);
code = syncNodeOnAppendEntriesReplyCb(pSyncNode, pSyncMsg);
syncAppendEntriesReplyDestroy(pSyncMsg);
} else if (pMsg->msgType == TDMT_MND_SET_STANDBY) {
} else if (pMsg->msgType == TDMT_SYNC_SET_MNODE_STANDBY) {
code = syncSetStandby(pMgmt->sync);
SRpcMsg rsp = {.code = code, .info = pMsg->info};
tmsgSendRsp(&rsp);
......@@ -501,7 +496,6 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
}
}
mndReleaseSyncRef(pMnode);
return code;
}
......@@ -754,24 +748,3 @@ void mndSetStop(SMnode *pMnode) {
}
bool mndGetStop(SMnode *pMnode) { return pMnode->stopped; }
int32_t mndAcquireSyncRef(SMnode *pMnode) {
int32_t code = 0;
taosThreadRwlockRdlock(&pMnode->lock);
if (pMnode->stopped) {
terrno = TSDB_CODE_APP_NOT_READY;
code = -1;
} else {
int32_t ref = atomic_add_fetch_32(&pMnode->syncRef, 1);
// mTrace("mnode sync is acquired, ref:%d", ref);
}
taosThreadRwlockUnlock(&pMnode->lock);
return code;
}
void mndReleaseSyncRef(SMnode *pMnode) {
taosThreadRwlockRdlock(&pMnode->lock);
int32_t ref = atomic_sub_fetch_32(&pMnode->syncRef, 1);
// mTrace("mnode sync is released, ref:%d", ref);
taosThreadRwlockUnlock(&pMnode->lock);
}
source/dnode/mnode/impl/src/mndMnode.c
浏览文件 @ 7a6f94a8
......@@ -55,7 +55,8 @@ int32_t mndInitMnode(SMnode *pMnode) {
mndSetMsgHandle(pMnode, TDMT_MND_ALTER_MNODE_RSP, mndTransProcessRsp);
mndSetMsgHandle(pMnode, TDMT_MND_DROP_MNODE, mndProcessDropMnodeReq);
mndSetMsgHandle(pMnode, TDMT_DND_DROP_MNODE_RSP, mndTransProcessRsp);
mndSetMsgHandle(pMnode, TDMT_MND_SET_STANDBY_RSP, mndTransProcessRsp);
mndSetMsgHandle(pMnode, TDMT_SYNC_SET_MNODE_STANDBY_RSP, mndTransProcessRsp);
mndSetMsgHandle(pMnode, TDMT_SYNC_SET_VNODE_STANDBY_RSP, mndTransProcessRsp);
mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_MNODE, mndRetrieveMnodes);
mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_MNODE, mndCancelGetNextMnode);
......@@ -511,7 +512,7 @@ static int32_t mndSetDropMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnode
.epSet = dropEpSet,
.pCont = pReq,
.contLen = contLen,
.msgType = TDMT_MND_SET_STANDBY,
.msgType = TDMT_SYNC_SET_MNODE_STANDBY,
.acceptableCode = TSDB_CODE_NODE_NOT_DEPLOYED,
};
......
source/dnode/mnode/impl/src/mndQuery.c
浏览文件 @ 7a6f94a8
......@@ -18,11 +18,10 @@
#include "mndMnode.h"
#include "qworker.h"
int32_t mndPreprocessQueryMsg(SMnode * pMnode, SRpcMsg * pMsg) {
if (TDMT_VND_QUERY != pMsg->msgType) {
return 0;
}
int32_t mndPreProcessMsg(SRpcMsg *pMsg) {
if (TDMT_VND_QUERY != pMsg->msgType) return 0;
SMnode *pMnode = pMsg->info.node;
return qWorkerPreprocessQueryMsg(pMnode->pQuery, pMsg);
}
......
source/dnode/mnode/impl/src/mndSma.c
浏览文件 @ 7a6f94a8
......@@ -38,10 +38,10 @@ static int32_t mndSmaActionInsert(SSdb *pSdb, SSmaObj *pSma);
static int32_t mndSmaActionDelete(SSdb *pSdb, SSmaObj *pSpSmatb);
static int32_t mndSmaActionUpdate(SSdb *pSdb, SSmaObj *pOld, SSmaObj *pNew);
static int32_t mndSmaGetVgEpSet(SMnode *pMnode, SDbObj *pDb, SVgEpSet **ppVgEpSet, int32_t *numOfVgroups);
static int32_t mndProcessMCreateSmaReq(SRpcMsg *pReq);
static int32_t mndProcessMDropSmaReq(SRpcMsg *pReq);
static int32_t mndProcessCreateSmaReq(SRpcMsg *pReq);
static int32_t mndProcessDropSmaReq(SRpcMsg *pReq);
static int32_t mndProcessGetSmaReq(SRpcMsg *pReq);
static int32_t mndProcessGetTbSmaReq(SRpcMsg *pReq);
static int32_t mndProcessGetTbSmaReq(SRpcMsg *pReq);
static int32_t mndRetrieveSma(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
static void mndCancelGetNextSma(SMnode *pMnode, void *pIter);
......@@ -56,8 +56,8 @@ int32_t mndInitSma(SMnode *pMnode) {
.deleteFp = (SdbDeleteFp)mndSmaActionDelete,
};
mndSetMsgHandle(pMnode, TDMT_MND_CREATE_SMA, mndProcessMCreateSmaReq);
mndSetMsgHandle(pMnode, TDMT_MND_DROP_SMA, mndProcessMDropSmaReq);
mndSetMsgHandle(pMnode, TDMT_MND_CREATE_SMA, mndProcessCreateSmaReq);
mndSetMsgHandle(pMnode, TDMT_MND_DROP_SMA, mndProcessDropSmaReq);
mndSetMsgHandle(pMnode, TDMT_VND_CREATE_SMA_RSP, mndTransProcessRsp);
mndSetMsgHandle(pMnode, TDMT_VND_DROP_SMA_RSP, mndTransProcessRsp);
mndSetMsgHandle(pMnode, TDMT_MND_GET_INDEX, mndProcessGetSmaReq);
......@@ -79,7 +79,6 @@ static SSdbRaw *mndSmaActionEncode(SSmaObj *pSma) {
if (pRaw == NULL) goto _OVER;
int32_t dataPos = 0;
SDB_SET_BINARY(pRaw, dataPos, pSma->name, TSDB_TABLE_FNAME_LEN, _OVER)
SDB_SET_BINARY(pRaw, dataPos, pSma->stb, TSDB_TABLE_FNAME_LEN, _OVER)
SDB_SET_BINARY(pRaw, dataPos, pSma->db, TSDB_DB_FNAME_LEN, _OVER)
......@@ -100,6 +99,7 @@ static SSdbRaw *mndSmaActionEncode(SSmaObj *pSma) {
SDB_SET_INT32(pRaw, dataPos, pSma->tagsFilterLen, _OVER)
SDB_SET_INT32(pRaw, dataPos, pSma->sqlLen, _OVER)
SDB_SET_INT32(pRaw, dataPos, pSma->astLen, _OVER)
if (pSma->exprLen > 0) {
SDB_SET_BINARY(pRaw, dataPos, pSma->expr, pSma->exprLen, _OVER)
}
......@@ -115,6 +115,7 @@ static SSdbRaw *mndSmaActionEncode(SSmaObj *pSma) {
SDB_SET_RESERVE(pRaw, dataPos, TSDB_SMA_RESERVE_SIZE, _OVER)
SDB_SET_DATALEN(pRaw, dataPos, _OVER)
terrno = 0;
_OVER:
......@@ -193,6 +194,7 @@ static SSdbRow *mndSmaActionDecode(SSdbRaw *pRaw) {
}
SDB_GET_RESERVE(pRaw, dataPos, TSDB_SMA_RESERVE_SIZE, _OVER)
terrno = 0;
_OVER:
......@@ -383,6 +385,25 @@ static int32_t mndSetCreateSmaVgroupCommitLogs(SMnode *pMnode, STrans *pTrans, S
return 0;
}
static int32_t mndSetUpdateSmaStbCommitLogs(SMnode *pMnode, STrans *pTrans, SStbObj *pStb) {
SStbObj stbObj = {0};
taosRLockLatch(&pStb->lock);
memcpy(&stbObj, pStb, sizeof(SStbObj));
taosRUnLockLatch(&pStb->lock);
stbObj.pColumns = NULL;
stbObj.pTags = NULL;
stbObj.updateTime = taosGetTimestampMs();
stbObj.lock = 0;
stbObj.smaVer++;
SSdbRaw *pCommitRaw = mndStbActionEncode(&stbObj);
if (pCommitRaw == NULL) return -1;
if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) return -1;
if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY) != 0) return -1;
return 0;
}
static int32_t mndSetCreateSmaRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SSmaObj *pSma) {
SSdb *pSdb = pMnode->pSdb;
SVgObj *pVgroup = NULL;
......@@ -457,7 +478,6 @@ static int32_t mndSetCreateSmaVgroupRedoActions(SMnode *pMnode, STrans *pTrans,
pSma->schemaTag.pSchema[0].flags = 0;
snprintf(pSma->schemaTag.pSchema[0].name, TSDB_COL_NAME_LEN, "groupId");
int32_t smaContLen = 0;
void *pSmaReq = mndBuildVCreateSmaReq(pMnode, pVgroup, pSma, &smaContLen);
if (pSmaReq == NULL) return -1;
......@@ -560,6 +580,7 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
if (mndSetCreateSmaVgroupRedoLogs(pMnode, pTrans, &streamObj.fixedSinkVg) != 0) goto _OVER;
if (mndSetCreateSmaCommitLogs(pMnode, pTrans, &smaObj) != 0) goto _OVER;
if (mndSetCreateSmaVgroupCommitLogs(pMnode, pTrans, &streamObj.fixedSinkVg) != 0) goto _OVER;
if (mndSetUpdateSmaStbCommitLogs(pMnode, pTrans, pStb) != 0) goto _OVER;
if (mndSetCreateSmaRedoActions(pMnode, pTrans, pDb, &smaObj) != 0) goto _OVER;
if (mndSetCreateSmaVgroupRedoActions(pMnode, pTrans, pDb, &streamObj.fixedSinkVg, &smaObj) != 0) goto _OVER;
if (mndAddStreamToTrans(pMnode, &streamObj, pCreate->ast, STREAM_TRIGGER_AT_ONCE, 0, pTrans) != 0) goto _OVER;
......@@ -600,7 +621,7 @@ static int32_t mndCheckCreateSmaReq(SMCreateSmaReq *pCreate) {
return 0;
}
static int32_t mndProcessMCreateSmaReq(SRpcMsg *pReq) {
static int32_t mndProcessCreateSmaReq(SRpcMsg *pReq) {
SMnode *pMnode = pReq->info.node;
int32_t code = -1;
SStbObj *pStb = NULL;
......@@ -782,13 +803,17 @@ static int32_t mndSetDropSmaVgroupRedoActions(SMnode *pMnode, STrans *pTrans, SD
}
static int32_t mndDropSma(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SSmaObj *pSma) {
int32_t code = -1;
SVgObj *pVgroup = NULL;
STrans *pTrans = NULL;
int32_t code = -1;
SVgObj *pVgroup = NULL;
SStbObj *pStb = NULL;
STrans *pTrans = NULL;
pVgroup = mndAcquireVgroup(pMnode, pSma->dstVgId);
if (pVgroup == NULL) goto _OVER;
pStb = mndAcquireStb(pMnode, pSma->stb);
if (pStb == NULL) goto _OVER;
pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB, pReq);
if (pTrans == NULL) goto _OVER;
......@@ -799,6 +824,7 @@ static int32_t mndDropSma(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SSmaObj *p
if (mndSetDropSmaVgroupRedoLogs(pMnode, pTrans, pVgroup) != 0) goto _OVER;
if (mndSetDropSmaCommitLogs(pMnode, pTrans, pSma) != 0) goto _OVER;
if (mndSetDropSmaVgroupCommitLogs(pMnode, pTrans, pVgroup) != 0) goto _OVER;
if (mndSetUpdateSmaStbCommitLogs(pMnode, pTrans, pStb) != 0) goto _OVER;
if (mndSetDropSmaRedoActions(pMnode, pTrans, pDb, pSma) != 0) goto _OVER;
if (mndSetDropSmaVgroupRedoActions(pMnode, pTrans, pDb, pVgroup) != 0) goto _OVER;
if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
......@@ -808,10 +834,78 @@ static int32_t mndDropSma(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SSmaObj *p
_OVER:
mndTransDrop(pTrans);
mndReleaseVgroup(pMnode, pVgroup);
mndReleaseStb(pMnode, pStb);
return code;
}
static int32_t mndProcessMDropSmaReq(SRpcMsg *pReq) {
int32_t mndDropSmasByStb(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
SSdb *pSdb = pMnode->pSdb;
SSmaObj *pSma = NULL;
void *pIter = NULL;
SVgObj *pVgroup = NULL;
int32_t code = -1;
while (1) {
pIter = sdbFetch(pSdb, SDB_SMA, pIter, (void **)&pSma);
if (pIter == NULL) break;
if (pSma->stbUid == pStb->uid) {
pVgroup = mndAcquireVgroup(pMnode, pSma->dstVgId);
if (pVgroup == NULL) goto _OVER;
if (mndSetDropSmaVgroupCommitLogs(pMnode, pTrans, pVgroup) != 0) goto _OVER;
if (mndSetDropSmaVgroupRedoActions(pMnode, pTrans, pDb, pVgroup) != 0) goto _OVER;
if (mndSetDropSmaCommitLogs(pMnode, pTrans, pSma) != 0) goto _OVER;
if (mndSetDropSmaRedoActions(pMnode, pTrans, pDb, pSma) != 0) goto _OVER;
mndReleaseVgroup(pMnode, pVgroup);
pVgroup = NULL;
}
sdbRelease(pSdb, pSma);
}
code = 0;
_OVER:
sdbCancelFetch(pSdb, pIter);
sdbRelease(pSdb, pSma);
mndReleaseVgroup(pMnode, pVgroup);
return code;
}
int32_t mndDropSmasByDb(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) {
SSdb *pSdb = pMnode->pSdb;
SSmaObj *pSma = NULL;
void *pIter = NULL;
SVgObj *pVgroup = NULL;
int32_t code = -1;
while (1) {
pIter = sdbFetch(pSdb, SDB_SMA, pIter, (void **)&pSma);
if (pIter == NULL) break;
if (pSma->dbUid == pDb->uid) {
pVgroup = mndAcquireVgroup(pMnode, pSma->dstVgId);
if (pVgroup == NULL) goto _OVER;
if (mndSetDropSmaVgroupCommitLogs(pMnode, pTrans, pVgroup) != 0) goto _OVER;
if (mndSetDropSmaVgroupRedoActions(pMnode, pTrans, pDb, pVgroup) != 0) goto _OVER;
if (mndSetDropSmaCommitLogs(pMnode, pTrans, pSma) != 0) goto _OVER;
mndReleaseVgroup(pMnode, pVgroup);
pVgroup = NULL;
}
sdbRelease(pSdb, pSma);
}
code = 0;
_OVER:
sdbCancelFetch(pSdb, pIter);
sdbRelease(pSdb, pSma);
mndReleaseVgroup(pMnode, pVgroup);
return code;
}
static int32_t mndProcessDropSmaReq(SRpcMsg *pReq) {
SMnode *pMnode = pReq->info.node;
int32_t code = -1;
SUserObj *pUser = NULL;
......@@ -901,13 +995,13 @@ static int32_t mndGetSma(SMnode *pMnode, SUserIndexReq *indexReq, SUserIndexRsp
}
int32_t mndGetTableSma(SMnode *pMnode, char *tbFName, STableIndexRsp *rsp, bool *exist) {
int32_t code = 0;
SSmaObj *pSma = NULL;
SSdb *pSdb = pMnode->pSdb;
void *pIter = NULL;
int32_t code = 0;
SSmaObj *pSma = NULL;
SSdb *pSdb = pMnode->pSdb;
void *pIter = NULL;
STableIndexInfo info;
SStbObj* pStb = mndAcquireStb(pMnode, tbFName);
SStbObj *pStb = mndAcquireStb(pMnode, tbFName);
if (NULL == pStb) {
*exist = false;
return TSDB_CODE_SUCCESS;
......@@ -936,14 +1030,14 @@ int32_t mndGetTableSma(SMnode *pMnode, char *tbFName, STableIndexRsp *rsp, bool
info.dstTbUid = pSma->dstTbUid;
info.dstVgId = pSma->dstVgId;
SVgObj* pVg = mndAcquireVgroup(pMnode, pSma->dstVgId);
SVgObj *pVg = mndAcquireVgroup(pMnode, pSma->dstVgId);
if (pVg == NULL) {
code = -1;
sdbRelease(pSdb, pSma);
return code;
}
info.epSet = mndGetVgroupEpset(pMnode, pVg);
info.expr = taosMemoryMalloc(pSma->exprLen + 1);
if (info.expr == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
......@@ -967,7 +1061,7 @@ int32_t mndGetTableSma(SMnode *pMnode, char *tbFName, STableIndexRsp *rsp, bool
sdbRelease(pSdb, pSma);
}
return code;
}
......@@ -1019,10 +1113,10 @@ _OVER:
static int32_t mndProcessGetTbSmaReq(SRpcMsg *pReq) {
STableIndexReq indexReq = {0};
SMnode *pMnode = pReq->info.node;
int32_t code = -1;
SMnode *pMnode = pReq->info.node;
int32_t code = -1;
STableIndexRsp rsp = {0};
bool exist = false;
bool exist = false;
if (tDeserializeSTableIndexReq(pReq->pCont, pReq->contLen, &indexReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
......@@ -1069,7 +1163,6 @@ _OVER:
return code;
}
static int32_t mndRetrieveSma(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
SMnode *pMnode = pReq->info.node;
SSdb *pSdb = pMnode->pSdb;
......
source/dnode/mnode/impl/src/mndStb.c
浏览文件 @ 7a6f94a8
......@@ -23,6 +23,7 @@
#include "mndPerfSchema.h"
#include "mndScheduler.h"
#include "mndShow.h"
#include "mndSma.h"
#include "mndTopic.h"
#include "mndTrans.h"
#include "mndUser.h"
......@@ -37,9 +38,9 @@ static SSdbRow *mndStbActionDecode(SSdbRaw *pRaw);
static int32_t mndStbActionInsert(SSdb *pSdb, SStbObj *pStb);
static int32_t mndStbActionDelete(SSdb *pSdb, SStbObj *pStb);
static int32_t mndStbActionUpdate(SSdb *pSdb, SStbObj *pOld, SStbObj *pNew);
static int32_t mndProcessMCreateStbReq(SRpcMsg *pReq);
static int32_t mndProcessMAlterStbReq(SRpcMsg *pReq);
static int32_t mndProcessMDropStbReq(SRpcMsg *pReq);
static int32_t mndProcessCreateStbReq(SRpcMsg *pReq);
static int32_t mndProcessAlterStbReq(SRpcMsg *pReq);
static int32_t mndProcessDropStbReq(SRpcMsg *pReq);
static int32_t mndProcessTableMetaReq(SRpcMsg *pReq);
static int32_t mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
static void mndCancelGetNextStb(SMnode *pMnode, void *pIter);
......@@ -55,9 +56,9 @@ int32_t mndInitStb(SMnode *pMnode) {
.deleteFp = (SdbDeleteFp)mndStbActionDelete,
};
mndSetMsgHandle(pMnode, TDMT_MND_CREATE_STB, mndProcessMCreateStbReq);
mndSetMsgHandle(pMnode, TDMT_MND_ALTER_STB, mndProcessMAlterStbReq);
mndSetMsgHandle(pMnode, TDMT_MND_DROP_STB, mndProcessMDropStbReq);
mndSetMsgHandle(pMnode, TDMT_MND_CREATE_STB, mndProcessCreateStbReq);
mndSetMsgHandle(pMnode, TDMT_MND_ALTER_STB, mndProcessAlterStbReq);
mndSetMsgHandle(pMnode, TDMT_MND_DROP_STB, mndProcessDropStbReq);
mndSetMsgHandle(pMnode, TDMT_VND_CREATE_STB_RSP, mndTransProcessRsp);
mndSetMsgHandle(pMnode, TDMT_VND_ALTER_STB_RSP, mndTransProcessRsp);
mndSetMsgHandle(pMnode, TDMT_VND_DROP_STB_RSP, mndTransProcessRsp);
......@@ -319,6 +320,7 @@ static int32_t mndStbActionUpdate(SSdb *pSdb, SStbObj *pOld, SStbObj *pNew) {
pOld->updateTime = pNew->updateTime;
pOld->tagVer = pNew->tagVer;
pOld->colVer = pNew->colVer;
pOld->smaVer = pNew->smaVer;
pOld->nextColId = pNew->nextColId;
pOld->ttl = pNew->ttl;
pOld->numOfColumns = pNew->numOfColumns;
......@@ -362,7 +364,7 @@ SDbObj *mndAcquireDbByStb(SMnode *pMnode, const char *stbName) {
return mndAcquireDb(pMnode, db);
}
static FORCE_INLINE int schemaExColIdCompare(const void *colId, const void *pSchema) {
static FORCE_INLINE int32_t schemaExColIdCompare(const void *colId, const void *pSchema) {
if (*(col_id_t *)colId < ((SSchema *)pSchema)->colId) {
return -1;
} else if (*(col_id_t *)colId > ((SSchema *)pSchema)->colId) {
......@@ -396,14 +398,14 @@ static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pSt
req.pRSmaParam.xFilesFactor = pStb->xFilesFactor;
req.pRSmaParam.delay = pStb->delay;
if (pStb->ast1Len > 0) {
if (mndConvertRsmaTask(&req.pRSmaParam.qmsg1, &req.pRSmaParam.qmsg1Len, pStb->pAst1, pStb->uid, STREAM_TRIGGER_AT_ONCE, 0,
req.pRSmaParam.xFilesFactor) != TSDB_CODE_SUCCESS) {
if (mndConvertRsmaTask(&req.pRSmaParam.qmsg1, &req.pRSmaParam.qmsg1Len, pStb->pAst1, pStb->uid,
STREAM_TRIGGER_AT_ONCE, 0, req.pRSmaParam.xFilesFactor) != TSDB_CODE_SUCCESS) {
return NULL;
}
}
if (pStb->ast2Len > 0) {
if (mndConvertRsmaTask(&req.pRSmaParam.qmsg2, &req.pRSmaParam.qmsg2Len, pStb->pAst2, pStb->uid, STREAM_TRIGGER_AT_ONCE, 0,
req.pRSmaParam.xFilesFactor) != TSDB_CODE_SUCCESS) {
if (mndConvertRsmaTask(&req.pRSmaParam.qmsg2, &req.pRSmaParam.qmsg2Len, pStb->pAst2, pStb->uid,
STREAM_TRIGGER_AT_ONCE, 0, req.pRSmaParam.xFilesFactor) != TSDB_CODE_SUCCESS) {
return NULL;
}
}
......@@ -762,7 +764,7 @@ int32_t mndAddStbToTrans(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *p
return 0;
}
static int32_t mndProcessMCreateStbReq(SRpcMsg *pReq) {
static int32_t mndProcessCreateStbReq(SRpcMsg *pReq) {
SMnode *pMnode = pReq->info.node;
int32_t code = -1;
SStbObj *pStb = NULL;
......@@ -1301,7 +1303,7 @@ static int32_t mndBuildStbSchema(SMnode *pMnode, const char *dbFName, const char
static int32_t mndBuildSMAlterStbRsp(SDbObj *pDb, const SMAlterStbReq *pAlter, SStbObj *pObj, void **pCont,
int32_t *pLen) {
int ret;
int32_t ret;
SEncoder ec = {0};
uint32_t contLen = 0;
SMAlterStbRsp alterRsp = {0};
......@@ -1420,7 +1422,7 @@ _OVER:
return code;
}
static int32_t mndProcessMAlterStbReq(SRpcMsg *pReq) {
static int32_t mndProcessAlterStbReq(SRpcMsg *pReq) {
SMnode *pMnode = pReq->info.node;
int32_t code = -1;
SDbObj *pDb = NULL;
......@@ -1550,6 +1552,7 @@ static int32_t mndDropStb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *p
if (mndSetDropStbRedoLogs(pMnode, pTrans, pStb) != 0) goto _OVER;
if (mndSetDropStbCommitLogs(pMnode, pTrans, pStb) != 0) goto _OVER;
if (mndSetDropStbRedoActions(pMnode, pTrans, pDb, pStb) != 0) goto _OVER;
if (mndDropSmasByStb(pMnode, pTrans, pDb, pStb) != 0) goto _OVER;
if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
code = 0;
......@@ -1559,7 +1562,7 @@ _OVER:
return code;
}
static int32_t mndProcessMDropStbReq(SRpcMsg *pReq) {
static int32_t mndProcessDropStbReq(SRpcMsg *pReq) {
SMnode *pMnode = pReq->info.node;
int32_t code = -1;
SUserObj *pUser = NULL;
......
source/dnode/mnode/impl/src/mndVgroup.c
浏览文件 @ 7a6f94a8
......@@ -322,6 +322,33 @@ void *mndBuildAlterVnodeReq(SMnode *pMnode, SDbObj *pDb, SVgObj *pVgroup, int32_
return pReq;
}
void *mndBuildSetVnodeStandbyReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen) {
SSetStandbyReq standbyReq = {0};
standbyReq.dnodeId = pDnode->id;
standbyReq.standby = 1;
int32_t contLen = tSerializeSSetStandbyReq(NULL, 0, &standbyReq);
if (contLen < 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
contLen += sizeof(SMsgHead);
void *pReq = taosMemoryMalloc(contLen);
if (pReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
tSerializeSSetStandbyReq((char *)pReq + sizeof(SMsgHead), contLen, &standbyReq);
SMsgHead *pHead = pReq;
pHead->contLen = htonl(contLen);
pHead->vgId = htonl(pVgroup->vgId);
*pContLen = contLen;
return pReq;
}
void *mndBuildDropVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen) {
SDropVnodeReq dropReq = {0};
dropReq.dnodeId = pDnode->id;
......@@ -898,6 +925,39 @@ int32_t mndAddAlterVnodeAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgO
return 0;
}
static int32_t mndAddSetVnodeStandByAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup,
SVnodeGid *pVgid, bool isRedo) {
STransAction action = {0};
SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
if (pDnode == NULL) return -1;
action.epSet = mndGetDnodeEpset(pDnode);
mndReleaseDnode(pMnode, pDnode);
int32_t contLen = 0;
void *pReq = mndBuildSetVnodeStandbyReq(pMnode, pDnode, pDb, pVgroup, &contLen);
if (pReq == NULL) return -1;
action.pCont = pReq;
action.contLen = contLen;
action.msgType = TDMT_DND_DROP_VNODE;
action.acceptableCode = TSDB_CODE_NODE_NOT_DEPLOYED;
if (isRedo) {
if (mndTransAppendRedoAction(pTrans, &action) != 0) {
taosMemoryFree(pReq);
return -1;
}
} else {
if (mndTransAppendUndoAction(pTrans, &action) != 0) {
taosMemoryFree(pReq);
return -1;
}
}
return 0;
}
int32_t mndAddDropVnodeAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, SVnodeGid *pVgid,
bool isRedo) {
STransAction action = {0};
......@@ -952,6 +1012,7 @@ int32_t mndSetMoveVgroupInfoToTrans(SMnode *pMnode, STrans *pTrans, SDbObj *pDb,
SVnodeGid del = newVg.vnodeGid[vnIndex];
newVg.vnodeGid[vnIndex] = newVg.vnodeGid[newVg.replica];
memset(&newVg.vnodeGid[newVg.replica], 0, sizeof(SVnodeGid));
if (mndAddSetVnodeStandByAction(pMnode, pTrans, pDb, pVgroup, &del, true) != 0) return -1;
if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, &newVg, TDMT_VND_ALTER_REPLICA) != 0) return -1;
if (mndAddDropVnodeAction(pMnode, pTrans, pDb, &newVg, &del, true) != 0) return -1;
if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, &newVg) != 0) return -1;
......@@ -1031,6 +1092,7 @@ static int32_t mndAddDecVgroupReplicaFromTrans(SMnode *pMnode, STrans *pTrans, S
memcpy(pGid, &pVgroup->vnodeGid[pVgroup->replica], sizeof(SVnodeGid));
memset(&pVgroup->vnodeGid[pVgroup->replica], 0, sizeof(SVnodeGid));
if (mndAddSetVnodeStandByAction(pMnode, pTrans, pDb, pVgroup, &delGid, true) != 0) return -1;
if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, pVgroup, TDMT_VND_ALTER_REPLICA) != 0) return -1;
if (mndAddDropVnodeAction(pMnode, pTrans, pDb, pVgroup, &delGid, true) != 0) return -1;
if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, pVgroup) != 0) return -1;
......@@ -1341,12 +1403,14 @@ int32_t mndBuildAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, S
SVnodeGid del1 = {0};
if (mndRemoveVnodeFromVgroup(pMnode, &newVgroup, pArray, &del1) != 0) return -1;
if (mndAddSetVnodeStandByAction(pMnode, pTrans, pDb, pVgroup, &del1, true) != 0) return -1;
if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, &newVgroup, TDMT_VND_ALTER_REPLICA) != 0) return -1;
if (mndAddDropVnodeAction(pMnode, pTrans, pDb, &newVgroup, &del1, true) != 0) return -1;
if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, &newVgroup) != 0) return -1;
SVnodeGid del2 = {0};
if (mndRemoveVnodeFromVgroup(pMnode, &newVgroup, pArray, &del2) != 0) return -1;
if (mndAddSetVnodeStandByAction(pMnode, pTrans, pDb, pVgroup, &del2, true) != 0) return -1;
if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, &newVgroup, TDMT_VND_ALTER_REPLICA) != 0) return -1;
if (mndAddDropVnodeAction(pMnode, pTrans, pDb, &newVgroup, &del2, true) != 0) return -1;
if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, &newVgroup) != 0) return -1;
......@@ -1396,6 +1460,7 @@ static int32_t mndSplitVgroup(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SVgObj
} else if (newVg1.replica == 3) {
SVnodeGid del1 = {0};
if (mndRemoveVnodeFromVgroup(pMnode, &newVg1, pArray, &del1) != 0) goto _OVER;
if (mndAddSetVnodeStandByAction(pMnode, pTrans, pDb, pVgroup, &del1, true) != 0) return -1;
if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, &newVg1, TDMT_VND_ALTER_REPLICA) != 0) goto _OVER;
if (mndAddDropVnodeAction(pMnode, pTrans, pDb, &newVg1, &del1, true) != 0) goto _OVER;
if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, &newVg1) != 0) goto _OVER;
......
source/dnode/vnode/CMakeLists.txt
浏览文件 @ 7a6f94a8
......@@ -31,7 +31,7 @@ target_sources(
"src/sma/smaEnv.c"
"src/sma/smaOpen.c"
"src/sma/smaRollup.c"
"src/sma/smaTimeRange2.c"
"src/sma/smaTimeRange.c"
# tsdb
"src/tsdb/tsdbCommit.c"
......
source/dnode/vnode/src/inc/sma.h
浏览文件 @ 7a6f94a8
......@@ -38,8 +38,6 @@ typedef struct SSmaStatItem SSmaStatItem;
typedef struct SSmaKey SSmaKey;
typedef struct SRSmaInfo SRSmaInfo;
#define SMA_IVLD_FID INT_MIN
struct SSmaEnv {
TdThreadRwlock lock;
int8_t type;
......@@ -49,45 +47,38 @@ struct SSmaEnv {
#define SMA_ENV_LOCK(env) ((env)->lock)
#define SMA_ENV_TYPE(env) ((env)->type)
#define SMA_ENV_STAT(env) ((env)->pStat)
#define SMA_ENV_STAT_ITEMS(env) ((env)->pStat->smaStatItems)
#define SMA_ENV_STAT_ITEM(env) ((env)->pStat->tsmaStatItem)
struct SSmaStatItem {
int8_t state; // ETsdbSmaStat
STSma *pTSma; // cache schema
int8_t state; // ETsdbSmaStat
STSma *pTSma; // cache schema
STSchema *pTSchema;
};
struct SSmaStat {
union {
SHashObj *smaStatItems; // key: indexUid, value: SSmaStatItem for tsma
SHashObj *rsmaInfoHash; // key: stbUid, value: SRSmaInfo;
SSmaStatItem tsmaStatItem;
SHashObj *rsmaInfoHash; // key: stbUid, value: SRSmaInfo;
};
T_REF_DECLARE()
};
#define SMA_STAT_ITEMS(s) ((s)->smaStatItems)
#define SMA_STAT_ITEM(s) ((s)->tsmaStatItem)
#define SMA_STAT_INFO_HASH(s) ((s)->rsmaInfoHash)
void tdDestroySmaEnv(SSmaEnv *pSmaEnv);
void *tdFreeSmaEnv(SSmaEnv *pSmaEnv);
#if 0
int32_t tbGetTSmaStatus(SSma *pSma, STSma *param, void *result);
int32_t tbRemoveTSmaData(SSma *pSma, STSma *param, STimeWindow *pWin);
#endif
int32_t tdInitSma(SSma *pSma);
int32_t tdDropTSma(SSma *pSma, char *pMsg);
int32_t tdDropTSmaData(SSma *pSma, int64_t indexUid);
int32_t tdInsertRSmaData(SSma *pSma, char *msg);
int32_t tdRefSmaStat(SSma *pSma, SSmaStat *pStat);
int32_t tdUnRefSmaStat(SSma *pSma, SSmaStat *pStat);
int32_t tdCheckAndInitSmaEnv(SSma *pSma, int8_t smaType, bool onlyCheck);
int32_t tdCheckAndInitSmaEnv(SSma *pSma, int8_t smaType);
int32_t tdLockSma(SSma *pSma);
int32_t tdUnLockSma(SSma *pSma);
static FORCE_INLINE int16_t tdTSmaAdd(SSma *pSma, int16_t n) { return atomic_add_fetch_16(&SMA_TSMA_NUM(pSma), n); }
static FORCE_INLINE int16_t tdTSmaSub(SSma *pSma, int16_t n) { return atomic_sub_fetch_16(&SMA_TSMA_NUM(pSma), n); }
static FORCE_INLINE int32_t tdRLockSmaEnv(SSmaEnv *pEnv) {
int code = taosThreadRwlockRdlock(&(pEnv->lock));
if (code != 0) {
......@@ -160,11 +151,10 @@ static FORCE_INLINE void tdSmaStatSetDropped(SSmaStatItem *pStatItem) {
}
}
static int32_t tdInitSmaStat(SSmaStat **pSmaStat, int8_t smaType);
void *tdFreeSmaStatItem(SSmaStatItem *pSmaStatItem);
static int32_t tdDestroySmaState(SSmaStat *pSmaStat, int8_t smaType);
static SSmaEnv *tdNewSmaEnv(const SSma *pSma, int8_t smaType, const char *path, SDiskID did);
static int32_t tdInitSmaEnv(SSma *pSma, int8_t smaType, const char *path, SDiskID did, SSmaEnv **pEnv);
static int32_t tdInitSmaStat(SSmaStat **pSmaStat, int8_t smaType);
void *tdFreeSmaStatItem(SSmaStatItem *pSmaStatItem);
static int32_t tdDestroySmaState(SSmaStat *pSmaStat, int8_t smaType);
void *tdFreeSmaState(SSmaStat *pSmaStat, int8_t smaType);
void *tdFreeRSmaInfo(SRSmaInfo *pInfo);
......
source/dnode/vnode/src/inc/vnodeInt.h
浏览文件 @ 7a6f94a8
......@@ -147,6 +147,9 @@ int32_t tqProcessTaskRecoverReq(STQ* pTq, SRpcMsg* pMsg);
int32_t tqProcessTaskDispatchRsp(STQ* pTq, SRpcMsg* pMsg);
int32_t tqProcessTaskRecoverRsp(STQ* pTq, SRpcMsg* pMsg);
SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pSchema, bool createTb, int64_t suid,
const char* stbFullName, int32_t vgId);
// sma
int32_t smaOpen(SVnode* pVnode);
int32_t smaClose(SSma* pSma);
......@@ -245,7 +248,6 @@ struct STbUidStore {
};
struct SSma {
int16_t nTSma;
bool locked;
TdThreadMutex mutex;
SVnode* pVnode;
......@@ -261,7 +263,6 @@ struct SSma {
#define SMA_META(s) ((s)->pVnode->pMeta)
#define SMA_VID(s) TD_VID((s)->pVnode)
#define SMA_TFS(s) ((s)->pVnode->pTfs)
#define SMA_TSMA_NUM(s) ((s)->nTSma)
#define SMA_TSMA_ENV(s) ((s)->pTSmaEnv)
#define SMA_RSMA_ENV(s) ((s)->pRSmaEnv)
#define SMA_RSMA_TSDB0(s) ((s)->pVnode->pTsdb)
......
source/dnode/vnode/src/meta/metaEntry.c
浏览文件 @ 7a6f94a8
......@@ -75,7 +75,7 @@ int metaDecodeEntry(SDecoder *pCoder, SMetaEntry *pME) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
if (tDecodeTSma(pCoder, pME->smaEntry.tsma) < 0) return -1;
if (tDecodeTSma(pCoder, pME->smaEntry.tsma, true) < 0) return -1;
} else {
ASSERT(0);
}
......
source/dnode/vnode/src/sma/sma.c
浏览文件 @ 7a6f94a8
......@@ -44,3 +44,209 @@ int32_t smaGetTSmaDays(SVnodeCfg* pCfg, void* pCont, uint32_t contLen, int32_t*
smaDebug("vgId:%d, get tsma days %d", pCfg->vgId, *days);
return code;
}
#if 0
/**
* @brief TODO: Assume that the final generated result it less than 3M
*
* @param pReq
* @param pDataBlocks
* @param vgId
* @param suid // TODO: check with Liao whether suid response is reasonable
*
* TODO: colId should be set
*/
int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks, STSchema* pTSchema, int32_t vgId,
tb_uid_t suid, const char* stbName, bool isCreateCtb) {
int32_t sz = taosArrayGetSize(pDataBlocks);
int32_t bufSize = sizeof(SSubmitReq);
for (int32_t i = 0; i < sz; ++i) {
SDataBlockInfo* pBlkInfo = &((SSDataBlock*)taosArrayGet(pDataBlocks, i))->info;
bufSize += pBlkInfo->rows * (TD_ROW_HEAD_LEN + pBlkInfo->rowSize + BitmapLen(pBlkInfo->numOfCols));
bufSize += sizeof(SSubmitBlk);
}
*pReq = taosMemoryCalloc(1, bufSize);
if (!(*pReq)) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return TSDB_CODE_FAILED;
}
void* pDataBuf = *pReq;
SArray* pTagArray = NULL;
int32_t msgLen = sizeof(SSubmitReq);
int32_t numOfBlks = 0;
int32_t schemaLen = 0;
SRowBuilder rb = {0};
tdSRowInit(&rb, pTSchema->version);
for (int32_t i = 0; i < sz; ++i) {
SSDataBlock* pDataBlock = taosArrayGet(pDataBlocks, i);
SDataBlockInfo* pDataBlkInfo = &pDataBlock->info;
int32_t colNum = pDataBlkInfo->numOfCols;
int32_t rows = pDataBlkInfo->rows;
int32_t rowSize = pDataBlkInfo->rowSize;
int64_t groupId = pDataBlkInfo->groupId;
if (rb.nCols != colNum) {
tdSRowSetTpInfo(&rb, colNum, pTSchema->flen);
}
if(isCreateCtb) {
SMetaReader mr = {0};
const char* ctbName = buildCtbNameByGroupId(stbName, pDataBlock->info.groupId);
if (metaGetTableEntryByName(&mr, ctbName) != 0) {
smaDebug("vgId:%d, no tsma ctb %s exists", vgId, ctbName);
}
SVCreateTbReq ctbReq = {0};
ctbReq.name = ctbName;
ctbReq.type = TSDB_CHILD_TABLE;
ctbReq.ctb.suid = suid;
STagVal tagVal = {.cid = colNum + PRIMARYKEY_TIMESTAMP_COL_ID,
.type = TSDB_DATA_TYPE_BIGINT,
.i64 = groupId};
STag* pTag = NULL;
if(!pTagArray) {
pTagArray = taosArrayInit(1, sizeof(STagVal));
if (!pTagArray) goto _err;
}
taosArrayClear(pTagArray);
taosArrayPush(pTagArray, &tagVal);
tTagNew(pTagArray, 1, false, &pTag);
if (pTag == NULL) {
tdDestroySVCreateTbReq(&ctbReq);
goto _err;
}
ctbReq.ctb.pTag = (uint8_t*)pTag;
int32_t code;
tEncodeSize(tEncodeSVCreateTbReq, &ctbReq, schemaLen, code);
tdDestroySVCreateTbReq(&ctbReq);
if (code < 0) {
goto _err;
}
}
SSubmitBlk* pSubmitBlk = POINTER_SHIFT(pDataBuf, msgLen);
pSubmitBlk->suid = suid;
pSubmitBlk->uid = groupId;
pSubmitBlk->numOfRows = rows;
msgLen += sizeof(SSubmitBlk);
int32_t dataLen = 0;
for (int32_t j = 0; j < rows; ++j) { // iterate by row
tdSRowResetBuf(&rb, POINTER_SHIFT(pDataBuf, msgLen)); // set row buf
bool isStartKey = false;
int32_t offset = 0;
for (int32_t k = 0; k < colNum; ++k) { // iterate by column
SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, k);
STColumn* pCol = &pTSchema->columns[k];
void* var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);
switch (pColInfoData->info.type) {
case TSDB_DATA_TYPE_TIMESTAMP:
if (!isStartKey) {
isStartKey = true;
tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID, TSDB_DATA_TYPE_TIMESTAMP, TD_VTYPE_NORM, var, true,
offset, k);
} else {
tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, TSDB_DATA_TYPE_TIMESTAMP, TD_VTYPE_NORM, var,
true, offset, k);
}
break;
case TSDB_DATA_TYPE_NCHAR: {
tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, TSDB_DATA_TYPE_NCHAR, TD_VTYPE_NORM, var, true,
offset, k);
break;
}
case TSDB_DATA_TYPE_VARCHAR: { // TSDB_DATA_TYPE_BINARY
tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, TSDB_DATA_TYPE_VARCHAR, TD_VTYPE_NORM, var, true,
offset, k);
break;
}
case TSDB_DATA_TYPE_VARBINARY:
case TSDB_DATA_TYPE_DECIMAL:
case TSDB_DATA_TYPE_BLOB:
case TSDB_DATA_TYPE_JSON:
case TSDB_DATA_TYPE_MEDIUMBLOB:
uError("the column type %" PRIi16 " is defined but not implemented yet", pColInfoData->info.type);
TASSERT(0);
break;
default:
if (pColInfoData->info.type < TSDB_DATA_TYPE_MAX && pColInfoData->info.type > TSDB_DATA_TYPE_NULL) {
if (pCol->type == pColInfoData->info.type) {
tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, pCol->type, TD_VTYPE_NORM, var, true, offset,
k);
} else {
char tv[8] = {0};
if (pColInfoData->info.type == TSDB_DATA_TYPE_FLOAT) {
float v = 0;
GET_TYPED_DATA(v, float, pColInfoData->info.type, var);
SET_TYPED_DATA(&tv, pCol->type, v);
} else if (pColInfoData->info.type == TSDB_DATA_TYPE_DOUBLE) {
double v = 0;
GET_TYPED_DATA(v, double, pColInfoData->info.type, var);
SET_TYPED_DATA(&tv, pCol->type, v);
} else if (IS_SIGNED_NUMERIC_TYPE(pColInfoData->info.type)) {
int64_t v = 0;
GET_TYPED_DATA(v, int64_t, pColInfoData->info.type, var);
SET_TYPED_DATA(&tv, pCol->type, v);
} else {
uint64_t v = 0;
GET_TYPED_DATA(v, uint64_t, pColInfoData->info.type, var);
SET_TYPED_DATA(&tv, pCol->type, v);
}
tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, pCol->type, TD_VTYPE_NORM, tv, true, offset,
k);
}
} else {
uError("the column type %" PRIi16 " is undefined\n", pColInfoData->info.type);
TASSERT(0);
}
break;
}
offset += TYPE_BYTES[pCol->type]; // sum/avg would convert to int64_t/uint64_t/double during aggregation
}
dataLen += TD_ROW_LEN(rb.pBuf);
#ifdef TD_DEBUG_PRINT_ROW
tdSRowPrint(rb.pBuf, pTSchema, __func__);
#endif
}
++numOfBlks;
pSubmitBlk->dataLen = dataLen;
msgLen += pSubmitBlk->dataLen;
}
(*pReq)->length = msgLen;
(*pReq)->header.vgId = htonl(vgId);
(*pReq)->header.contLen = htonl(msgLen);
(*pReq)->length = (*pReq)->header.contLen;
(*pReq)->numOfBlocks = htonl(numOfBlks);
SSubmitBlk* blk = (SSubmitBlk*)((*pReq) + 1);
while (numOfBlks--) {
int32_t dataLen = blk->dataLen;
blk->uid = htobe64(blk->uid);
blk->suid = htobe64(blk->suid);
blk->padding = htonl(blk->padding);
blk->sversion = htonl(blk->sversion);
blk->dataLen = htonl(blk->dataLen);
blk->schemaLen = htonl(blk->schemaLen);
blk->numOfRows = htons(blk->numOfRows);
blk = (SSubmitBlk*)(blk->data + dataLen);
}
return TSDB_CODE_SUCCESS;
_err:
taosMemoryFreeClear(*pReq);
taosArrayDestroy(pTagArray);
return TSDB_CODE_FAILED;
}
#endif
source/dnode/vnode/src/sma/smaEnv.c
浏览文件 @ 7a6f94a8
......@@ -17,123 +17,17 @@
typedef struct SSmaStat SSmaStat;
static const char *TSDB_SMA_DNAME[] = {
"", // TSDB_SMA_TYPE_BLOCK
"tsma", // TSDB_SMA_TYPE_TIME_RANGE
"rsma", // TSDB_SMA_TYPE_ROLLUP
};
#define SMA_TEST_INDEX_NAME "smaTestIndexName" // TODO: just for test
#define SMA_TEST_INDEX_UID 2000000001 // TODO: just for test
#define SMA_STATE_HASH_SLOT 4
#define RSMA_TASK_INFO_HASH_SLOT 8
typedef struct SPoolMem {
int64_t size;
struct SPoolMem *prev;
struct SPoolMem *next;
} SPoolMem;
// declaration of static functions
// insert data
static void tdGetSmaDir(int32_t vgId, ETsdbSmaType smaType, char dirName[]);
// Pool Memory
static SPoolMem *openPool();
static void clearPool(SPoolMem *pPool);
static void closePool(SPoolMem *pPool);
static void *poolMalloc(void *arg, size_t size);
static void poolFree(void *arg, void *ptr);
static int32_t tdInitSmaStat(SSmaStat **pSmaStat, int8_t smaType);
static SSmaEnv *tdNewSmaEnv(const SSma *pSma, int8_t smaType, const char *path);
static int32_t tdInitSmaEnv(SSma *pSma, int8_t smaType, const char *path, SSmaEnv **pEnv);
// implementation
static SPoolMem *openPool() {
SPoolMem *pPool = (SPoolMem *)taosMemoryMalloc(sizeof(*pPool));
pPool->prev = pPool->next = pPool;
pPool->size = 0;
return pPool;
}
static void clearPool(SPoolMem *pPool) {
if (!pPool) return;
SPoolMem *pMem;
do {
pMem = pPool->next;
if (pMem == pPool) break;
pMem->next->prev = pMem->prev;
pMem->prev->next = pMem->next;
pPool->size -= pMem->size;
taosMemoryFree(pMem);
} while (1);
assert(pPool->size == 0);
}
static void closePool(SPoolMem *pPool) {
if (pPool) {
clearPool(pPool);
taosMemoryFree(pPool);
}
}
static void *poolMalloc(void *arg, size_t size) {
void *ptr = NULL;
SPoolMem *pPool = (SPoolMem *)arg;
SPoolMem *pMem;
pMem = (SPoolMem *)taosMemoryMalloc(sizeof(*pMem) + size);
if (!pMem) {
assert(0);
}
pMem->size = sizeof(*pMem) + size;
pMem->next = pPool->next;
pMem->prev = pPool;
pPool->next->prev = pMem;
pPool->next = pMem;
pPool->size += pMem->size;
ptr = (void *)(&pMem[1]);
return ptr;
}
static void poolFree(void *arg, void *ptr) {
SPoolMem *pPool = (SPoolMem *)arg;
SPoolMem *pMem;
pMem = &(((SPoolMem *)ptr)[-1]);
pMem->next->prev = pMem->prev;
pMem->prev->next = pMem->next;
pPool->size -= pMem->size;
taosMemoryFree(pMem);
}
int32_t tdInitSma(SSma *pSma) {
int32_t numOfTSma = taosArrayGetSize(metaGetSmaTbUids(SMA_META(pSma)));
if (numOfTSma > 0) {
atomic_store_16(&SMA_TSMA_NUM(pSma), (int16_t)numOfTSma);
}
return TSDB_CODE_SUCCESS;
}
static void tdGetSmaDir(int32_t vgId, ETsdbSmaType smaType, char dirName[]) {
snprintf(dirName, TSDB_FILENAME_LEN, "vnode%svnode%d%s%s", TD_DIRSEP, vgId, TD_DIRSEP, TSDB_SMA_DNAME[smaType]);
}
static SSmaEnv *tdNewSmaEnv(const SSma *pSma, int8_t smaType, const char *path, SDiskID did) {
static SSmaEnv *tdNewSmaEnv(const SSma *pSma, int8_t smaType, const char *path) {
SSmaEnv *pEnv = NULL;
pEnv = (SSmaEnv *)taosMemoryCalloc(1, sizeof(SSmaEnv));
......@@ -156,18 +50,17 @@ static SSmaEnv *tdNewSmaEnv(const SSma *pSma, int8_t smaType, const char *path,
return NULL;
}
return pEnv;
}
static int32_t tdInitSmaEnv(SSma *pSma, int8_t smaType, const char *path, SDiskID did, SSmaEnv **pEnv) {
static int32_t tdInitSmaEnv(SSma *pSma, int8_t smaType, const char *path, SSmaEnv **pEnv) {
if (!pEnv) {
terrno = TSDB_CODE_INVALID_PTR;
return TSDB_CODE_FAILED;
}
if (!(*pEnv)) {
if (!(*pEnv = tdNewSmaEnv(pSma, smaType, path, did))) {
if (!(*pEnv = tdNewSmaEnv(pSma, smaType, path))) {
return TSDB_CODE_FAILED;
}
}
......@@ -183,15 +76,16 @@ static int32_t tdInitSmaEnv(SSma *pSma, int8_t smaType, const char *path, SDiskI
*/
void tdDestroySmaEnv(SSmaEnv *pSmaEnv) {
if (pSmaEnv) {
tdDestroySmaState(pSmaEnv->pStat, SMA_ENV_TYPE(pSmaEnv));
taosMemoryFreeClear(pSmaEnv->pStat);
pSmaEnv->pStat = tdFreeSmaState(pSmaEnv->pStat, SMA_ENV_TYPE(pSmaEnv));
taosThreadRwlockDestroy(&(pSmaEnv->lock));
}
}
void *tdFreeSmaEnv(SSmaEnv *pSmaEnv) {
tdDestroySmaEnv(pSmaEnv);
taosMemoryFreeClear(pSmaEnv);
if (pSmaEnv) {
tdDestroySmaEnv(pSmaEnv);
taosMemoryFreeClear(pSmaEnv);
}
return NULL;
}
......@@ -239,13 +133,7 @@ static int32_t tdInitSmaStat(SSmaStat **pSmaStat, int8_t smaType) {
return TSDB_CODE_FAILED;
}
} else if (smaType == TSDB_SMA_TYPE_TIME_RANGE) {
SMA_STAT_ITEMS(*pSmaStat) =
taosHashInit(SMA_STATE_HASH_SLOT, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
if (!SMA_STAT_ITEMS(*pSmaStat)) {
taosMemoryFreeClear(*pSmaStat);
return TSDB_CODE_FAILED;
}
// TODO
} else {
ASSERT(0);
}
......@@ -262,6 +150,12 @@ void *tdFreeSmaStatItem(SSmaStatItem *pSmaStatItem) {
return NULL;
}
void* tdFreeSmaState(SSmaStat *pSmaStat, int8_t smaType) {
tdDestroySmaState(pSmaStat, smaType);
taosMemoryFreeClear(pSmaStat);
return NULL;
}
/**
* @brief Release resources allocated for its member fields, not including itself.
*
......@@ -270,16 +164,10 @@ void *tdFreeSmaStatItem(SSmaStatItem *pSmaStatItem) {
*/
int32_t tdDestroySmaState(SSmaStat *pSmaStat, int8_t smaType) {
if (pSmaStat) {
// TODO: use taosHashSetFreeFp when taosHashSetFreeFp is ready.
if (smaType == TSDB_SMA_TYPE_TIME_RANGE) {
void *item = taosHashIterate(SMA_STAT_ITEMS(pSmaStat), NULL);
while (item) {
SSmaStatItem *pItem = *(SSmaStatItem **)item;
tdFreeSmaStatItem(pItem);
item = taosHashIterate(SMA_STAT_ITEMS(pSmaStat), item);
}
taosHashCleanup(SMA_STAT_ITEMS(pSmaStat));
tdFreeSmaStatItem(&pSmaStat->tsmaStatItem);
} else if (smaType == TSDB_SMA_TYPE_ROLLUP) {
// TODO: use taosHashSetFreeFp when taosHashSetFreeFp is ready.
void *infoHash = taosHashIterate(SMA_STAT_INFO_HASH(pSmaStat), NULL);
while (infoHash) {
SRSmaInfo *pInfoHash = *(SRSmaInfo **)infoHash;
......@@ -317,7 +205,7 @@ int32_t tdUnLockSma(SSma *pSma) {
return 0;
}
int32_t tdCheckAndInitSmaEnv(SSma *pSma, int8_t smaType, bool onlyCheck) {
int32_t tdCheckAndInitSmaEnv(SSma *pSma, int8_t smaType) {
SSmaEnv *pEnv = NULL;
// return if already init
......@@ -344,26 +232,7 @@ int32_t tdCheckAndInitSmaEnv(SSma *pSma, int8_t smaType, bool onlyCheck) {
if (!pEnv) {
char rname[TSDB_FILENAME_LEN] = {0};
SDiskID did = {0};
if (tfsAllocDisk(SMA_TFS(pSma), TFS_PRIMARY_LEVEL, &did) < 0) {
tdUnLockSma(pSma);
return TSDB_CODE_FAILED;
}
if (did.level < 0 || did.id < 0) {
tdUnLockSma(pSma);
smaError("vgId:%d, init sma env failed since invalid did(%d,%d)", SMA_VID(pSma), did.level, did.id);
return TSDB_CODE_FAILED;
}
tdGetSmaDir(SMA_VID(pSma), smaType, rname);
if (tfsMkdirRecurAt(SMA_TFS(pSma), rname, did) < 0) {
tdUnLockSma(pSma);
return TSDB_CODE_FAILED;
}
if (tdInitSmaEnv(pSma, smaType, rname, did, &pEnv) < 0) {
if (tdInitSmaEnv(pSma, smaType, rname, &pEnv) < 0) {
tdUnLockSma(pSma);
return TSDB_CODE_FAILED;
}
......
source/dnode/vnode/src/sma/smaOpen.c
浏览文件 @ 7a6f94a8
......@@ -132,7 +132,9 @@ int32_t smaClose(SSma *pSma) {
if SMA_RSMA_TSDB0 (pSma) tsdbClose(&SMA_RSMA_TSDB0(pSma));
if SMA_RSMA_TSDB1 (pSma) tsdbClose(&SMA_RSMA_TSDB1(pSma));
if SMA_RSMA_TSDB2 (pSma) tsdbClose(&SMA_RSMA_TSDB2(pSma));
taosMemoryFree(pSma);
// SMA_TSMA_ENV(pSma) = tdFreeSmaEnv(SMA_TSMA_ENV(pSma));
// SMA_RSMA_ENV(pSma) = tdFreeSmaEnv(SMA_RSMA_ENV(pSma));
taosMemoryFreeClear(pSma);
}
return 0;
}
\ No newline at end of file
source/dnode/vnode/src/sma/smaRollup.c
浏览文件 @ 7a6f94a8
......@@ -181,7 +181,7 @@ int32_t tdProcessRSmaCreate(SVnode *pVnode, SVCreateStbReq *pReq) {
return TSDB_CODE_SUCCESS;
}
if (tdCheckAndInitSmaEnv(pSma, TSDB_SMA_TYPE_ROLLUP, false) != TSDB_CODE_SUCCESS) {
if (tdCheckAndInitSmaEnv(pSma, TSDB_SMA_TYPE_ROLLUP) != TSDB_CODE_SUCCESS) {
terrno = TSDB_CODE_TDB_INIT_FAILED;
return TSDB_CODE_FAILED;
}
......
source/dnode/vnode/src/sma/smaTimeRange.c
浏览文件 @ 7a6f94a8
......@@ -142,7 +142,6 @@ int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char *msg) {
ASSERT(pItem);
if (!pItem->pTSma) {
// cache smaMeta
STSma *pTSma = metaGetSmaInfoByIndex(SMA_META(pSma), indexUid);
if (!pTSma) {
terrno = TSDB_CODE_TSMA_NO_INDEX_IN_META;
......@@ -150,27 +149,28 @@ int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char *msg) {
return TSDB_CODE_FAILED;
}
pItem->pTSma = pTSma;
pItem->pTSchema = metaGetTbTSchema(SMA_META(pSma), pTSma->dstTbUid, -1);
ASSERT(pItem->pTSchema); // TODO
}
STSma *pTSma = pItem->pTSma;
ASSERT(pItem->pTSma->indexUid == indexUid);
ASSERT(pTSma->indexUid == indexUid);
SSubmitReq *pSubmitReq = NULL;
SMetaReader mr = {0};
pSubmitReq = tdBlockToSubmit((const SArray *)msg, pItem->pTSchema, true, pItem->pTSma->dstTbUid,
pItem->pTSma->dstTbName, pItem->pTSma->dstVgId);
const char *dbName = "testDb";
if (metaGetTableEntryByName(&mr, dbName) != 0) {
smaDebug("vgId:%d, tsma no table testTb exists for smaIndex %" PRIi64 " since %s", SMA_VID(pSma), indexUid, tstrerror(terrno));
SVCreateStbReq pReq = {0};
pReq.name = dbName;
pReq.suid = pTSma->dstTbUid;
pReq.schemaRow = pCfg->schemaRow;
pReq.schemaTag = pCfg->schemaTag;
}
ASSERT(pSubmitReq); // TODO
SSubmitReq *pSubmitReq = NULL;
buildSubmitReqFromDataBlock(&pSubmitReq, (const SArray *)msg, NULL, pItem->pTSma->dstVgId,
pItem->pTSma->dstTbUid);
ASSERT(!strncasecmp("td.tsma.rst.tb", pItem->pTSma->dstTbName, 14));
SRpcMsg submitReqMsg = {
.msgType = TDMT_VND_SUBMIT,
.pCont = pSubmitReq,
.contLen = ntohl(pSubmitReq->length),
};
ASSERT(tmsgPutToQueue(&pSma->pVnode->msgCb, WRITE_QUEUE, &submitReqMsg) == 0);
tdUnRefSmaStat(pSma, pStat);
......
source/dnode/vnode/src/sma/smaTimeRange2.c 已删除 100644 → 0
浏览文件 @ 6da9abd8
/*
* 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 "sma.h"
#include "tsdb.h"
typedef STsdbCfg STSmaKeepCfg;
#undef _TEST_SMA_PRINT_DEBUG_LOG_
#define SMA_STORAGE_MINUTES_MAX 86400
#define SMA_STORAGE_MINUTES_DAY 1440
#define SMA_STORAGE_MINUTES_MIN 1440
#define SMA_STORAGE_TSDB_MINUTES 86400
#define SMA_STORAGE_TSDB_TIMES 10
#define SMA_STORAGE_SPLIT_FACTOR 14400 // least records in tsma file TODO: the feasible value?
#define SMA_KEY_LEN 16 // TSKEY+groupId 8+8
#define SMA_DROP_EXPIRED_TIME 10 // default is 10 seconds
#define SMA_STATE_ITEM_HASH_SLOT 32
// static func
/**
* @brief Judge the tsma file split days
*
* @param pCfg
* @param pCont
* @param contLen
* @param days unit is minute
* @return int32_t
*/
int32_t tdProcessTSmaGetDaysImpl(SVnodeCfg *pCfg, void *pCont, uint32_t contLen, int32_t *days) {
SDecoder coder = {0};
tDecoderInit(&coder, pCont, contLen);
STSma tsma = {0};
if (tDecodeSVCreateTSmaReq(&coder, &tsma) < 0) {
terrno = TSDB_CODE_MSG_DECODE_ERROR;
goto _err;
}
STsdbCfg *pTsdbCfg = &pCfg->tsdbCfg;
int64_t sInterval = convertTimeFromPrecisionToUnit(tsma.interval, pTsdbCfg->precision, TIME_UNIT_SECOND);
if (sInterval <= 0) {
*days = pTsdbCfg->days;
return 0;
}
int64_t records = pTsdbCfg->days * 60 / sInterval;
if (records >= SMA_STORAGE_SPLIT_FACTOR) {
*days = pTsdbCfg->days;
} else {
int64_t mInterval = convertTimeFromPrecisionToUnit(tsma.interval, pTsdbCfg->precision, TIME_UNIT_MINUTE);
int64_t daysPerFile = mInterval * SMA_STORAGE_MINUTES_DAY * 2;
if (daysPerFile > SMA_STORAGE_MINUTES_MAX) {
*days = SMA_STORAGE_MINUTES_MAX;
} else {
*days = (int32_t)daysPerFile;
}
if (*days < pTsdbCfg->days) {
*days = pTsdbCfg->days;
}
}
tDecoderClear(&coder);
return 0;
_err:
tDecoderClear(&coder);
return -1;
}
// read data
// implementation
/**
* @brief Insert/Update Time-range-wise SMA data.
* - If interval < SMA_STORAGE_SPLIT_HOURS(e.g. 24), save the SMA data as a part of DFileSet to e.g.
* v3f1900.tsma.${sma_index_name}. The days is the same with that for TS data files.
* - If interval >= SMA_STORAGE_SPLIT_HOURS, save the SMA data to e.g. vnode3/tsma/v3f632.tsma.${sma_index_name}. The
* days is 30 times of the interval, and the minimum days is SMA_STORAGE_TSDB_DAYS(30d).
* - The destination file of one data block for some interval is determined by its start TS key.
*
* @param pSma
* @param msg
* @return int32_t
*/
int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char *msg) {
STsdbCfg *pCfg = SMA_TSDB_CFG(pSma);
const SArray *pDataBlocks = (const SArray *)msg;
// TODO: destroy SSDataBlocks(msg)
// For super table aggregation, the sma data is stored in vgroup calculated from the hash value of stable name. Thus
// the sma data would arrive ahead of the update-expired-window msg.
if (tdCheckAndInitSmaEnv(pSma, TSDB_SMA_TYPE_TIME_RANGE, false) != TSDB_CODE_SUCCESS) {
terrno = TSDB_CODE_TDB_INIT_FAILED;
return TSDB_CODE_FAILED;
}
if (!pDataBlocks) {
terrno = TSDB_CODE_INVALID_PTR;
smaWarn("vgId:%d, insert tsma data failed since pDataBlocks is NULL", SMA_VID(pSma));
return terrno;
}
if (taosArrayGetSize(pDataBlocks) <= 0) {
terrno = TSDB_CODE_INVALID_PARA;
smaWarn("vgId:%d, insert tsma data failed since pDataBlocks is empty", SMA_VID(pSma));
return TSDB_CODE_FAILED;
}
SSmaEnv *pEnv = SMA_TSMA_ENV(pSma);
SSmaStat *pStat = SMA_ENV_STAT(pEnv);
SSmaStatItem *pItem = NULL;
tdRefSmaStat(pSma, pStat);
if (pStat && SMA_STAT_ITEMS(pStat)) {
pItem = taosHashGet(SMA_STAT_ITEMS(pStat), &indexUid, sizeof(indexUid));
}
if (!pItem || !(pItem = *(SSmaStatItem **)pItem) || tdSmaStatIsDropped(pItem)) {
terrno = TSDB_CODE_TSMA_INVALID_STAT;
tdUnRefSmaStat(pSma, pStat);
return TSDB_CODE_FAILED;
}
STSma *pTSma = pItem->pTSma;
tdUnRefSmaStat(pSma, pStat);
return TSDB_CODE_SUCCESS;
}
int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t version, const char *pMsg) {
SSmaCfg *pCfg = (SSmaCfg *)pMsg;
if (metaCreateTSma(SMA_META(pSma), version, pCfg) < 0) {
return -1;
}
if (TD_VID(pSma->pVnode) == pCfg->dstVgId) {
// create stable to save tsma result in dstVgId
SVCreateStbReq pReq = {0};
pReq.name = pCfg->dstTbName;
pReq.suid = pCfg->dstTbUid;
pReq.schemaRow = pCfg->schemaRow;
pReq.schemaTag = pCfg->schemaTag;
if (metaCreateSTable(SMA_META(pSma), version, &pReq) < 0) {
return -1;
}
}
tdTSmaAdd(pSma, 1);
return 0;
}
\ No newline at end of file
source/dnode/vnode/src/tq/tqSink.c
浏览文件 @ 7a6f94a8
......@@ -15,10 +15,7 @@
#include "tq.h"
static SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pSchema, bool createTb, int64_t suid,
const char* stbFullName, int32_t vgId);
static SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, bool createTb, int64_t suid,
SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, bool createTb, int64_t suid,
const char* stbFullName, int32_t vgId) {
SSubmitReq* ret = NULL;
SArray* tagArray = taosArrayInit(1, sizeof(STagVal));
......
source/dnode/vnode/src/vnd/vnodeSvr.c
浏览文件 @ 7a6f94a8
......@@ -284,7 +284,7 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, int64_t version, SRpcMsg *pMsg, SRp
void smaHandleRes(void *pVnode, int64_t smaId, const SArray *data) {
// TODO
// blockDebugShowData(data, __func__);
blockDebugShowData(data, __func__);
tdProcessTSmaInsert(((SVnode *)pVnode)->pSma, smaId, (const char *)data);
}
......@@ -375,6 +375,10 @@ int32_t vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
ret = syncNodeOnAppendEntriesReplyCb(pSyncNode, pSyncMsg);
syncAppendEntriesReplyDestroy(pSyncMsg);
} else if (pRpcMsg->msgType == TDMT_SYNC_SET_MNODE_STANDBY) {
ret = syncSetStandby(pVnode->sync);
SRpcMsg rsp = {.code = ret, .info = pMsg->info};
tmsgSendRsp(&rsp);
} else {
vError("==vnodeProcessSyncReq== error msg type:%d", pRpcMsg->msgType);
ret = TAOS_SYNC_OTHER_ERROR;
......
source/dnode/vnode/test/tsdbSmaTest.cpp
浏览文件 @ 7a6f94a8
......@@ -121,7 +121,7 @@ TEST(testCase, tSma_Meta_Encode_Decode_Test) {
// decode
STSmaWrapper dstTSmaWrapper = {0};
void *result = tDecodeTSmaWrapper(pSW, &dstTSmaWrapper);
void *result = tDecodeTSmaWrapper(pSW, &dstTSmaWrapper, false);
EXPECT_NE(result, nullptr);
EXPECT_EQ(tSmaWrapper.number, dstTSmaWrapper.number);
......
source/libs/function/inc/builtinsimpl.h
浏览文件 @ 7a6f94a8
......@@ -65,9 +65,12 @@ int32_t getAvgInfoSize();
bool getStddevFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool stddevFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
int32_t stddevFunction(SqlFunctionCtx* pCtx);
int32_t stddevFunctionMerge(SqlFunctionCtx* pCtx);
int32_t stddevFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
int32_t stddevPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
int32_t stddevInvertFunction(SqlFunctionCtx* pCtx);
int32_t stddevCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
int32_t getStddevInfoSize();
bool getLeastSQRFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool leastSQRFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
......
source/libs/function/src/builtins.c
浏览文件 @ 7a6f94a8
......@@ -184,6 +184,35 @@ static int32_t translateAvgMerge(SFunctionNode* pFunc, char* pErrBuf, int32_t le
return TSDB_CODE_SUCCESS;
}
static int32_t translateStddevPartial(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
if (1 != LIST_LENGTH(pFunc->pParameterList)) {
return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
}
uint8_t paraType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type;
if (!IS_NUMERIC_TYPE(paraType) && !IS_NULL_TYPE(paraType)) {
return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
}
pFunc->node.resType = (SDataType){.bytes = getStddevInfoSize() + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY};
return TSDB_CODE_SUCCESS;
}
static int32_t translateStddevMerge(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
if (1 != LIST_LENGTH(pFunc->pParameterList)) {
return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
}
uint8_t paraType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type;
if (TSDB_DATA_TYPE_BINARY != paraType) {
return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
}
pFunc->node.resType = (SDataType){.bytes = tDataTypes[TSDB_DATA_TYPE_DOUBLE].bytes, .type = TSDB_DATA_TYPE_DOUBLE};
return TSDB_CODE_SUCCESS;
}
static int32_t translateWduration(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
// pseudo column do not need to check parameters
pFunc->node.resType = (SDataType){.bytes = sizeof(int64_t), .type = TSDB_DATA_TYPE_BIGINT};
......@@ -422,7 +451,8 @@ static int32_t translateTopBotImpl(SFunctionNode* pFunc, char* pErrBuf, int32_t
pValue->notReserved = true;
// set result type
pFunc->node.resType = (SDataType){.bytes = getTopBotInfoSize(pValue->datum.i) + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY};
pFunc->node.resType =
(SDataType){.bytes = getTopBotInfoSize(pValue->datum.i) + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY};
} else {
if (1 != numOfParams) {
return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
......@@ -985,11 +1015,11 @@ static int32_t translateFirstLast(SFunctionNode* pFunc, char* pErrBuf, int32_t l
static int32_t translateFirstLastImpl(SFunctionNode* pFunc, char* pErrBuf, int32_t len, bool isPartial) {
// first(col_list) will be rewritten as first(col)
if (2 != LIST_LENGTH(pFunc->pParameterList)) { //input has two params c0,ts, is this a bug?
if (2 != LIST_LENGTH(pFunc->pParameterList)) { // input has two params c0,ts, is this a bug?
return TSDB_CODE_SUCCESS;
}
SNode* pPara = nodesListGetNode(pFunc->pParameterList, 0);
SNode* pPara = nodesListGetNode(pFunc->pParameterList, 0);
uint8_t paraType = ((SExprNode*)pPara)->resType.type;
int32_t paraBytes = ((SExprNode*)pPara)->resType.bytes;
if (isPartial) {
......@@ -998,8 +1028,8 @@ static int32_t translateFirstLastImpl(SFunctionNode* pFunc, char* pErrBuf, int32
"The parameters of first/last can only be columns");
}
pFunc->node.resType = (SDataType){.bytes = getFirstLastInfoSize(paraBytes) + VARSTR_HEADER_SIZE,
.type = TSDB_DATA_TYPE_BINARY};
pFunc->node.resType =
(SDataType){.bytes = getFirstLastInfoSize(paraBytes) + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY};
} else {
if (TSDB_DATA_TYPE_BINARY != paraType) {
return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
......@@ -1522,6 +1552,32 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
.finalizeFunc = stddevFinalize,
.invertFunc = stddevInvertFunction,
.combineFunc = stddevCombine,
.pPartialFunc = "_stddev_partial",
.pMergeFunc = "_stddev_merge"
},
{
.name = "_stddev_partial",
.type = FUNCTION_TYPE_STDDEV_PARTIAL,
.classification = FUNC_MGT_AGG_FUNC,
.translateFunc = translateStddevPartial,
.getEnvFunc = getStddevFuncEnv,
.initFunc = stddevFunctionSetup,
.processFunc = stddevFunction,
.finalizeFunc = stddevPartialFinalize,
.invertFunc = stddevInvertFunction,
.combineFunc = stddevCombine,
},
{
.name = "_stddev_merge",
.type = FUNCTION_TYPE_STDDEV_MERGE,
.classification = FUNC_MGT_AGG_FUNC,
.translateFunc = translateStddevMerge,
.getEnvFunc = getStddevFuncEnv,
.initFunc = stddevFunctionSetup,
.processFunc = stddevFunctionMerge,
.finalizeFunc = stddevFinalize,
.invertFunc = stddevInvertFunction,
.combineFunc = stddevCombine,
},
{
.name = "leastsquares",
......@@ -1576,7 +1632,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
.name = "percentile",
.type = FUNCTION_TYPE_PERCENTILE,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_REPEAT_SCAN_FUNC,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_REPEAT_SCAN_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
.translateFunc = translatePercentile,
.getEnvFunc = getPercentileFuncEnv,
.initFunc = percentileFunctionSetup,
......@@ -1737,7 +1793,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
.name = "elapsed",
.type = FUNCTION_TYPE_ELAPSED,
.classification = FUNC_MGT_AGG_FUNC,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
.dataRequiredFunc = statisDataRequired,
.translateFunc = translateElapsed,
.getEnvFunc = getElapsedFuncEnv,
......@@ -1858,7 +1914,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
.name = "twa",
.type = FUNCTION_TYPE_TWA,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_TIMELINE_FUNC,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
.translateFunc = translateInNumOutDou,
.getEnvFunc = getTwaFuncEnv,
.initFunc = twaFunctionSetup,
......@@ -1944,7 +2000,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
.name = "diff",
.type = FUNCTION_TYPE_DIFF,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
.translateFunc = translateDiff,
.getEnvFunc = getDiffFuncEnv,
.initFunc = diffFunctionSetup,
......@@ -1954,7 +2010,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
.name = "statecount",
.type = FUNCTION_TYPE_STATE_COUNT,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
.translateFunc = translateStateCount,
.getEnvFunc = getStateFuncEnv,
.initFunc = functionSetup,
......@@ -1964,7 +2020,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
.name = "stateduration",
.type = FUNCTION_TYPE_STATE_DURATION,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
.translateFunc = translateStateDuration,
.getEnvFunc = getStateFuncEnv,
.initFunc = functionSetup,
......@@ -1974,7 +2030,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
.name = "csum",
.type = FUNCTION_TYPE_CSUM,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
.translateFunc = translateCsum,
.getEnvFunc = getCsumFuncEnv,
.initFunc = functionSetup,
......@@ -1984,7 +2040,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
.name = "mavg",
.type = FUNCTION_TYPE_MAVG,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
.translateFunc = translateMavg,
.getEnvFunc = getMavgFuncEnv,
.initFunc = mavgFunctionSetup,
......@@ -1994,7 +2050,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
.name = "sample",
.type = FUNCTION_TYPE_SAMPLE,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_INDEFINITE_ROWS_FUNC,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
.translateFunc = translateSample,
.getEnvFunc = getSampleFuncEnv,
.initFunc = sampleFunctionSetup,
......@@ -2004,7 +2060,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
.name = "tail",
.type = FUNCTION_TYPE_TAIL,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
.translateFunc = translateTail,
.getEnvFunc = getTailFuncEnv,
.initFunc = tailFunctionSetup,
......@@ -2014,7 +2070,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
.name = "unique",
.type = FUNCTION_TYPE_UNIQUE,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
.translateFunc = translateUnique,
.getEnvFunc = getUniqueFuncEnv,
.initFunc = uniqueFunctionSetup,
......
source/libs/function/src/builtinsimpl.c
浏览文件 @ 7a6f94a8
......@@ -89,6 +89,7 @@ typedef struct SStddevRes {
double dsum;
int64_t isum;
};
int16_t type;
} SStddevRes;
typedef struct SLeastSQRInfo {
......@@ -1488,6 +1489,8 @@ int32_t maxCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) {
return minMaxCombine(pDestCtx, pSourceCtx, 0);
}
int32_t getStddevInfoSize() { return (int32_t)sizeof(SStddevRes); }
bool getStddevFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
pEnv->calcMemSize = sizeof(SStddevRes);
return true;
......@@ -1511,6 +1514,7 @@ int32_t stddevFunction(SqlFunctionCtx* pCtx) {
int32_t type = pInput->pData[0]->info.type;
SStddevRes* pStddevRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
pStddevRes->type = type;
// computing based on the true data block
SColumnInfoData* pCol = pInput->pData[0];
......@@ -1627,6 +1631,39 @@ _stddev_over:
return TSDB_CODE_SUCCESS;
}
static void stddevTransferInfo(SStddevRes* pInput, SStddevRes* pOutput) {
pOutput->type = pInput->type;
if (IS_INTEGER_TYPE(pOutput->type)) {
pOutput->quadraticISum += pInput->quadraticISum;
pOutput->isum += pInput->isum;
} else {
pOutput->quadraticDSum += pInput->quadraticDSum;
pOutput->dsum += pInput->dsum;
}
pOutput->count += pInput->count;
return;
}
int32_t stddevFunctionMerge(SqlFunctionCtx* pCtx) {
SInputColumnInfoData* pInput = &pCtx->input;
SColumnInfoData* pCol = pInput->pData[0];
ASSERT(pCol->info.type == TSDB_DATA_TYPE_BINARY);
SStddevRes* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
int32_t start = pInput->startRowIndex;
char* data = colDataGetData(pCol, start);
SStddevRes* pInputInfo = (SStddevRes*)varDataVal(data);
stddevTransferInfo(pInputInfo, pInfo);
SET_VAL(GET_RES_INFO(pCtx), 1, 1);
return TSDB_CODE_SUCCESS;
}
#define LIST_STDDEV_SUB_N(sumT, T) \
do { \
T* plist = (T*)pCol->pData; \
......@@ -1692,9 +1729,10 @@ int32_t stddevInvertFunction(SqlFunctionCtx* pCtx) {
int32_t stddevFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
SInputColumnInfoData* pInput = &pCtx->input;
int32_t type = pInput->pData[0]->info.type;
SStddevRes* pStddevRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
int32_t type = pStddevRes->type;
double avg;
if (IS_INTEGER_TYPE(type)) {
avg = pStddevRes->isum / ((double)pStddevRes->count);
pStddevRes->result = sqrt(pStddevRes->quadraticISum / ((double)pStddevRes->count) - avg * avg);
......@@ -1706,6 +1744,24 @@ int32_t stddevFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
return functionFinalize(pCtx, pBlock);
}
int32_t stddevPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
SStddevRes* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
int32_t resultBytes = getStddevInfoSize();
char* res = taosMemoryCalloc(resultBytes + VARSTR_HEADER_SIZE, sizeof(char));
memcpy(varDataVal(res), pInfo, resultBytes);
varDataSetLen(res, resultBytes);
int32_t slotId = pCtx->pExpr->base.resSchema.slotId;
SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
colDataAppend(pCol, pBlock->info.rows, res, false);
taosMemoryFree(res);
return pResInfo->numOfRes;
}
int32_t stddevCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) {
SResultRowEntryInfo* pDResInfo = GET_RES_INFO(pDestCtx);
SStddevRes* pDBuf = GET_ROWCELL_INTERBUF(pDResInfo);
......
source/libs/parser/src/parAstCreater.c
浏览文件 @ 7a6f94a8
......@@ -230,9 +230,13 @@ SNode* releaseRawExprNode(SAstCreateContext* pCxt, SNode* pNode) {
SRawExprNode* pRawExpr = (SRawExprNode*)pNode;
SNode* pExpr = pRawExpr->pNode;
if (nodesIsExprNode(pExpr)) {
int32_t len = TMIN(sizeof(((SExprNode*)pExpr)->aliasName) - 1, pRawExpr->n);
strncpy(((SExprNode*)pExpr)->aliasName, pRawExpr->p, len);
((SExprNode*)pExpr)->aliasName[len] = '\0';
if (QUERY_NODE_COLUMN == nodeType(pExpr)) {
strcpy(((SExprNode*)pExpr)->aliasName, ((SColumnNode*)pExpr)->colName);
} else {
int32_t len = TMIN(sizeof(((SExprNode*)pExpr)->aliasName) - 1, pRawExpr->n);
strncpy(((SExprNode*)pExpr)->aliasName, pRawExpr->p, len);
((SExprNode*)pExpr)->aliasName[len] = '\0';
}
}
taosMemoryFreeClear(pNode);
return pExpr;
......@@ -801,7 +805,7 @@ SNode* setDatabaseOption(SAstCreateContext* pCxt, SNode* pOptions, EDatabaseOpti
((SDatabaseOptions*)pOptions)->pRetentions = pVal;
break;
case DB_OPTION_SCHEMALESS:
// ((SDatabaseOptions*)pOptions)->schemaless = taosStr2Int8(((SToken*)pVal)->z, NULL, 10);
// ((SDatabaseOptions*)pOptions)->schemaless = taosStr2Int8(((SToken*)pVal)->z, NULL, 10);
((SDatabaseOptions*)pOptions)->schemaless = 1;
break;
default:
......
source/libs/parser/src/parTranslater.c
浏览文件 @ 7a6f94a8
......@@ -537,13 +537,15 @@ static int32_t findAndSetColumn(STranslateContext* pCxt, SColumnNode** pColRef,
SNode* pNode;
FOREACH(pNode, pProjectList) {
SExprNode* pExpr = (SExprNode*)pNode;
if (0 == strcmp(pCol->colName, pExpr->aliasName) ||
(isPrimaryKey((STempTableNode*)pTable, pNode) && isInternalPrimaryKey(pCol))) {
if (0 == strcmp(pCol->colName, pExpr->aliasName)) {
if (*pFound) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_AMBIGUOUS_COLUMN, pCol->colName);
}
setColumnInfoByExpr(pTable, pExpr, pColRef);
*pFound = true;
} else if (isPrimaryKey((STempTableNode*)pTable, pNode) && isInternalPrimaryKey(pCol)) {
setColumnInfoByExpr(pTable, pExpr, pColRef);
*pFound = true;
}
}
}
......@@ -662,9 +664,24 @@ static int32_t parseTimeFromValueNode(STranslateContext* pCxt, SValueNode* pVal)
}
char* pEnd = NULL;
pVal->datum.i = taosStr2Int64(pVal->literal, &pEnd, 10);
return (NULL != pEnd && '\0' == *pEnd) ? TSDB_CODE_SUCCESS : TSDB_CODE_FAILED;
return (NULL != pEnd && '\0' == *pEnd) ? TSDB_CODE_SUCCESS : TSDB_CODE_PAR_WRONG_VALUE_TYPE;
} else {
return TSDB_CODE_FAILED;
return TSDB_CODE_PAR_WRONG_VALUE_TYPE;
}
}
static int32_t parseBoolFromValueNode(STranslateContext* pCxt, SValueNode* pVal) {
if (IS_VAR_DATA_TYPE(pVal->node.resType.type) || TSDB_DATA_TYPE_BOOL == pVal->node.resType.type) {
pVal->datum.b = (0 == strcasecmp(pVal->literal, "true"));
return TSDB_CODE_SUCCESS;
} else if (IS_INTEGER_TYPE(pVal->node.resType.type)) {
pVal->datum.b = (0 != taosStr2Int64(pVal->literal, NULL, 10));
return TSDB_CODE_SUCCESS;
} else if (IS_FLOAT_TYPE(pVal->node.resType.type)) {
pVal->datum.b = (0 != taosStr2Double(pVal->literal, NULL));
return TSDB_CODE_SUCCESS;
} else {
return TSDB_CODE_PAR_WRONG_VALUE_TYPE;
}
}
......@@ -685,7 +702,9 @@ static EDealRes translateValueImpl(STranslateContext* pCxt, SValueNode* pVal, SD
case TSDB_DATA_TYPE_NULL:
break;
case TSDB_DATA_TYPE_BOOL:
pVal->datum.b = (0 == strcasecmp(pVal->literal, "true"));
if (TSDB_CODE_SUCCESS != parseBoolFromValueNode(pCxt, pVal)) {
return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, pVal->literal);
}
*(bool*)&pVal->typeData = pVal->datum.b;
break;
case TSDB_DATA_TYPE_TINYINT: {
......@@ -1068,6 +1087,16 @@ static int32_t translateForbidFillFunc(STranslateContext* pCxt, SFunctionNode* p
return TSDB_CODE_SUCCESS;
}
static int32_t translateWindowPseudoColumnFunc(STranslateContext* pCxt, SFunctionNode* pFunc) {
if (!fmIsWindowPseudoColumnFunc(pFunc->funcId)) {
return TSDB_CODE_SUCCESS;
}
if (NULL == pCxt->pCurrSelectStmt->pWindow) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_WINDOW_PC);
}
return TSDB_CODE_SUCCESS;
}
static void setFuncClassification(SSelectStmt* pSelect, SFunctionNode* pFunc) {
if (NULL != pSelect) {
pSelect->hasAggFuncs = pSelect->hasAggFuncs ? true : fmIsAggFunc(pFunc->funcId);
......@@ -1097,6 +1126,9 @@ static EDealRes translateFunction(STranslateContext* pCxt, SFunctionNode* pFunc)
if (TSDB_CODE_SUCCESS == pCxt->errCode) {
pCxt->errCode = translateForbidFillFunc(pCxt, pFunc);
}
if (TSDB_CODE_SUCCESS == pCxt->errCode) {
pCxt->errCode = translateWindowPseudoColumnFunc(pCxt, pFunc);
}
if (TSDB_CODE_SUCCESS == pCxt->errCode) {
setFuncClassification(pCxt->pCurrSelectStmt, pFunc);
}
......@@ -3106,7 +3138,11 @@ static int32_t translateDropSuperTable(STranslateContext* pCxt, SDropSuperTableS
pStmt->ignoreNotExists);
}
static int32_t setAlterTableField(SAlterTableStmt* pStmt, SMAlterStbReq* pAlterReq) {
static int32_t buildAlterSuperTableReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, SMAlterStbReq* pAlterReq) {
SName tableName;
tNameExtractFullName(toName(pCxt->pParseCxt->acctId, pStmt->dbName, pStmt->tableName, &tableName), pAlterReq->name);
pAlterReq->alterType = pStmt->alterType;
if (TSDB_ALTER_TABLE_UPDATE_OPTIONS == pStmt->alterType) {
pAlterReq->ttl = pStmt->pOptions->ttl;
if ('\0' != pStmt->pOptions->comment[0]) {
......@@ -3154,15 +3190,45 @@ static int32_t setAlterTableField(SAlterTableStmt* pStmt, SMAlterStbReq* pAlterR
return TSDB_CODE_SUCCESS;
}
static int32_t translateAlterTable(STranslateContext* pCxt, SAlterTableStmt* pStmt) {
SMAlterStbReq alterReq = {0};
SName tableName;
tNameExtractFullName(toName(pCxt->pParseCxt->acctId, pStmt->dbName, pStmt->tableName, &tableName), alterReq.name);
alterReq.alterType = pStmt->alterType;
static SSchema* getColSchema(STableMeta* pTableMeta, const char* pTagName) {
int32_t numOfFields = getNumOfTags(pTableMeta) + getNumOfColumns(pTableMeta);
for (int32_t i = 0; i < numOfFields; ++i) {
SSchema* pTagSchema = pTableMeta->schema + i;
if (0 == strcmp(pTagName, pTagSchema->name)) {
return pTagSchema;
}
}
return NULL;
}
static int32_t checkAlterSuperTable(STranslateContext* pCxt, SAlterTableStmt* pStmt) {
if (TSDB_ALTER_TABLE_UPDATE_TAG_VAL == pStmt->alterType || TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME == pStmt->alterType) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_ALTER_TABLE);
}
int32_t code = setAlterTableField(pStmt, &alterReq);
if (TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES == pStmt->alterType ||
TSDB_ALTER_TABLE_UPDATE_TAG_BYTES == pStmt->alterType) {
STableMeta* pTableMeta = NULL;
int32_t code = getTableMeta(pCxt, pStmt->dbName, pStmt->tableName, &pTableMeta);
if (TSDB_CODE_SUCCESS == code) {
SSchema* pSchema = getColSchema(pTableMeta, pStmt->colName);
if (NULL == pSchema) {
code = generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_COLUMN, pStmt->colName);
} else if (!IS_VAR_DATA_TYPE(pSchema->type) || pSchema->type != pStmt->dataType.type ||
pSchema->bytes >= calcTypeBytes(pStmt->dataType)) {
code = generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_MODIFY_COL);
}
}
return code;
}
return TSDB_CODE_SUCCESS;
}
static int32_t translateAlterSuperTable(STranslateContext* pCxt, SAlterTableStmt* pStmt) {
SMAlterStbReq alterReq = {0};
int32_t code = checkAlterSuperTable(pCxt, pStmt);
if (TSDB_CODE_SUCCESS == code) {
code = buildAlterSuperTableReq(pCxt, pStmt, &alterReq);
}
if (TSDB_CODE_SUCCESS == code) {
code = buildCmdMsg(pCxt, TDMT_MND_ALTER_STB, (FSerializeFunc)tSerializeSMAlterStbReq, &alterReq);
}
......@@ -3835,7 +3901,7 @@ static int32_t translateQuery(STranslateContext* pCxt, SNode* pNode) {
code = translateDropSuperTable(pCxt, (SDropSuperTableStmt*)pNode);
break;
case QUERY_NODE_ALTER_TABLE_STMT:
code = translateAlterTable(pCxt, (SAlterTableStmt*)pNode);
code = translateAlterSuperTable(pCxt, (SAlterTableStmt*)pNode);
break;
case QUERY_NODE_CREATE_USER_STMT:
code = translateCreateUser(pCxt, (SCreateUserStmt*)pNode);
......@@ -4445,10 +4511,35 @@ static int32_t translateTagVal(STranslateContext* pCxt, uint8_t precision, SSche
? pCxt->errCode
: TSDB_CODE_SUCCESS);
} else {
// return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_WRONG_VALUE_TYPE, ((SExprNode*)pNode)->aliasName);
return TSDB_CODE_FAILED;
}
}
static int32_t buildJsonTagVal(STranslateContext* pCxt, SSchema* pTagSchema, SValueNode* pVal, SArray* pTagArray,
STag** ppTag) {
if (pVal->literal && strlen(pVal->literal) > (TSDB_MAX_JSON_TAG_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
return buildSyntaxErrMsg(&pCxt->msgBuf, "json string too long than 4095", pVal->literal);
}
return parseJsontoTagData(pVal->literal, pTagArray, ppTag, &pCxt->msgBuf);
}
static int32_t buildNormalTagVal(STranslateContext* pCxt, SSchema* pTagSchema, SValueNode* pVal, SArray* pTagArray) {
if (pVal->node.resType.type != TSDB_DATA_TYPE_NULL) {
void* nodeVal = nodesGetValueFromNode(pVal);
STagVal val = {.cid = pTagSchema->colId, .type = pTagSchema->type};
if (IS_VAR_DATA_TYPE(pTagSchema->type)) {
val.pData = varDataVal(nodeVal);
val.nData = varDataLen(nodeVal);
} else {
memcpy(&val.i64, nodeVal, pTagSchema->bytes);
}
taosArrayPush(pTagArray, &val);
}
return TSDB_CODE_SUCCESS;
}
static int32_t buildKVRowForBindTags(STranslateContext* pCxt, SCreateSubTableClause* pStmt, STableMeta* pSuperTableMeta,
STag** ppTag) {
int32_t numOfTags = getNumOfTags(pSuperTableMeta);
......@@ -4458,11 +4549,10 @@ static int32_t buildKVRowForBindTags(STranslateContext* pCxt, SCreateSubTableCla
}
SArray* pTagArray = taosArrayInit(LIST_LENGTH(pStmt->pValsOfTags), sizeof(STagVal));
if (!pTagArray) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_TSC_OUT_OF_MEMORY);
if (NULL == pTagArray) {
return TSDB_CODE_OUT_OF_MEMORY;
}
int32_t code = TSDB_CODE_SUCCESS;
int16_t nTags = 0, nBufPos = 0;
SSchema* pTagSchema = getTableTagSchema(pSuperTableMeta);
SNode * pTag = NULL, *pNode = NULL;
bool isJson = false;
......@@ -4490,27 +4580,12 @@ static int32_t buildKVRowForBindTags(STranslateContext* pCxt, SCreateSubTableCla
} else {
REPLACE_LIST2_NODE(pVal);
}
if (pTagSchema->type == TSDB_DATA_TYPE_JSON) {
if (pVal->literal && strlen(pVal->literal) > (TSDB_MAX_JSON_TAG_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
code = buildSyntaxErrMsg(&pCxt->msgBuf, "json string too long than 4095", pVal->literal);
goto end;
}
if (pSchema->type == TSDB_DATA_TYPE_JSON) {
isJson = true;
code = parseJsontoTagData(pVal->literal, pTagArray, ppTag, &pCxt->msgBuf);
if (code != TSDB_CODE_SUCCESS) {
goto end;
}
code = buildJsonTagVal(pCxt, pSchema, pVal, pTagArray, ppTag);
} else if (pVal->node.resType.type != TSDB_DATA_TYPE_NULL) {
void* nodeVal = nodesGetValueFromNode(pVal);
STagVal val = {.cid = pTagSchema->colId, .type = pTagSchema->type};
if (IS_VAR_DATA_TYPE(pTagSchema->type)) {
val.pData = varDataVal(nodeVal);
val.nData = varDataLen(nodeVal);
} else {
memcpy(&val.i64, nodeVal, pTagSchema->bytes);
}
taosArrayPush(pTagArray, &val);
code = buildNormalTagVal(pCxt, pSchema, pVal, pTagArray);
}
}
......@@ -4825,17 +4900,6 @@ static int32_t rewriteDropTable(STranslateContext* pCxt, SQuery* pQuery) {
return rewriteToVnodeModifyOpStmt(pQuery, pBufArray);
}
static SSchema* getColSchema(STableMeta* pTableMeta, const char* pTagName) {
int32_t numOfFields = getNumOfTags(pTableMeta) + getNumOfColumns(pTableMeta);
for (int32_t i = 0; i < numOfFields; ++i) {
SSchema* pTagSchema = pTableMeta->schema + i;
if (0 == strcmp(pTagName, pTagSchema->name)) {
return pTagSchema;
}
}
return NULL;
}
static int32_t buildUpdateTagValReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, STableMeta* pTableMeta,
SVAlterTbReq* pReq) {
SSchema* pSchema = getColSchema(pTableMeta, pStmt->colName);
......@@ -4853,6 +4917,10 @@ static int32_t buildUpdateTagValReq(STranslateContext* pCxt, SAlterTableStmt* pS
return pCxt->errCode;
}
if (IS_VAR_DATA_TYPE(pSchema->type) && strlen(pStmt->pVal->literal) > pSchema->bytes) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_WRONG_VALUE_TYPE, pStmt->pVal->literal);
}
pReq->isNull = (TSDB_DATA_TYPE_NULL == pStmt->pVal->node.resType.type);
if (pStmt->pVal->node.resType.type == TSDB_DATA_TYPE_JSON) {
if (pStmt->pVal->literal &&
......
source/libs/parser/src/parUtil.c
浏览文件 @ 7a6f94a8
......@@ -186,6 +186,8 @@ static char* getSyntaxErrFormat(int32_t errCode) {
return "The REDISTRIBUTE VGROUP statement only support 1 to 3 dnodes";
case TSDB_CODE_PAR_FILL_NOT_ALLOWED_FUNC:
return "%s function not allowed in fill query";
case TSDB_CODE_PAR_INVALID_WINDOW_PC:
return "_WSTARTTS, _WENDTS and _WDURATION can only be used in window queries";
case TSDB_CODE_OUT_OF_MEMORY:
return "Out of memory";
default:
......
source/libs/parser/src/parser.c
浏览文件 @ 7a6f94a8
......@@ -207,6 +207,8 @@ int32_t qSetSTableIdForRsma(SNode* pStmt, int64_t uid) {
return TSDB_CODE_FAILED;
}
void qCleanupKeywordsTable() { taosCleanupKeywordsTable(); }
int32_t qStmtBindParams(SQuery* pQuery, TAOS_MULTI_BIND* pParams, int32_t colIdx) {
int32_t code = TSDB_CODE_SUCCESS;
......
source/libs/parser/test/mockCatalog.cpp
浏览文件 @ 7a6f94a8
......@@ -141,16 +141,18 @@ void generateTestT1(MockCatalogService* mcs) {
* c2 | column | VARCHAR | 20 |
* tag1 | tag | INT | 4 |
* tag2 | tag | VARCHAR | 20 |
* tag3 | tag | TIMESTAMP | 8 |
* Child Table: st1s1, st1s2
*/
void generateTestST1(MockCatalogService* mcs) {
ITableBuilder& builder = mcs->createTableBuilder("test", "st1", TSDB_SUPER_TABLE, 3, 2)
ITableBuilder& builder = mcs->createTableBuilder("test", "st1", TSDB_SUPER_TABLE, 3, 3)
.setPrecision(TSDB_TIME_PRECISION_MILLI)
.addColumn("ts", TSDB_DATA_TYPE_TIMESTAMP)
.addColumn("c1", TSDB_DATA_TYPE_INT)
.addColumn("c2", TSDB_DATA_TYPE_BINARY, 20)
.addTag("tag1", TSDB_DATA_TYPE_INT)
.addTag("tag2", TSDB_DATA_TYPE_BINARY, 20);
.addTag("tag2", TSDB_DATA_TYPE_BINARY, 20)
.addTag("tag3", TSDB_DATA_TYPE_TIMESTAMP);
builder.done();
mcs->createSubTable("test", "st1", "st1s1", 1);
mcs->createSubTable("test", "st1", "st1s2", 2);
......@@ -189,17 +191,17 @@ void generateFunctions(MockCatalogService* mcs) {
int32_t __catalogGetHandle(const char* clusterId, struct SCatalog** catalogHandle) { return 0; }
int32_t __catalogGetTableMeta(struct SCatalog* pCatalog, SRequestConnInfo *pConn, const SName* pTableName,
int32_t __catalogGetTableMeta(struct SCatalog* pCatalog, SRequestConnInfo* pConn, const SName* pTableName,
STableMeta** pTableMeta) {
return g_mockCatalogService->catalogGetTableMeta(pTableName, pTableMeta);
}
int32_t __catalogGetTableHashVgroup(struct SCatalog* pCatalog, SRequestConnInfo *pConn,
const SName* pTableName, SVgroupInfo* vgInfo) {
int32_t __catalogGetTableHashVgroup(struct SCatalog* pCatalog, SRequestConnInfo* pConn, const SName* pTableName,
SVgroupInfo* vgInfo) {
return g_mockCatalogService->catalogGetTableHashVgroup(pTableName, vgInfo);
}
int32_t __catalogGetTableDistVgInfo(SCatalog* pCtg, SRequestConnInfo *pConn, const SName* pTableName,
int32_t __catalogGetTableDistVgInfo(SCatalog* pCtg, SRequestConnInfo* pConn, const SName* pTableName,
SArray** pVgList) {
return g_mockCatalogService->catalogGetTableDistVgInfo(pTableName, pVgList);
}
......@@ -209,28 +211,26 @@ int32_t __catalogGetDBVgVersion(SCatalog* pCtg, const char* dbFName, int32_t* ve
return 0;
}
int32_t __catalogGetDBVgInfo(SCatalog* pCtg, SRequestConnInfo *pConn, const char* dbFName,
SArray** pVgList) {
int32_t __catalogGetDBVgInfo(SCatalog* pCtg, SRequestConnInfo* pConn, const char* dbFName, SArray** pVgList) {
return g_mockCatalogService->catalogGetDBVgInfo(dbFName, pVgList);
}
int32_t __catalogGetDBCfg(SCatalog* pCtg, SRequestConnInfo *pConn, const char* dbFName, SDbCfgInfo* pDbCfg) {
int32_t __catalogGetDBCfg(SCatalog* pCtg, SRequestConnInfo* pConn, const char* dbFName, SDbCfgInfo* pDbCfg) {
return 0;
}
int32_t __catalogChkAuth(SCatalog* pCtg, SRequestConnInfo *pConn, const char* user, const char* dbFName,
AUTH_TYPE type, bool* pass) {
int32_t __catalogChkAuth(SCatalog* pCtg, SRequestConnInfo* pConn, const char* user, const char* dbFName, AUTH_TYPE type,
bool* pass) {
*pass = true;
return 0;
}
int32_t __catalogGetUdfInfo(SCatalog* pCtg, SRequestConnInfo *pConn, const char* funcName,
SFuncInfo* pInfo) {
int32_t __catalogGetUdfInfo(SCatalog* pCtg, SRequestConnInfo* pConn, const char* funcName, SFuncInfo* pInfo) {
return g_mockCatalogService->catalogGetUdfInfo(funcName, pInfo);
}
int32_t __catalogRefreshGetTableMeta(SCatalog* pCatalog, SRequestConnInfo *pConn,
const SName* pTableName, STableMeta** pTableMeta, int32_t isSTable) {
int32_t __catalogRefreshGetTableMeta(SCatalog* pCatalog, SRequestConnInfo* pConn, const SName* pTableName,
STableMeta** pTableMeta, int32_t isSTable) {
return g_mockCatalogService->catalogGetTableMeta(pTableName, pTableMeta);
}
......
source/libs/parser/test/parInitialATest.cpp
浏览文件 @ 7a6f94a8
......@@ -163,9 +163,9 @@ TEST_F(ParserInitialATest, alterSTable) {
run("ALTER TABLE st1 DROP COLUMN c1");
clearAlterStbReq();
setAlterStbReqFunc("st1", TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES, 1, "c1", TSDB_DATA_TYPE_VARCHAR,
20 + VARSTR_HEADER_SIZE);
run("ALTER TABLE st1 MODIFY COLUMN c1 VARCHAR(20)");
setAlterStbReqFunc("st1", TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES, 1, "c2", TSDB_DATA_TYPE_VARCHAR,
30 + VARSTR_HEADER_SIZE);
run("ALTER TABLE st1 MODIFY COLUMN c2 VARCHAR(30)");
clearAlterStbReq();
// setAlterStbReqFunc("st1", TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME, 2, "c1", 0, 0, "cc1");
......@@ -179,9 +179,9 @@ TEST_F(ParserInitialATest, alterSTable) {
run("ALTER TABLE st1 DROP TAG tag1");
clearAlterStbReq();
setAlterStbReqFunc("st1", TSDB_ALTER_TABLE_UPDATE_TAG_BYTES, 1, "tag1", TSDB_DATA_TYPE_VARCHAR,
20 + VARSTR_HEADER_SIZE);
run("ALTER TABLE st1 MODIFY TAG tag1 VARCHAR(20)");
setAlterStbReqFunc("st1", TSDB_ALTER_TABLE_UPDATE_TAG_BYTES, 1, "tag2", TSDB_DATA_TYPE_VARCHAR,
30 + VARSTR_HEADER_SIZE);
run("ALTER TABLE st1 MODIFY TAG tag2 VARCHAR(30)");
clearAlterStbReq();
setAlterStbReqFunc("st1", TSDB_ALTER_TABLE_UPDATE_TAG_NAME, 2, "tag1", 0, 0, "tag11");
......@@ -196,6 +196,10 @@ TEST_F(ParserInitialATest, alterSTableSemanticCheck) {
useDb("root", "test");
run("ALTER TABLE st1 RENAME COLUMN c1 cc1", TSDB_CODE_PAR_INVALID_ALTER_TABLE);
run("ALTER TABLE st1 MODIFY COLUMN c2 NCHAR(10)", TSDB_CODE_PAR_INVALID_MODIFY_COL);
run("ALTER TABLE st1 MODIFY TAG tag2 NCHAR(10)", TSDB_CODE_PAR_INVALID_MODIFY_COL);
}
TEST_F(ParserInitialATest, alterTable) {
......@@ -336,6 +340,8 @@ TEST_F(ParserInitialATest, alterTableSemanticCheck) {
useDb("root", "test");
run("ALTER TABLE st1s1 RENAME COLUMN c1 cc1", TSDB_CODE_PAR_INVALID_ALTER_TABLE);
run("ALTER TABLE st1s1 SET TAG tag2 = '123456789012345678901'", TSDB_CODE_PAR_WRONG_VALUE_TYPE);
}
TEST_F(ParserInitialATest, alterUser) {
......
source/libs/parser/test/parInitialCTest.cpp
浏览文件 @ 7a6f94a8
......@@ -548,12 +548,14 @@ TEST_F(ParserInitialCTest, createTable) {
"a14 NCHAR(30), a15 VARCHAR(50)) "
"TTL 100 COMMENT 'test create table' SMA(c1, c2, c3) ROLLUP (MIN) FILE_FACTOR 0.1");
run("CREATE TABLE IF NOT EXISTS t1 USING st1 TAGS(1, 'wxy')");
run("CREATE TABLE IF NOT EXISTS t1 USING st1 TAGS(1, 'wxy', NOW)");
run("CREATE TABLE "
"IF NOT EXISTS test.t1 USING test.st1 (tag1, tag2) TAGS(1, 'abc') "
"IF NOT EXISTS test.t2 USING test.st1 (tag1, tag2) TAGS(2, 'abc') "
"IF NOT EXISTS test.t3 USING test.st1 (tag1, tag2) TAGS(3, 'abc') ");
// run("CREATE TABLE IF NOT EXISTS t1 USING st1 TAGS(1, 'wxy', NOW + 1S)");
}
TEST_F(ParserInitialCTest, createTopic) {
......
source/libs/parser/test/parInsertTest.cpp
浏览文件 @ 7a6f94a8
......@@ -245,8 +245,8 @@ TEST_F(InsertTest, autoCreateTableTest) {
setDatabase("root", "test");
bind(
"insert into st1s1 using st1 tags(1, 'wxy') values (now, 1, \"beijing\")(now+1s, 2, \"shanghai\")(now+2s, 3, "
"\"guangzhou\")");
"insert into st1s1 using st1 tags(1, 'wxy', now) "
"values (now, 1, \"beijing\")(now+1s, 2, \"shanghai\")(now+2s, 3, \"guangzhou\")");
ASSERT_EQ(run(), TSDB_CODE_SUCCESS);
dumpReslut();
checkReslut(1, 3);
......@@ -257,8 +257,8 @@ TEST_F(InsertTest, autoCreateTableTest) {
ASSERT_EQ(run(), TSDB_CODE_SUCCESS);
bind(
"insert into st1s1 using st1 tags(1, 'wxy') values (now, 1, \"beijing\")(now+1s, 2, \"shanghai\")(now+2s, 3, "
"\"guangzhou\")");
"insert into st1s1 using st1 tags(1, 'wxy', now) "
"values (now, 1, \"beijing\")(now+1s, 2, \"shanghai\")(now+2s, 3, \"guangzhou\")");
ASSERT_EQ(runAsync(), TSDB_CODE_SUCCESS);
bind(
......
source/libs/parser/test/parSelectTest.cpp
浏览文件 @ 7a6f94a8
......@@ -219,6 +219,7 @@ TEST_F(ParserSelectTest, intervalSemanticCheck) {
run("SELECT HISTOGRAM(c1, 'log_bin', '{\"start\": -33,\"factor\": 55,\"count\": 5,\"infinity\": false}', 1) FROM t1 "
"WHERE ts > TIMESTAMP '2022-04-01 00:00:00' and ts < TIMESTAMP '2022-04-30 23:59:59' INTERVAL(10s) FILL(NULL)",
TSDB_CODE_PAR_FILL_NOT_ALLOWED_FUNC);
run("SELECT _WSTARTTS, _WENDTS, _WDURATION, sum(c1) FROM t1", TSDB_CODE_PAR_INVALID_WINDOW_PC);
}
TEST_F(ParserSelectTest, subquery) {
......@@ -231,13 +232,22 @@ TEST_F(ParserSelectTest, subquery) {
run("SELECT SUM(a) FROM (SELECT MAX(c1) a, ts FROM st1s1 PARTITION BY TBNAME INTERVAL(1m)) INTERVAL(1n)");
run("SELECT SUM(a) FROM (SELECT MAX(c1) a, _wstartts FROM st1s1 PARTITION BY TBNAME INTERVAL(1m)) INTERVAL(1n)");
run("SELECT _C0 FROM (SELECT _ROWTS, ts FROM st1s1)");
run("SELECT ts FROM (SELECT t1.ts FROM st1s1 t1)");
}
TEST_F(ParserSelectTest, subquerySemanticCheck) {
useDb("root", "test");
run("SELECT SUM(a) FROM (SELECT MAX(c1) a FROM st1s1 INTERVAL(1m)) INTERVAL(1n)", TSDB_CODE_PAR_NOT_ALLOWED_WIN_QUERY,
PARSER_STAGE_TRANSLATE);
run("SELECT SUM(a) FROM (SELECT MAX(c1) a FROM st1s1 INTERVAL(1m)) INTERVAL(1n)",
TSDB_CODE_PAR_NOT_ALLOWED_WIN_QUERY);
run("SELECT ts FROM (SELECT t1.ts AS ts, t2.ts FROM st1s1 t1, st1s2 t2 WHERE t1.ts = t2.ts)",
TSDB_CODE_PAR_AMBIGUOUS_COLUMN);
run("SELECT ts FROM (SELECT ts AS c1 FROM st1s1 t1)", TSDB_CODE_PAR_INVALID_COLUMN);
}
TEST_F(ParserSelectTest, semanticCheck) {
......
source/libs/parser/test/parTestUtil.cpp
浏览文件 @ 7a6f94a8
......@@ -62,7 +62,7 @@ int32_t getLogLevel() { return g_logLevel; }
class ParserTestBaseImpl {
public:
ParserTestBaseImpl(ParserTestBase* pBase) : pBase_(pBase) {}
ParserTestBaseImpl(ParserTestBase* pBase) : pBase_(pBase), sqlNo_(0) {}
void login(const std::string& user) { caseEnv_.user_ = user; }
......@@ -73,6 +73,7 @@ class ParserTestBaseImpl {
}
void run(const string& sql, int32_t expect, ParserStage checkStage) {
++sqlNo_;
if (caseEnv_.nsql_ > 0) {
--(caseEnv_.nsql_);
return;
......@@ -174,7 +175,7 @@ class ParserTestBaseImpl {
}
void dump() {
cout << "==========================================sql : [" << stmtEnv_.sql_ << "]" << endl;
cout << "========================================== " << sqlNo_ << " sql : [" << stmtEnv_.sql_ << "]" << endl;
if (!res_.parsedAst_.empty()) {
cout << "raw syntax tree : " << endl;
cout << res_.parsedAst_ << endl;
......@@ -425,6 +426,7 @@ class ParserTestBaseImpl {
stmtEnv stmtEnv_;
stmtRes res_;
ParserTestBase* pBase_;
int32_t sqlNo_;
};
ParserTestBase::ParserTestBase() : impl_(new ParserTestBaseImpl(this)) {}
......
source/libs/planner/src/planSpliter.c
浏览文件 @ 7a6f94a8
......@@ -416,6 +416,8 @@ static int32_t stbSplCreatePartAggNode(SAggLogicNode* pMergeAgg, SLogicNode** pO
pMergeAgg->node.pTargets = NULL;
SNodeList* pChildren = pMergeAgg->node.pChildren;
pMergeAgg->node.pChildren = NULL;
SNode* pConditions = pMergeAgg->node.pConditions;
pMergeAgg->node.pConditions = NULL;
int32_t code = TSDB_CODE_SUCCESS;
SAggLogicNode* pPartAgg = (SAggLogicNode*)nodesCloneNode((SNode*)pMergeAgg);
......@@ -434,6 +436,7 @@ static int32_t stbSplCreatePartAggNode(SAggLogicNode* pMergeAgg, SLogicNode** pO
}
}
if (TSDB_CODE_SUCCESS == code) {
pMergeAgg->node.pConditions = pConditions;
pMergeAgg->node.pTargets = pTargets;
pPartAgg->node.pChildren = pChildren;
......
source/libs/planner/test/planGroupByTest.cpp
浏览文件 @ 7a6f94a8
......@@ -82,4 +82,6 @@ TEST_F(PlanGroupByTest, stable) {
run("SELECT COUNT(*) FROM st1");
run("SELECT COUNT(*) FROM st1 GROUP BY c1");
run("SELECT SUM(c1) FROM st1 GROUP BY c2 HAVING SUM(c1) IS NOT NULL");
}
source/util/src/terror.c
浏览文件 @ 7a6f94a8
......@@ -559,8 +559,10 @@ TAOS_DEFINE_ERROR(TSDB_CODE_TSMA_ALREADY_EXIST, "Tsma already exists
TAOS_DEFINE_ERROR(TSDB_CODE_TSMA_NO_INDEX_IN_META, "No tsma index in meta")
TAOS_DEFINE_ERROR(TSDB_CODE_TSMA_INVALID_ENV, "Invalid tsma env")
TAOS_DEFINE_ERROR(TSDB_CODE_TSMA_INVALID_STAT, "Invalid tsma state")
TAOS_DEFINE_ERROR(TSDB_CODE_TSMA_INVALID_PTR, "Invalid tsma pointer")
TAOS_DEFINE_ERROR(TSDB_CODE_TSMA_INVALID_PARA, "Invalid tsma parameters")
TAOS_DEFINE_ERROR(TSDB_CODE_TSMA_NO_INDEX_IN_CACHE, "No tsma index in cache")
TAOS_DEFINE_ERROR(TSDB_CODE_TSMA_RM_SKEY_IN_HASH, "Rm tsma skey in cache")
//rsma
TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_INVALID_ENV, "Invalid rsma env")
......
tests/script/jenkins/basic.txt
浏览文件 @ 7a6f94a8
......@@ -2,20 +2,20 @@
#======================b1-start===============
# ---- user
#./test.sh -f tsim/user/basic1.sim
#./test.sh -f tsim/user/pass_alter.sim
#./test.sh -f tsim/user/pass_len.sim
#./test.sh -f tsim/user/user_len.sim
#./test.sh -f tsim/user/privilege1.sim
#./test.sh -f tsim/user/privilege2.sim#
./test.sh -f tsim/user/basic1.sim
./test.sh -f tsim/user/pass_alter.sim
./test.sh -f tsim/user/pass_len.sim
./test.sh -f tsim/user/user_len.sim
./test.sh -f tsim/user/privilege1.sim
./test.sh -f tsim/user/privilege2.sim
## ---- db
#./test.sh -f tsim/db/create_all_options.sim
#./test.sh -f tsim/db/alter_option.sim
#./test.sh -f tsim/db/basic1.sim
#./test.sh -f tsim/db/basic2.sim
#./test.sh -f tsim/db/basic3.sim
#./test.sh -f tsim/db/basic6.sim
./test.sh -f tsim/db/create_all_options.sim
./test.sh -f tsim/db/alter_option.sim
./test.sh -f tsim/db/basic1.sim
./test.sh -f tsim/db/basic2.sim
./test.sh -f tsim/db/basic3.sim
./test.sh -f tsim/db/basic6.sim
./test.sh -f tsim/db/basic7.sim
./test.sh -f tsim/db/error1.sim
./test.sh -f tsim/db/taosdlog.sim
......@@ -78,6 +78,7 @@
# ./test.sh -f tsim/stream/state0.sim
# ./test.sh -f tsim/stream/triggerInterval0.sim
# ./test.sh -f tsim/stream/triggerSession0.sim
# ./test.sh -f tsim/stream/partitionby.sim
# ---- transaction
......
tests/script/tsim/stream/partitionby.sim 0 → 100644
浏览文件 @ 7a6f94a8
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/exec.sh -n dnode1 -s start
sleep 50
sql connect
sql create database test vgroups 4;
sql use test;
sql create stable st(ts timestamp, a int, b int , c int, d double) tags(ta int,tb int,tc int);
sql create table ts1 using st tags(1,1,1);
sql create table ts2 using st tags(2,2,2);
sql create table ts3 using st tags(3,2,2);
sql create table ts4 using st tags(4,2,2);
sql create stream stream_t1 trigger at_once into streamtST1 as select _wstartts, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5 from st partition by ta,tb,tc interval(10s);
sql insert into ts1 values(1648791213001,1,12,3,1.0);
sql insert into ts2 values(1648791213001,1,12,3,1.0);
sql insert into ts3 values(1648791213001,1,12,3,1.0);
sql insert into ts4 values(1648791213001,1,12,3,1.0);
$loop_count = 0
loop0:
sleep 300
sql select * from streamtST1;
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
if $rows != 4 then
print =====rows=$rows
goto loop0
endi
sql create database test1 vgroups 1;
sql use test1;
sql create stable st(ts timestamp,a int,b int,c int) tags(ta int,tb int,tc int);
sql create table ts1 using st tags(1,2,3);
sql create table ts2 using st tags(1,3,4);
sql create table ts3 using st tags(1,4,5);
sql create stream streams1 trigger at_once into streamt as select _wstartts, count(*) c1, count(a) c2 from st partition by ta,tb,tc interval(10s);
sql insert into ts1 values(1648791211000,1,2,3);
sql insert into ts2 values(1648791211000,1,2,3);
$loop_count = 0
loop0:
sleep 300
sql select * from streamt;
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
if $rows != 2 then
print =====rows=$rows
goto loop0
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT
\ No newline at end of file
tests/system-test/fulltest.bat
浏览文件 @ 7a6f94a8
......@@ -6,7 +6,7 @@ python3 .\test.py -f 0-others\telemetry.py
python3 .\test.py -f 0-others\taosdMonitor.py
python3 .\test.py -f 0-others\udfTest.py
python3 .\test.py -f 0-others\udf_create.py
@REM python3 .\test.py -f 0-others\udf_restart_taosd.py
python3 .\test.py -f 0-others\udf_restart_taosd.py
python3 .\test.py -f 0-others\cachelast.py
python3 .\test.py -f 0-others\user_control.py
......@@ -38,7 +38,7 @@ python3 .\test.py -f 2-query\concat_ws.py
python3 .\test.py -f 2-query\concat_ws2.py
python3 .\test.py -f 2-query\check_tsdb.py
python3 .\test.py -f 2-query\spread.py
@REM python3 .\test.py -f 2-query\hyperloglog.py
python3 .\test.py -f 2-query\hyperloglog.py
python3 .\test.py -f 2-query\timezone.py
python3 .\test.py -f 2-query\Now.py
......@@ -82,7 +82,7 @@ python3 .\test.py -f 2-query\elapsed.py
python3 .\test.py -f 2-query\csum.py
python3 .\test.py -f 2-query\mavg.py
python3 .\test.py -f 2-query\diff.py
@REM python3 .\test.py -f 2-query\sample.py
python3 .\test.py -f 2-query\sample.py
python3 .\test.py -f 2-query\function_diff.py
python3 .\test.py -f 2-query\unique.py
python3 .\test.py -f 2-query\stateduration.py
......@@ -91,7 +91,7 @@ python3 .\test.py -f 2-query\statecount.py
python3 .\test.py -f 7-tmq\basic5.py
python3 .\test.py -f 7-tmq\subscribeDb.py
@REM python3 .\test.py -f 7-tmq\subscribeDb0.py
python3 .\test.py -f 7-tmq\subscribeDb0.py
python3 .\test.py -f 7-tmq\subscribeDb1.py
python3 .\test.py -f 7-tmq\subscribeStb.py
python3 .\test.py -f 7-tmq\subscribeStb0.py
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或