diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 71be1a5014..1a46a462b1 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -2097,6 +2097,18 @@ enum { TOPIC_SUB_TYPE__TABLE, }; +typedef struct { + SMsgHead head; + int64_t leftForVer; + int32_t vgId; + int64_t consumerId; + char subKey[TSDB_SUBSCRIBE_KEY_LEN]; +} SMqVDeleteReq; + +typedef struct { + int8_t reserved; +} SMqVDeleteRsp; + typedef struct { int64_t leftForVer; int32_t vgId; @@ -2255,20 +2267,22 @@ static FORCE_INLINE void tdDestroyTSma(STSma* pSma) { } } -static FORCE_INLINE void tdDestroyTSmaWrapper(STSmaWrapper* pSW) { +static FORCE_INLINE void tdDestroyTSmaWrapper(STSmaWrapper* pSW, bool deepCopy) { if (pSW) { if (pSW->tSma) { - for (uint32_t i = 0; i < pSW->number; ++i) { - tdDestroyTSma(pSW->tSma + i); + if (deepCopy) { + for (uint32_t i = 0; i < pSW->number; ++i) { + tdDestroyTSma(pSW->tSma + i); + } } taosMemoryFreeClear(pSW->tSma); } } } -static FORCE_INLINE void* tdFreeTSmaWrapper(STSmaWrapper* pSW) { - tdDestroyTSmaWrapper(pSW); - taosMemoryFree(pSW); +static FORCE_INLINE void* tdFreeTSmaWrapper(STSmaWrapper* pSW, bool deepCopy) { + tdDestroyTSmaWrapper(pSW, deepCopy); + taosMemoryFreeClear(pSW); return NULL; } @@ -2532,11 +2546,15 @@ static FORCE_INLINE void tDeleteSMqAskEpRsp(SMqAskEpRsp* pRsp) { } typedef struct { - void* data; + int64_t streamId; + int32_t taskId; + int32_t sourceVg; + int64_t sourceVer; + SArray* data; // SArray } SStreamDispatchReq; typedef struct { - int8_t status; + int8_t inputStatus; } SStreamDispatchRsp; #define TD_AUTO_CREATE_TABLE 0x1 diff --git a/include/common/tmsgdef.h b/include/common/tmsgdef.h index c7deaa7845..93b2e75360 100644 --- a/include/common/tmsgdef.h +++ b/include/common/tmsgdef.h @@ -178,6 +178,7 @@ enum { TD_DEF_MSG_TYPE(TDMT_VND_MQ_CONNECT, "vnode-mq-connect", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_VND_MQ_DISCONNECT, "vnode-mq-disconnect", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_VND_MQ_VG_CHANGE, "vnode-mq-vg-change", SMqRebVgReq, SMqRebVgRsp) + TD_DEF_MSG_TYPE(TDMT_VND_MQ_VG_DELETE, "vnode-mq-vg-delete", SMqVDeleteReq, SMqVDeleteRsp) TD_DEF_MSG_TYPE(TDMT_VND_RES_READY, "vnode-res-ready", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_VND_TASKS_STATUS, "vnode-tasks-status", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_VND_CANCEL_TASK, "vnode-cancel-task", NULL, NULL) diff --git a/include/libs/function/functionMgt.h b/include/libs/function/functionMgt.h index 2b58ed7c0b..aec1476663 100644 --- a/include/libs/function/functionMgt.h +++ b/include/libs/function/functionMgt.h @@ -41,6 +41,7 @@ typedef enum EFunctionType { FUNCTION_TYPE_SUM, FUNCTION_TYPE_TWA, FUNCTION_TYPE_HISTOGRAM, + FUNCTION_TYPE_HYPERLOGLOG, // nonstandard SQL function FUNCTION_TYPE_BOTTOM = 500, diff --git a/include/libs/function/tudf.h b/include/libs/function/tudf.h index 6a98138c6c..28b1fbe8ce 100644 --- a/include/libs/function/tudf.h +++ b/include/libs/function/tudf.h @@ -87,6 +87,7 @@ typedef struct SUdfInterBuf { } SUdfInterBuf; typedef void *UdfcFuncHandle; +//low level APIs /** * setup udf * @param udf, in @@ -115,6 +116,9 @@ int32_t doCallUdfScalarFunc(UdfcFuncHandle handle, SScalarParam *input, int32_t */ int32_t doTeardownUdf(UdfcFuncHandle handle); +void freeUdfInterBuf(SUdfInterBuf *buf); + +//high level APIs bool udfAggGetEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv); bool udfAggInit(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResultCellInfo); int32_t udfAggProcess(struct SqlFunctionCtx *pCtx); diff --git a/include/util/taoserror.h b/include/util/taoserror.h index cc10348d24..31b6dad2d8 100644 --- a/include/util/taoserror.h +++ b/include/util/taoserror.h @@ -281,6 +281,7 @@ int32_t* taosGetErrno(); #define TSDB_CODE_MND_SUBSCRIBE_NOT_EXIST TAOS_DEF_ERROR_CODE(0, 0x03E8) #define TSDB_CODE_MND_OFFSET_NOT_EXIST TAOS_DEF_ERROR_CODE(0, 0x03E9) #define TSDB_CODE_MND_CONSUMER_NOT_READY TAOS_DEF_ERROR_CODE(0, 0x03EA) +#define TSDB_CODE_MND_TOPIC_SUBSCRIBED TAOS_DEF_ERROR_CODE(0, 0x03EB) // mnode-stream #define TSDB_CODE_MND_STREAM_ALREADY_EXIST TAOS_DEF_ERROR_CODE(0, 0x03F0) diff --git a/include/util/thash.h b/include/util/thash.h index f2ef445777..fc8785a8fb 100644 --- a/include/util/thash.h +++ b/include/util/thash.h @@ -40,6 +40,7 @@ typedef void (*_hash_free_fn_t)(void *); */ uint32_t MurmurHash3_32(const char *key, uint32_t len); +uint64_t MurmurHash3_64(const char *key, uint32_t len); /** * * @param key diff --git a/source/dnode/mgmt/mgmt_mnode/src/mmHandle.c b/source/dnode/mgmt/mgmt_mnode/src/mmHandle.c index 5548a23c55..b998f776c0 100644 --- a/source/dnode/mgmt/mgmt_mnode/src/mmHandle.c +++ b/source/dnode/mgmt/mgmt_mnode/src/mmHandle.c @@ -216,6 +216,7 @@ SArray *mmGetMsgHandles() { if (dmSetMgmtHandle(pArray, TDMT_MND_MQ_COMMIT_OFFSET, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_MND_MQ_ASK_EP, mmPutNodeMsgToReadQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_CHANGE_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_DELETE_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_MND_CREATE_STREAM, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_TASK_DEPLOY_RSP, mmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_MND_GET_DB_CFG, mmPutNodeMsgToReadQueue, 0) == NULL) goto _OVER; diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c index 0814568b73..408613041f 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c @@ -308,6 +308,7 @@ SArray *vmGetMsgHandles() { if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_SMA, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_SUBMIT_RSMA, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_CHANGE, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_DELETE, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_CONSUME, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_TASK_DEPLOY, vmPutNodeMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY_HEARTBEAT, vmPutNodeMsgToFetchQueue, 0) == NULL) goto _OVER; diff --git a/source/dnode/mnode/impl/inc/mndDef.h b/source/dnode/mnode/impl/inc/mndDef.h index 5eb6866387..0ce67dd3c2 100644 --- a/source/dnode/mnode/impl/inc/mndDef.h +++ b/source/dnode/mnode/impl/inc/mndDef.h @@ -459,6 +459,7 @@ typedef struct { char* ast; char* physicalPlan; SSchemaWrapper schema; + int32_t refConsumerCnt; } SMqTopicObj; typedef struct { diff --git a/source/dnode/mnode/impl/inc/mndOffset.h b/source/dnode/mnode/impl/inc/mndOffset.h index b468496165..900181858b 100644 --- a/source/dnode/mnode/impl/inc/mndOffset.h +++ b/source/dnode/mnode/impl/inc/mndOffset.h @@ -38,6 +38,9 @@ static FORCE_INLINE int32_t mndMakePartitionKey(char *key, const char *cgroup, c } int32_t mndDropOffsetByDB(SMnode *pMnode, STrans *pTrans, SDbObj *pDb); +int32_t mndDropOffsetByTopic(SMnode *pMnode, STrans *pTrans, const char *topic); + +bool mndOffsetFromTopic(SMqOffsetObj *pOffset, const char *topic); #ifdef __cplusplus } diff --git a/source/dnode/mnode/impl/inc/mndSubscribe.h b/source/dnode/mnode/impl/inc/mndSubscribe.h index 5ca672e8dd..50cede62ce 100644 --- a/source/dnode/mnode/impl/inc/mndSubscribe.h +++ b/source/dnode/mnode/impl/inc/mndSubscribe.h @@ -32,6 +32,7 @@ void mndReleaseSubscribe(SMnode *pMnode, SMqSubscribeObj *pSub); int32_t mndMakeSubscribeKey(char *key, const char *cgroup, const char *topicName); int32_t mndDropSubByDB(SMnode *pMnode, STrans *pTrans, SDbObj *pDb); +int32_t mndDropSubByTopic(SMnode *pMnode, STrans *pTrans, const char *topic); #ifdef __cplusplus } diff --git a/source/dnode/mnode/impl/inc/mndTopic.h b/source/dnode/mnode/impl/inc/mndTopic.h index be3f9c3283..d7e6f9c87b 100644 --- a/source/dnode/mnode/impl/inc/mndTopic.h +++ b/source/dnode/mnode/impl/inc/mndTopic.h @@ -35,6 +35,8 @@ int32_t mndDropTopicByDB(SMnode *pMnode, STrans *pTrans, SDbObj *pDb); const char *mndTopicGetShowName(const char topic[TSDB_TOPIC_FNAME_LEN]); +int32_t mndSetTopicRedoLogs(SMnode *pMnode, STrans *pTrans, SMqTopicObj *pTopic); + #ifdef __cplusplus } #endif diff --git a/source/dnode/mnode/impl/src/mndConsumer.c b/source/dnode/mnode/impl/src/mndConsumer.c index 9c8c6d32eb..76b8081849 100644 --- a/source/dnode/mnode/impl/src/mndConsumer.c +++ b/source/dnode/mnode/impl/src/mndConsumer.c @@ -399,6 +399,9 @@ static int32_t mndProcessSubscribeReq(SNodeMsg *pMsg) { int32_t newTopicNum = taosArrayGetSize(newSub); // check topic existance + STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_SUBSCRIBE, &pMsg->rpcMsg); + if (pTrans == NULL) goto SUBSCRIBE_OVER; + for (int32_t i = 0; i < newTopicNum; i++) { char *topic = taosArrayGetP(newSub, i); SMqTopicObj *pTopic = mndAcquireTopic(pMnode, topic); @@ -406,7 +409,14 @@ static int32_t mndProcessSubscribeReq(SNodeMsg *pMsg) { terrno = TSDB_CODE_MND_TOPIC_NOT_EXIST; goto SUBSCRIBE_OVER; } - // TODO lock topic to prevent drop + + // ref topic to prevent drop + // TODO make topic complete + SMqTopicObj topicObj = {0}; + memcpy(&topicObj, pTopic, sizeof(SMqTopicObj)); + topicObj.refConsumerCnt = pTopic->refConsumerCnt + 1; + if (mndSetTopicRedoLogs(pMnode, pTrans, &topicObj) != 0) goto SUBSCRIBE_OVER; + mndReleaseTopic(pMnode, pTopic); } @@ -422,8 +432,6 @@ static int32_t mndProcessSubscribeReq(SNodeMsg *pMsg) { taosArrayPush(pConsumerNew->assignedTopics, &newTopicCopy); } - STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_SUBSCRIBE, &pMsg->rpcMsg); - if (pTrans == NULL) goto SUBSCRIBE_OVER; if (mndSetConsumerCommitLogs(pMnode, pTrans, pConsumerNew) != 0) goto SUBSCRIBE_OVER; if (mndTransPrepare(pMnode, pTrans) != 0) goto SUBSCRIBE_OVER; @@ -494,8 +502,6 @@ static int32_t mndProcessSubscribeReq(SNodeMsg *pMsg) { goto SUBSCRIBE_OVER; } - STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_SUBSCRIBE, &pMsg->rpcMsg); - if (pTrans == NULL) goto SUBSCRIBE_OVER; if (mndSetConsumerCommitLogs(pMnode, pTrans, pConsumerNew) != 0) goto SUBSCRIBE_OVER; if (mndTransPrepare(pMnode, pTrans) != 0) goto SUBSCRIBE_OVER; } @@ -503,6 +509,8 @@ static int32_t mndProcessSubscribeReq(SNodeMsg *pMsg) { code = TSDB_CODE_MND_ACTION_IN_PROGRESS; SUBSCRIBE_OVER: + mndTransDrop(pTrans); + if (pConsumerOld) { /*taosRUnLockLatch(&pConsumerOld->lock);*/ mndReleaseConsumer(pMnode, pConsumerOld); diff --git a/source/dnode/mnode/impl/src/mndOffset.c b/source/dnode/mnode/impl/src/mndOffset.c index 24a7c0d389..a7d174c9c5 100644 --- a/source/dnode/mnode/impl/src/mndOffset.c +++ b/source/dnode/mnode/impl/src/mndOffset.c @@ -50,6 +50,14 @@ int32_t mndInitOffset(SMnode *pMnode) { void mndCleanupOffset(SMnode *pMnode) {} +bool mndOffsetFromTopic(SMqOffsetObj *pOffset, const char *topic) { + int32_t i = 0; + while (pOffset->key[i] != ':') i++; + while (pOffset->key[i] != ':') i++; + if (strcmp(&pOffset->key[i + 1], topic) == 0) return true; + return false; +} + SSdbRaw *mndOffsetActionEncode(SMqOffsetObj *pOffset) { terrno = TSDB_CODE_OUT_OF_MEMORY; void *buf = NULL; @@ -134,10 +142,11 @@ int32_t mndCreateOffsets(STrans *pTrans, const char *cgroup, const char *topicNa int32_t sz = taosArrayGetSize(vgs); for (int32_t i = 0; i < sz; i++) { int32_t vgId = *(int32_t *)taosArrayGet(vgs, i); - SMqOffsetObj offsetObj; + SMqOffsetObj offsetObj = {0}; if (mndMakePartitionKey(offsetObj.key, cgroup, topicName, vgId) < 0) { return -1; } + // TODO assign db offsetObj.offset = -1; SSdbRaw *pOffsetRaw = mndOffsetActionEncode(&offsetObj); if (pOffsetRaw == NULL) { @@ -240,6 +249,14 @@ static int32_t mndSetDropOffsetCommitLogs(SMnode *pMnode, STrans *pTrans, SMqOff return 0; } +static int32_t mndSetDropOffsetRedoLogs(SMnode *pMnode, STrans *pTrans, SMqOffsetObj *pOffset) { + SSdbRaw *pRedoRaw = mndOffsetActionEncode(pOffset); + if (pRedoRaw == NULL) return -1; + if (mndTransAppendRedolog(pTrans, pRedoRaw) != 0) return -1; + if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPED) != 0) return -1; + return 0; +} + int32_t mndDropOffsetByDB(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) { int32_t code = -1; SSdb *pSdb = pMnode->pSdb; @@ -247,7 +264,7 @@ int32_t mndDropOffsetByDB(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) { void *pIter = NULL; SMqOffsetObj *pOffset = NULL; while (1) { - pIter = sdbFetch(pSdb, SDB_SUBSCRIBE, pIter, (void **)&pOffset); + pIter = sdbFetch(pSdb, SDB_OFFSET, pIter, (void **)&pOffset); if (pIter == NULL) break; if (pOffset->dbUid != pDb->uid) { @@ -256,8 +273,39 @@ int32_t mndDropOffsetByDB(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) { } if (mndSetDropOffsetCommitLogs(pMnode, pTrans, pOffset) < 0) { + sdbRelease(pSdb, pOffset); goto END; } + + sdbRelease(pSdb, pOffset); + } + + code = 0; +END: + return code; +} + +int32_t mndDropOffsetByTopic(SMnode *pMnode, STrans *pTrans, const char *topic) { + int32_t code = -1; + SSdb *pSdb = pMnode->pSdb; + + void *pIter = NULL; + SMqOffsetObj *pOffset = NULL; + while (1) { + pIter = sdbFetch(pSdb, SDB_OFFSET, pIter, (void **)&pOffset); + if (pIter == NULL) break; + + if (!mndOffsetFromTopic(pOffset, topic)) { + sdbRelease(pSdb, pOffset); + continue; + } + + if (mndSetDropOffsetRedoLogs(pMnode, pTrans, pOffset) < 0) { + sdbRelease(pSdb, pOffset); + goto END; + } + + sdbRelease(pSdb, pOffset); } code = 0; diff --git a/source/dnode/mnode/impl/src/mndSma.c b/source/dnode/mnode/impl/src/mndSma.c index 869bf0c0a9..a9963651fd 100644 --- a/source/dnode/mnode/impl/src/mndSma.c +++ b/source/dnode/mnode/impl/src/mndSma.c @@ -530,6 +530,7 @@ static int32_t mndProcessMCreateSmaReq(SNodeMsg *pReq) { pStream = mndAcquireStream(pMnode, createReq.name); if (pStream != NULL) { mError("sma:%s, failed to create since stream:%s already exist", createReq.name, createReq.name); + terrno = TSDB_CODE_MND_STREAM_ALREADY_EXIST; goto _OVER; } @@ -565,7 +566,7 @@ static int32_t mndProcessMCreateSmaReq(SNodeMsg *pReq) { _OVER: if (code != 0 && code != TSDB_CODE_MND_ACTION_IN_PROGRESS) { - mError("sma:%s, failed to create since %s", createReq.name, terrstr()); + mError("sma:%s, failed to create since %s", createReq.name, terrstr(terrno)); } mndReleaseStb(pMnode, pStb); diff --git a/source/dnode/mnode/impl/src/mndStb.c b/source/dnode/mnode/impl/src/mndStb.c index 5a834f4d33..faa0d71d3b 100644 --- a/source/dnode/mnode/impl/src/mndStb.c +++ b/source/dnode/mnode/impl/src/mndStb.c @@ -386,7 +386,7 @@ static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pSt req.schema.sver = pStb->version; req.schema.pSchema = pStb->pColumns; req.schemaTag.nCols = pStb->numOfTags; - req.schemaTag.nCols = 0; + req.schemaTag.sver = 1; req.schemaTag.pSchema = pStb->pTags; if (req.rollup) { diff --git a/source/dnode/mnode/impl/src/mndSubscribe.c b/source/dnode/mnode/impl/src/mndSubscribe.c index 5ad4863322..3e932e8a67 100644 --- a/source/dnode/mnode/impl/src/mndSubscribe.c +++ b/source/dnode/mnode/impl/src/mndSubscribe.c @@ -73,6 +73,7 @@ int32_t mndInitSubscribe(SMnode *pMnode) { .deleteFp = (SdbDeleteFp)mndSubActionDelete}; mndSetMsgHandle(pMnode, TDMT_VND_MQ_VG_CHANGE_RSP, mndProcessSubscribeInternalRsp); + mndSetMsgHandle(pMnode, TDMT_VND_MQ_VG_DELETE_RSP, mndProcessSubscribeInternalRsp); mndSetMsgHandle(pMnode, TDMT_MND_MQ_DO_REBALANCE, mndProcessRebalanceReq); mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_SUBSCRIPTIONS, mndRetrieveSubscribe); @@ -389,7 +390,7 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR } static int32_t mndPersistRebResult(SMnode *pMnode, SNodeMsg *pMsg, const SMqRebOutputObj *pOutput) { - STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_REBALANCE, &pMsg->rpcMsg); + STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_REBALANCE, &pMsg->rpcMsg); if (pTrans == NULL) { return -1; } @@ -458,6 +459,20 @@ static int32_t mndPersistRebResult(SMnode *pMnode, SNodeMsg *pMsg, const SMqRebO goto REB_FAIL; } } + if (consumerNum) { + char topic[TSDB_TOPIC_FNAME_LEN]; + char cgroup[TSDB_CGROUP_LEN]; + mndSplitSubscribeKey(pOutput->pSub->key, topic, cgroup, true); + SMqTopicObj *pTopic = mndAcquireTopic(pMnode, topic); + if (pTopic) { + // TODO make topic complete + SMqTopicObj topicObj = {0}; + memcpy(&topicObj, pTopic, sizeof(SMqTopicObj)); + topicObj.refConsumerCnt = pTopic->refConsumerCnt - consumerNum; + if (mndSetTopicRedoLogs(pMnode, pTrans, &topicObj) != 0) goto REB_FAIL; + } + } + // 4. TODO commit log: modification log // 5. set cb @@ -688,6 +703,14 @@ static int32_t mndProcessSubscribeInternalRsp(SNodeMsg *pRsp) { return 0; } +static int32_t mndSetDropSubRedoLogs(SMnode *pMnode, STrans *pTrans, SMqSubscribeObj *pSub) { + SSdbRaw *pRedoRaw = mndSubActionEncode(pSub); + if (pRedoRaw == NULL) return -1; + if (mndTransAppendRedolog(pTrans, pRedoRaw) != 0) return -1; + if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPED) != 0) return -1; + return 0; +} + static int32_t mndSetDropSubCommitLogs(SMnode *pMnode, STrans *pTrans, SMqSubscribeObj *pSub) { SSdbRaw *pCommitRaw = mndSubActionEncode(pSub); if (pCommitRaw == NULL) return -1; @@ -712,6 +735,57 @@ int32_t mndDropSubByDB(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) { } if (mndSetDropSubCommitLogs(pMnode, pTrans, pSub) < 0) { + sdbRelease(pSdb, pSub); + goto END; + } + } + + code = 0; +END: + return code; +} + +int32_t mndDropSubByTopic(SMnode *pMnode, STrans *pTrans, const char *topicName) { + int32_t code = -1; + SSdb *pSdb = pMnode->pSdb; + + void *pIter = NULL; + SMqSubscribeObj *pSub = NULL; + while (1) { + pIter = sdbFetch(pSdb, SDB_SUBSCRIBE, pIter, (void **)&pSub); + if (pIter == NULL) break; + + char topic[TSDB_TOPIC_FNAME_LEN]; + char cgroup[TSDB_CGROUP_LEN]; + mndSplitSubscribeKey(pSub->key, topic, cgroup, true); + if (strcmp(topic, topicName) != 0) { + sdbRelease(pSdb, pSub); + continue; + } + + // iter all vnode to delete handle + ASSERT(taosHashGetSize(pSub->consumerHash) == 0); + int32_t sz = taosArrayGetSize(pSub->unassignedVgs); + for (int32_t i = 0; i < sz; i++) { + SMqVgEp *pVgEp = taosArrayGetP(pSub->unassignedVgs, i); + SMqVDeleteReq *pReq = taosMemoryCalloc(1, sizeof(SMqVDeleteReq)); + pReq->head.vgId = htonl(pVgEp->vgId); + pReq->vgId = pVgEp->vgId; + pReq->consumerId = -1; + memcpy(pReq->subKey, pSub->key, TSDB_SUBSCRIBE_KEY_LEN); + STransAction action = {0}; + action.epSet = pVgEp->epSet; + action.pCont = pReq; + action.contLen = sizeof(SMqVDeleteReq); + action.msgType = TDMT_VND_MQ_VG_DELETE; + if (mndTransAppendRedoAction(pTrans, &action) != 0) { + taosMemoryFree(pReq); + return -1; + } + } + + if (mndSetDropSubRedoLogs(pMnode, pTrans, pSub) < 0) { + sdbRelease(pSdb, pSub); goto END; } } diff --git a/source/dnode/mnode/impl/src/mndTopic.c b/source/dnode/mnode/impl/src/mndTopic.c index bd2923ac1a..bc650ec95b 100644 --- a/source/dnode/mnode/impl/src/mndTopic.c +++ b/source/dnode/mnode/impl/src/mndTopic.c @@ -18,8 +18,10 @@ #include "mndDb.h" #include "mndDnode.h" #include "mndMnode.h" +#include "mndOffset.h" #include "mndShow.h" #include "mndStb.h" +#include "mndSubscribe.h" #include "mndTrans.h" #include "mndUser.h" #include "mndVgroup.h" @@ -106,6 +108,7 @@ SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic) { taosEncodeSSchemaWrapper(&aswBuf, &pTopic->schema); SDB_SET_INT32(pRaw, dataPos, schemaLen, TOPIC_ENCODE_OVER); SDB_SET_BINARY(pRaw, dataPos, swBuf, schemaLen, TOPIC_ENCODE_OVER); + SDB_SET_INT32(pRaw, dataPos, pTopic->refConsumerCnt, TOPIC_ENCODE_OVER); SDB_SET_RESERVE(pRaw, dataPos, MND_TOPIC_RESERVE_SIZE, TOPIC_ENCODE_OVER); SDB_SET_DATALEN(pRaw, dataPos, TOPIC_ENCODE_OVER); @@ -190,6 +193,8 @@ SSdbRow *mndTopicActionDecode(SSdbRaw *pRaw) { goto TOPIC_DECODE_OVER; } + SDB_GET_INT32(pRaw, dataPos, &pTopic->refConsumerCnt, TOPIC_DECODE_OVER); + SDB_GET_RESERVE(pRaw, dataPos, MND_TOPIC_RESERVE_SIZE, TOPIC_DECODE_OVER); terrno = TSDB_CODE_SUCCESS; @@ -220,11 +225,13 @@ static int32_t mndTopicActionUpdate(SSdb *pSdb, SMqTopicObj *pOldTopic, SMqTopic atomic_exchange_64(&pOldTopic->updateTime, pNewTopic->updateTime); atomic_exchange_32(&pOldTopic->version, pNewTopic->version); - taosWLockLatch(&pOldTopic->lock); + atomic_store_32(&pOldTopic->refConsumerCnt, pNewTopic->refConsumerCnt); + + /*taosWLockLatch(&pOldTopic->lock);*/ // TODO handle update - taosWUnLockLatch(&pOldTopic->lock); + /*taosWUnLockLatch(&pOldTopic->lock);*/ return 0; } @@ -292,6 +299,7 @@ static int32_t mndCreateTopic(SMnode *pMnode, SNodeMsg *pReq, SCMCreateTopicReq topicObj.version = 1; topicObj.sql = strdup(pCreate->sql); topicObj.sqlLen = strlen(pCreate->sql) + 1; + topicObj.refConsumerCnt = 0; if (pCreate->ast && pCreate->ast[0]) { topicObj.ast = strdup(pCreate->ast); @@ -436,15 +444,7 @@ CREATE_TOPIC_OVER: return code; } -static int32_t mndDropTopic(SMnode *pMnode, SNodeMsg *pReq, SMqTopicObj *pTopic) { - // TODO: cannot drop when subscribed - STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_DROP_TOPIC, &pReq->rpcMsg); - if (pTrans == NULL) { - mError("topic:%s, failed to drop since %s", pTopic->name, terrstr()); - return -1; - } - mDebug("trans:%d, used to drop topic:%s", pTrans->id, pTopic->name); - +static int32_t mndDropTopic(SMnode *pMnode, STrans *pTrans, SNodeMsg *pReq, SMqTopicObj *pTopic) { SSdbRaw *pRedoRaw = mndTopicActionEncode(pTopic); if (pRedoRaw == NULL || mndTransAppendRedolog(pTrans, pRedoRaw) != 0) { mError("trans:%d, failed to append redo log since %s", pTrans->id, terrstr()); @@ -465,6 +465,7 @@ static int32_t mndDropTopic(SMnode *pMnode, SNodeMsg *pReq, SMqTopicObj *pTopic) static int32_t mndProcessDropTopicReq(SNodeMsg *pReq) { SMnode *pMnode = pReq->pNode; + SSdb *pSdb = pMnode->pSdb; SMDropTopicReq dropReq = {0}; if (tDeserializeSMDropTopicReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &dropReq) != 0) { @@ -485,10 +486,35 @@ static int32_t mndProcessDropTopicReq(SNodeMsg *pReq) { return -1; } } - // TODO: check ref - int32_t code = mndDropTopic(pMnode, pReq, pTopic); - // TODO: iterate and drop related subscriptions and offsets + // check ref + if (pTopic->refConsumerCnt != 0) { + terrno = TSDB_CODE_MND_TOPIC_SUBSCRIBED; + mError("topic:%s, failed to drop since %s", dropReq.name, terrstr()); + return -1; + } + + STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_DROP_TOPIC, &pReq->rpcMsg); + if (pTrans == NULL) { + mError("topic:%s, failed to drop since %s", pTopic->name, terrstr()); + return -1; + } + + mDebug("trans:%d, used to drop topic:%s", pTrans->id, pTopic->name); + +#if 1 + if (mndDropOffsetByTopic(pMnode, pTrans, dropReq.name) < 0) { + ASSERT(0); + return -1; + } +#endif + + if (mndDropSubByTopic(pMnode, pTrans, dropReq.name) < 0) { + ASSERT(0); + return -1; + } + + int32_t code = mndDropTopic(pMnode, pTrans, pReq, pTopic); mndReleaseTopic(pMnode, pTopic); if (code != 0) { @@ -577,6 +603,15 @@ static int32_t mndRetrieveTopic(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock *pB return numOfRows; } +int32_t mndSetTopicRedoLogs(SMnode *pMnode, STrans *pTrans, SMqTopicObj *pTopic) { + SSdbRaw *pRedoRaw = mndTopicActionEncode(pTopic); + if (pRedoRaw == NULL) return -1; + if (mndTransAppendCommitlog(pTrans, pRedoRaw) != 0) return -1; + if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY) != 0) return -1; + + return 0; +} + static int32_t mndSetDropTopicCommitLogs(SMnode *pMnode, STrans *pTrans, SMqTopicObj *pTopic) { SSdbRaw *pCommitRaw = mndTopicActionEncode(pTopic); if (pCommitRaw == NULL) return -1; diff --git a/source/dnode/vnode/inc/vnode.h b/source/dnode/vnode/inc/vnode.h index 4431a2c48b..f7eac3a9d9 100644 --- a/source/dnode/vnode/inc/vnode.h +++ b/source/dnode/vnode/inc/vnode.h @@ -196,6 +196,8 @@ struct SMetaEntry { STSma *tsma; } smaEntry; }; + + uint8_t *pBuf; }; struct SMetaReader { diff --git a/source/dnode/vnode/src/inc/meta.h b/source/dnode/vnode/src/inc/meta.h index c5ca806829..d3abc95da9 100644 --- a/source/dnode/vnode/src/inc/meta.h +++ b/source/dnode/vnode/src/inc/meta.h @@ -24,7 +24,6 @@ extern "C" { typedef struct SMetaIdx SMetaIdx; typedef struct SMetaDB SMetaDB; -typedef struct SMSmaCursor SMSmaCursor; // metaDebug ================== // clang-format off @@ -114,22 +113,12 @@ typedef struct { int64_t smaUid; } SSmaIdxKey; -#if 1 - -SMSmaCursor* metaOpenSmaCursor(SMeta* pMeta, tb_uid_t uid); -void metaCloseSmaCursor(SMSmaCursor* pSmaCur); -int64_t metaSmaCursorNext(SMSmaCursor* pSmaCur); - #ifndef META_REFACT // SMetaDB int metaOpenDB(SMeta* pMeta); void metaCloseDB(SMeta* pMeta); int metaSaveTableToDB(SMeta* pMeta, STbCfg* pTbCfg, STbDdlH* pHandle); int metaRemoveTableFromDb(SMeta* pMeta, tb_uid_t uid); -int metaSaveSmaToDB(SMeta* pMeta, STSma* pTbCfg); -int metaRemoveSmaFromDb(SMeta* pMeta, int64_t indexUid); -#endif - #endif #ifdef __cplusplus diff --git a/source/dnode/vnode/src/inc/sma.h b/source/dnode/vnode/src/inc/sma.h index 76a30f58dd..a58e1808a6 100644 --- a/source/dnode/vnode/src/inc/sma.h +++ b/source/dnode/vnode/src/inc/sma.h @@ -218,6 +218,11 @@ static int32_t tdInitSmaEnv(SSma *pSma, int8_t smaType, const char *path, SDisk void *tdFreeRSmaInfo(SRSmaInfo *pInfo); +int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t version, const char *pMsg); +int32_t tdUpdateExpiredWindowImpl(SSma *pSma, SSubmitReq *pMsg, int64_t version); +// TODO: This is the basic params, and should wrap the params to a queryHandle. +int32_t tdGetTSmaDataImpl(SSma *pSma, char *pData, int64_t indexUid, TSKEY querySKey, int32_t nMaxResult); + #ifdef __cplusplus } #endif diff --git a/source/dnode/vnode/src/inc/vnodeInt.h b/source/dnode/vnode/src/inc/vnodeInt.h index 2018f0a68c..9727d9df9f 100644 --- a/source/dnode/vnode/src/inc/vnodeInt.h +++ b/source/dnode/vnode/src/inc/vnodeInt.h @@ -89,11 +89,13 @@ STSchema* metaGetTbTSchema(SMeta* pMeta, tb_uid_t uid, int32_t sver); int metaGetTableEntryByName(SMetaReader* pReader, const char* name); int metaGetTbNum(SMeta* pMeta); SMCtbCursor* metaOpenCtbCursor(SMeta* pMeta, tb_uid_t uid); -void metaCloseCtbCurosr(SMCtbCursor* pCtbCur); +void metaCloseCtbCursor(SMCtbCursor* pCtbCur); tb_uid_t metaCtbCursorNext(SMCtbCursor* pCtbCur); -SArray* metaGetSmaTbUids(SMeta* pMeta, bool isDup); -void* metaGetSmaInfoByIndex(SMeta* pMeta, int64_t indexUid, bool isDecode); -STSmaWrapper* metaGetSmaInfoByTable(SMeta* pMeta, tb_uid_t uid); +STSma* metaGetSmaInfoByIndex(SMeta* pMeta, int64_t indexUid); +STSmaWrapper* metaGetSmaInfoByTable(SMeta* pMeta, tb_uid_t uid, bool deepCopy); +SArray* metaGetSmaIdsByTable(SMeta* pMeta, tb_uid_t uid); +SArray* metaGetSmaTbUids(SMeta* pMeta); + int32_t metaCreateTSma(SMeta* pMeta, int64_t version, SSmaCfg* pCfg); int32_t metaDropTSma(SMeta* pMeta, int64_t indexUid); @@ -116,6 +118,7 @@ void tqClose(STQ*); int tqPushMsg(STQ*, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver); int tqCommit(STQ*); int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen); +int32_t tqProcessVgDeleteReq(STQ* pTq, char* msg, int32_t msgLen); int32_t tqProcessTaskExec(STQ* pTq, char* msg, int32_t msgLen, int32_t workerId); int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen); int32_t tqProcessStreamTrigger(STQ* pTq, void* data, int32_t dataLen, int32_t workerId); @@ -126,7 +129,7 @@ int32_t smaOpen(SVnode* pVnode); int32_t smaClose(SSma* pSma); int32_t tdUpdateExpireWindow(SSma* pSma, SSubmitReq* pMsg, int64_t version); -int32_t tdProcessTSmaCreate(SSma* pSma, char* pMsg); +int32_t tdProcessTSmaCreate(SSma* pSma, int64_t version, const char* msg); int32_t tdProcessTSmaInsert(SSma* pSma, int64_t indexUid, const char* msg); int32_t tdProcessRSmaCreate(SSma* pSma, SMeta* pMeta, SVCreateStbReq* pReq, SMsgCb* pMsgCb); diff --git a/source/dnode/vnode/src/meta/metaEntry.c b/source/dnode/vnode/src/meta/metaEntry.c index 2bc0d7517d..84a8957771 100644 --- a/source/dnode/vnode/src/meta/metaEntry.c +++ b/source/dnode/vnode/src/meta/metaEntry.c @@ -69,6 +69,11 @@ int metaDecodeEntry(SDecoder *pCoder, SMetaEntry *pME) { if (tDecodeI32v(pCoder, &pME->ntbEntry.ncid) < 0) return -1; if (tDecodeSSchemaWrapper(pCoder, &pME->ntbEntry.schema) < 0) return -1; } else if (pME->type == TSDB_TSMA_TABLE) { + pME->smaEntry.tsma = taosMemoryCalloc(1, sizeof(STSma)); + if(!pME->smaEntry.tsma) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return -1; + } if (tDecodeTSma(pCoder, pME->smaEntry.tsma) < 0) return -1; } else { ASSERT(0); diff --git a/source/dnode/vnode/src/meta/metaQuery.c b/source/dnode/vnode/src/meta/metaQuery.c index 1e2c94679f..369f16b430 100644 --- a/source/dnode/vnode/src/meta/metaQuery.c +++ b/source/dnode/vnode/src/meta/metaQuery.c @@ -225,7 +225,7 @@ SMCtbCursor *metaOpenCtbCursor(SMeta *pMeta, tb_uid_t uid) { return pCtbCur; } -void metaCloseCtbCurosr(SMCtbCursor *pCtbCur) { +void metaCloseCtbCursor(SMCtbCursor *pCtbCur) { if (pCtbCur) { if (pCtbCur->pMeta) metaULock(pCtbCur->pMeta); if (pCtbCur->pCur) { @@ -291,178 +291,268 @@ STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver) { return pTSchema; } -STSmaWrapper *metaGetSmaInfoByTable(SMeta *pMeta, tb_uid_t uid) { -#if 0 -#ifdef META_TDB_SMA_TEST - STSmaWrapper *pSW = NULL; - - pSW = taosMemoryCalloc(1, sizeof(*pSW)); - if (pSW == NULL) { - return NULL; - } - - SMSmaCursor *pCur = metaOpenSmaCursor(pMeta, uid); - if (pCur == NULL) { - taosMemoryFree(pSW); - return NULL; - } - - void *pBuf = NULL; - SSmaIdxKey *pSmaIdxKey = NULL; - - while (true) { - // TODO: lock during iterate? - if (tdbDbcNext(pCur->pCur, &pCur->pKey, &pCur->kLen, NULL, &pCur->vLen) == 0) { - pSmaIdxKey = pCur->pKey; - ASSERT(pSmaIdxKey != NULL); - - void *pSmaVal = metaGetSmaInfoByIndex(pMeta, pSmaIdxKey->smaUid, false); - - if (pSmaVal == NULL) { - tsdbWarn("no tsma exists for indexUid: %" PRIi64, pSmaIdxKey->smaUid); - continue; - } - - ++pSW->number; - STSma *tptr = (STSma *)taosMemoryRealloc(pSW->tSma, pSW->number * sizeof(STSma)); - if (tptr == NULL) { - tdbFree(pSmaVal); - metaCloseSmaCursor(pCur); - tdDestroyTSmaWrapper(pSW); - taosMemoryFreeClear(pSW); - return NULL; - } - pSW->tSma = tptr; - pBuf = pSmaVal; - if (tDecodeTSma(pBuf, pSW->tSma + pSW->number - 1) == NULL) { - tdbFree(pSmaVal); - metaCloseSmaCursor(pCur); - tdDestroyTSmaWrapper(pSW); - taosMemoryFreeClear(pSW); - return NULL; - } - tdbFree(pSmaVal); - continue; - } - break; - } - - metaCloseSmaCursor(pCur); - - return pSW; - -#endif -#endif - return NULL; -} - int metaGetTbNum(SMeta *pMeta) { // TODO // ASSERT(0); return 0; } -SArray *metaGetSmaTbUids(SMeta *pMeta, bool isDup) { -#if 0 - // TODO - // ASSERT(0); // comment this line to pass CI - // return NULL: -#ifdef META_TDB_SMA_TEST - SArray *pUids = NULL; - SMetaDB *pDB = pMeta->pDB; +typedef struct { + SMeta *pMeta; + TDBC *pCur; + tb_uid_t uid; void *pKey; + void *pVal; + int kLen; + int vLen; +} SMSmaCursor; - // TODO: lock? - SMSmaCursor *pCur = metaOpenSmaCursor(pMeta, 0); - if (pCur == NULL) { +SMSmaCursor *metaOpenSmaCursor(SMeta *pMeta, tb_uid_t uid) { + SMSmaCursor *pSmaCur = NULL; + SSmaIdxKey smaIdxKey; + int ret; + int c; + + pSmaCur = (SMSmaCursor *)taosMemoryCalloc(1, sizeof(*pSmaCur)); + if (pSmaCur == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; return NULL; } - // TODO: lock? - SSmaIdxKey *pSmaIdxKey = NULL; - tb_uid_t uid = 0; - while (true) { - // TODO: lock during iterate? - if (tdbDbcNext(pCur->pCur, &pCur->pKey, &pCur->kLen, NULL, &pCur->vLen) == 0) { - ASSERT(pSmaIdxKey != NULL); - pSmaIdxKey = pCur->pKey; + pSmaCur->pMeta = pMeta; + pSmaCur->uid = uid; + metaRLock(pMeta); - if (pSmaIdxKey->uid == 0 || pSmaIdxKey->uid == uid) { - continue; - } - uid = pSmaIdxKey->uid; - - if (!pUids) { - pUids = taosArrayInit(16, sizeof(tb_uid_t)); - if (!pUids) { - metaCloseSmaCursor(pCur); - return NULL; - } - } - - taosArrayPush(pUids, &uid); - - continue; - } - break; + ret = tdbDbcOpen(pMeta->pSmaIdx, &pSmaCur->pCur, NULL); + if (ret < 0) { + metaULock(pMeta); + taosMemoryFree(pSmaCur); + return NULL; } - metaCloseSmaCursor(pCur); + // move to the suid + smaIdxKey.uid = uid; + smaIdxKey.smaUid = INT64_MIN; + tdbDbcMoveTo(pSmaCur->pCur, &smaIdxKey, sizeof(smaIdxKey), &c); + if (c > 0) { + tdbDbcMoveToNext(pSmaCur->pCur); + } - return pUids; -#endif -#endif + return pSmaCur; +} + +void metaCloseSmaCursor(SMSmaCursor *pSmaCur) { + if (pSmaCur) { + if (pSmaCur->pMeta) metaULock(pSmaCur->pMeta); + if (pSmaCur->pCur) { + tdbDbcClose(pSmaCur->pCur); + + tdbFree(pSmaCur->pKey); + tdbFree(pSmaCur->pVal); + } + + taosMemoryFree(pSmaCur); + } +} + +tb_uid_t metaSmaCursorNext(SMSmaCursor *pSmaCur) { + int ret; + SSmaIdxKey *pSmaIdxKey; + + ret = tdbDbcNext(pSmaCur->pCur, &pSmaCur->pKey, &pSmaCur->kLen, &pSmaCur->pVal, &pSmaCur->vLen); + if (ret < 0) { + return 0; + } + + pSmaIdxKey = pSmaCur->pKey; + if (pSmaIdxKey->uid > pSmaCur->uid) { + return 0; + } + + return pSmaIdxKey->uid; +} + +STSmaWrapper *metaGetSmaInfoByTable(SMeta *pMeta, tb_uid_t uid, bool deepCopy) { + STSmaWrapper *pSW = NULL; + SArray *pSmaIds = NULL; + + if (!(pSmaIds = metaGetSmaIdsByTable(pMeta, uid))) { + return NULL; + } + + pSW = taosMemoryCalloc(1, sizeof(*pSW)); + if (!pSW) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + goto _err; + } + + pSW->number = taosArrayGetSize(pSmaIds); + pSW->tSma = taosMemoryCalloc(pSW->number, sizeof(STSma)); + + if (!pSW->tSma) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + goto _err; + } + + SMSmaCursor *pCur = metaOpenSmaCursor(pMeta, uid); + if (pCur == NULL) { + goto _err; + } + + SMetaReader mr = {0}; + metaReaderInit(&mr, pMeta, 0); + int64_t smaId; + int smaIdx = 0; + STSma *pTSma = NULL; + for (int i = 0; i < pSW->number; ++i) { + smaId = *(tb_uid_t *)taosArrayGet(pSmaIds, i); + if (metaGetTableEntryByUid(&mr, smaId) < 0) { + metaWarn("vgId:%d no entry for tbId: %" PRIi64 ", smaId: %" PRIi64, TD_VID(pMeta->pVnode), uid, smaId); + continue; + } + pTSma = pSW->tSma + smaIdx; + memcpy(pTSma, mr.me.smaEntry.tsma, sizeof(STSma)); + if (deepCopy) { + if (pTSma->exprLen > 0) { + if (!(pTSma->expr = taosMemoryCalloc(1, pTSma->exprLen))) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + goto _err; + } + memcpy((void*)pTSma->expr, mr.me.smaEntry.tsma->expr, pTSma->exprLen); + } + if (pTSma->tagsFilterLen > 0) { + if (!(pTSma->tagsFilter = taosMemoryCalloc(1, pTSma->tagsFilterLen))) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + goto _err; + } + } + memcpy((void*)pTSma->tagsFilter, mr.me.smaEntry.tsma->tagsFilter, pTSma->tagsFilterLen); + } else { + pTSma->exprLen = 0; + pTSma->expr = NULL; + pTSma->tagsFilterLen = 0; + pTSma->tagsFilter = NULL; + } + + ++smaIdx; + } + + if (smaIdx <= 0) goto _err; + pSW->number = smaIdx; + + metaReaderClear(&mr); + taosArrayDestroy(pSmaIds); + metaCloseSmaCursor(pCur); + return pSW; +_err: + metaReaderClear(&mr); + taosArrayDestroy(pSmaIds); + metaCloseSmaCursor(pCur); + tdFreeTSmaWrapper(pSW, deepCopy); return NULL; } -void *metaGetSmaInfoByIndex(SMeta *pMeta, int64_t indexUid, bool isDecode) { -#if 0 - // TODO - // ASSERT(0); - // return NULL; -#ifdef META_TDB_SMA_TEST - SMetaDB *pDB = pMeta->pDB; - void *pKey = NULL; - void *pVal = NULL; - int kLen = 0; - int vLen = 0; - int ret = -1; - - // Set key - pKey = (void *)&indexUid; - kLen = sizeof(indexUid); - - // Query - ret = tdbDbGet(pDB->pSmaDB, pKey, kLen, &pVal, &vLen); - if (ret != 0 || !pVal) { +STSma *metaGetSmaInfoByIndex(SMeta *pMeta, int64_t indexUid) { + STSma *pTSma = NULL; + SMetaReader mr = {0}; + metaReaderInit(&mr, pMeta, 0); + if (metaGetTableEntryByUid(&mr, indexUid) < 0) { + metaWarn("vgId:%d failed to get table entry for smaId: %" PRIi64, TD_VID(pMeta->pVnode), indexUid); + metaReaderClear(&mr); + return NULL; + } + pTSma = (STSma *)taosMemoryMalloc(sizeof(STSma)); + if (!pTSma) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + metaReaderClear(&mr); return NULL; } - if (!isDecode) { - // return raw value - return pVal; - } + memcpy(pTSma, mr.me.smaEntry.tsma, sizeof(STSma)); - // Decode - STSma *pCfg = (STSma *)taosMemoryCalloc(1, sizeof(STSma)); - if (pCfg == NULL) { - taosMemoryFree(pVal); + metaReaderClear(&mr); + return pTSma; +} + +SArray *metaGetSmaIdsByTable(SMeta *pMeta, tb_uid_t uid) { + SArray *pUids = NULL; + SSmaIdxKey *pSmaIdxKey = NULL; + + SMSmaCursor *pCur = metaOpenSmaCursor(pMeta, uid); + if (!pCur) { return NULL; } - void *pBuf = pVal; - if (tDecodeTSma(pBuf, pCfg) == NULL) { - tdDestroyTSma(pCfg); - taosMemoryFree(pCfg); - tdbFree(pVal); + while (1) { + tb_uid_t id = metaSmaCursorNext(pCur); + if (id == 0) { + break; + } + + if (!pUids) { + pUids = taosArrayInit(16, sizeof(tb_uid_t)); + if (!pUids) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + metaCloseSmaCursor(pCur); + return NULL; + } + } + + pSmaIdxKey = (SSmaIdxKey *)pCur->pKey; + + if (taosArrayPush(pUids, &pSmaIdxKey->smaUid) < 0) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + metaCloseSmaCursor(pCur); + taosArrayDestroy(pUids); + return NULL; + } + } + + metaCloseSmaCursor(pCur); + return pUids; +} + +SArray *metaGetSmaTbUids(SMeta *pMeta) { + SArray *pUids = NULL; + SSmaIdxKey *pSmaIdxKey = NULL; + tb_uid_t lastUid = 0; + + SMSmaCursor *pCur = metaOpenSmaCursor(pMeta, 0); + if (!pCur) { return NULL; } - tdbFree(pVal); - return pCfg; -#endif -#endif - return NULL; + while (1) { + tb_uid_t uid = metaSmaCursorNext(pCur); + if (uid == 0) { + break; + } + + if (lastUid == uid) { + continue; + } + + lastUid = uid; + + if (!pUids) { + pUids = taosArrayInit(16, sizeof(tb_uid_t)); + if (!pUids) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + metaCloseSmaCursor(pCur); + return NULL; + } + } + + if (taosArrayPush(pUids, &uid) < 0) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + metaCloseSmaCursor(pCur); + taosArrayDestroy(pUids); + return NULL; + } + } + + metaCloseSmaCursor(pCur); + return pUids; } #endif diff --git a/source/dnode/vnode/src/meta/metaSma.c b/source/dnode/vnode/src/meta/metaSma.c index 8ce7ea5895..d0a3541152 100644 --- a/source/dnode/vnode/src/meta/metaSma.c +++ b/source/dnode/vnode/src/meta/metaSma.c @@ -16,7 +16,7 @@ #include "meta.h" static int metaHandleSmaEntry(SMeta *pMeta, const SMetaEntry *pME); -static int metaSaveSmaToDB(SMeta *pMeta, const SMetaEntry *pME); +static int metaSaveSmaToDB(SMeta *pMeta, const SMetaEntry *pME); int32_t metaCreateTSma(SMeta *pMeta, int64_t version, SSmaCfg *pCfg) { // TODO: Validate the cfg @@ -81,55 +81,6 @@ int32_t metaDropTSma(SMeta *pMeta, int64_t indexUid) { return TSDB_CODE_SUCCESS; } -// static int metaSaveSmaToDB(SMeta *pMeta, STSma *pSmaCfg) { -// int32_t ret = 0; -// void *pBuf = NULL, *qBuf = NULL; -// void *key = {0}, *val = {0}; - -// // save sma info -// int32_t len = tEncodeTSma(NULL, pSmaCfg); -// pBuf = taosMemoryCalloc(1, len); -// if (pBuf == NULL) { -// terrno = TSDB_CODE_OUT_OF_MEMORY; -// return -1; -// } - -// key = (void *)&pSmaCfg->indexUid; -// qBuf = pBuf; -// tEncodeTSma(&qBuf, pSmaCfg); -// val = pBuf; - -// int32_t kLen = sizeof(pSmaCfg->indexUid); -// int32_t vLen = POINTER_DISTANCE(qBuf, pBuf); - -// ret = tdbDbInsert(pMeta->pTbDb, key, kLen, val, vLen, &pMeta->txn); -// if (ret < 0) { -// taosMemoryFreeClear(pBuf); -// return -1; -// } - -// // add sma idx -// SSmaIdxKey smaIdxKey; -// smaIdxKey.uid = pSmaCfg->tableUid; -// smaIdxKey.smaUid = pSmaCfg->indexUid; -// key = &smaIdxKey; -// kLen = sizeof(smaIdxKey); -// val = NULL; -// vLen = 0; - -// ret = tdbDbInsert(pMeta->pSmaIdx, key, kLen, val, vLen, &pMeta->txn); -// if (ret < 0) { -// taosMemoryFreeClear(pBuf); -// return -1; -// } - -// // release -// taosMemoryFreeClear(pBuf); - -// return 0; -// } - - static int metaSaveSmaToDB(SMeta *pMeta, const SMetaEntry *pME) { STbDbKey tbDbKey; void *pKey = NULL; @@ -182,6 +133,10 @@ static int metaUpdateUidIdx(SMeta *pMeta, const SMetaEntry *pME) { return tdbDbInsert(pMeta->pUidIdx, &pME->uid, sizeof(tb_uid_t), &pME->version, sizeof(int64_t), &pMeta->txn); } +static int metaUpdateNameIdx(SMeta *pMeta, const SMetaEntry *pME) { + return tdbDbInsert(pMeta->pNameIdx, pME->name, strlen(pME->name) + 1, &pME->uid, sizeof(tb_uid_t), &pMeta->txn); +} + static int metaUpdateSmaIdx(SMeta *pMeta, const SMetaEntry *pME) { SSmaIdxKey smaIdxKey = {.uid = pME->smaEntry.tsma->tableUid, .smaUid = pME->smaEntry.tsma->indexUid}; @@ -194,9 +149,13 @@ static int metaHandleSmaEntry(SMeta *pMeta, const SMetaEntry *pME) { // save to table.db if (metaSaveSmaToDB(pMeta, pME) < 0) goto _err; - // // update uid.idx + // update uid.idx if (metaUpdateUidIdx(pMeta, pME) < 0) goto _err; + // update name.idx + if (metaUpdateNameIdx(pMeta, pME) < 0) goto _err; + + // update sma.idx if (metaUpdateSmaIdx(pMeta, pME) < 0) goto _err; metaULock(pMeta); diff --git a/source/dnode/vnode/src/meta/metaTable.c b/source/dnode/vnode/src/meta/metaTable.c index e61064fe67..9064085b2b 100644 --- a/source/dnode/vnode/src/meta/metaTable.c +++ b/source/dnode/vnode/src/meta/metaTable.c @@ -420,7 +420,8 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl // get table entry SDecoder dc = {0}; tDecoderInit(&dc, pData, nData); - metaDecodeEntry(&dc, &entry); + ret = metaDecodeEntry(&dc, &entry); + ASSERT(ret == 0); if (entry.type != TSDB_NORMAL_TABLE) { terrno = TSDB_CODE_VND_INVALID_TABLE_ACTION; @@ -468,11 +469,11 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl goto _err; } pSchema->sver++; - pSchema->nCols--; tlen = (pSchema->nCols - iCol - 1) * sizeof(SSchema); if (tlen) { memmove(pColumn, pColumn + 1, tlen); } + pSchema->nCols--; break; case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES: if (pColumn == NULL) { @@ -498,6 +499,18 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl entry.version = version; + // do actual write + metaWLock(pMeta); + + // save to table db + metaSaveToTbDb(pMeta, &entry); + + tdbDbcUpsert(pUidIdxc, &entry.uid, sizeof(tb_uid_t), &version, sizeof(version), 0); + + metaSaveToSkmDb(pMeta, &entry); + + metaULock(pMeta); + tDecoderClear(&dc); tdbDbcClose(pTbDbc); tdbDbcClose(pUidIdxc); @@ -511,8 +524,128 @@ _err: } static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pAlterTbReq) { - // TODO + SMetaEntry ctbEntry = {0}; + SMetaEntry stbEntry = {0}; + void *pVal = NULL; + int nVal = 0; + int ret; + int c; + tb_uid_t uid; + int64_t oversion; + const void *pData = NULL; + int nData = 0; + + // search name index + ret = tdbDbGet(pMeta->pNameIdx, pAlterTbReq->tbName, strlen(pAlterTbReq->tbName) + 1, &pVal, &nVal); + if (ret < 0) { + terrno = TSDB_CODE_VND_TABLE_NOT_EXIST; + return -1; + } + + uid = *(tb_uid_t *)pVal; + tdbFree(pVal); + pVal = NULL; + + // search uid index + TDBC *pUidIdxc = NULL; + + tdbDbcOpen(pMeta->pUidIdx, &pUidIdxc, &pMeta->txn); + tdbDbcMoveTo(pUidIdxc, &uid, sizeof(uid), &c); + ASSERT(c == 0); + + tdbDbcGet(pUidIdxc, NULL, NULL, &pData, &nData); + oversion = *(int64_t *)pData; + + // search table.db + TDBC *pTbDbc = NULL; + SDecoder dc = {0}; + + /* get ctbEntry */ + tdbDbcOpen(pMeta->pTbDb, &pTbDbc, &pMeta->txn); + tdbDbcMoveTo(pTbDbc, &((STbDbKey){.uid = uid, .version = oversion}), sizeof(STbDbKey), &c); + ASSERT(c == 0); + tdbDbcGet(pTbDbc, NULL, NULL, &pData, &nData); + + ctbEntry.pBuf = taosMemoryMalloc(nData); + memcpy(ctbEntry.pBuf, pData, nData); + tDecoderInit(&dc, ctbEntry.pBuf, nData); + metaDecodeEntry(&dc, &ctbEntry); + tDecoderClear(&dc); + + /* get stbEntry*/ + tdbDbGet(pMeta->pUidIdx, &ctbEntry.ctbEntry.suid, sizeof(tb_uid_t), &pVal, &nVal); + tdbDbGet(pMeta->pTbDb, &((STbDbKey){.uid = ctbEntry.ctbEntry.suid, .version = *(int64_t *)pVal}), sizeof(STbDbKey), + (void **)&stbEntry.pBuf, &nVal); + tdbFree(pVal); + tDecoderInit(&dc, stbEntry.pBuf, nVal); + metaDecodeEntry(&dc, &stbEntry); + tDecoderClear(&dc); + + SSchemaWrapper *pTagSchema = &stbEntry.stbEntry.schemaTag; + SSchema *pColumn = NULL; + int32_t iCol = 0; + for (;;) { + pColumn = NULL; + + if (iCol >= pTagSchema->nCols) break; + pColumn = &pTagSchema->pSchema[iCol]; + + if (strcmp(pColumn->name, pAlterTbReq->tagName) == 0) break; + iCol++; + } + + if (pColumn == NULL) { + terrno = TSDB_CODE_VND_TABLE_COL_NOT_EXISTS; + goto _err; + } + + if (iCol == 0) { + // TODO : need to update tag index + } + + ctbEntry.version = version; + SKVRowBuilder kvrb = {0}; + const SKVRow pOldTag = (const SKVRow)ctbEntry.ctbEntry.pTags; + SKVRow pNewTag = NULL; + + tdInitKVRowBuilder(&kvrb); + for (int32_t i = 0; i < pTagSchema->nCols; i++) { + SSchema *pCol = &pTagSchema->pSchema[i]; + if (iCol == i) { + tdAddColToKVRow(&kvrb, pCol->colId, pAlterTbReq->pTagVal, pAlterTbReq->nTagVal); + } else { + void *p = tdGetKVRowValOfCol(pOldTag, pCol->colId); + if (p) { + if (IS_VAR_DATA_TYPE(pCol->type)) { + tdAddColToKVRow(&kvrb, pCol->colId, p, varDataTLen(p)); + } else { + tdAddColToKVRow(&kvrb, pCol->colId, p, pCol->bytes); + } + } + } + } + + ctbEntry.ctbEntry.pTags = tdGetKVRowFromBuilder(&kvrb); + tdDestroyKVRowBuilder(&kvrb); + + // save to table.db + metaSaveToTbDb(pMeta, &ctbEntry); + + // save to uid.idx + tdbDbUpsert(pMeta->pUidIdx, &ctbEntry.uid, sizeof(tb_uid_t), &version, sizeof(version), &pMeta->txn); + + if (ctbEntry.pBuf) taosMemoryFree(ctbEntry.pBuf); + if (stbEntry.pBuf) tdbFree(stbEntry.pBuf); + tdbDbcClose(pTbDbc); + tdbDbcClose(pUidIdxc); return 0; + +_err: + if (ctbEntry.pBuf) taosMemoryFree(ctbEntry.pBuf); + if (stbEntry.pBuf) tdbFree(stbEntry.pBuf); + tdbDbcClose(pTbDbc); + tdbDbcClose(pUidIdxc); + return -1; } static int metaUpdateTableOptions(SMeta *pMeta, int64_t version, SVAlterTbReq *pAlterTbReq) { diff --git a/source/dnode/vnode/src/sma/sma.c b/source/dnode/vnode/src/sma/sma.c index 2c54e10087..0e7ce385a1 100644 --- a/source/dnode/vnode/src/sma/sma.c +++ b/source/dnode/vnode/src/sma/sma.c @@ -27,4 +27,28 @@ int32_t tdProcessTSmaInsert(SSma* pSma, int64_t indexUid, const char* msg) { return code; } +int32_t tdProcessTSmaCreate(SSma* pSma, int64_t version, const char* msg) { + int32_t code = TSDB_CODE_SUCCESS; + if ((code = tdProcessTSmaCreateImpl(pSma, version, msg)) < 0) { + smaWarn("vgId:%d create tsma failed since %s", SMA_VID(pSma), tstrerror(terrno)); + } + // TODO: destroy SSDataBlocks(msg) + return code; +} + +int32_t tdUpdateExpireWindow(SSma* pSma, SSubmitReq* pMsg, int64_t version) { + int32_t code = TSDB_CODE_SUCCESS; + if ((code = tdUpdateExpiredWindowImpl(pSma, pMsg, version)) < 0) { + smaWarn("vgId:%d update expired sma window failed since %s", SMA_VID(pSma), tstrerror(terrno)); + } + return code; +} + +int32_t tdGetTSmaData(SSma* pSma, char* pData, int64_t indexUid, TSKEY querySKey, int32_t nMaxResult) { + int32_t code = TSDB_CODE_SUCCESS; + if ((code = tdGetTSmaDataImpl(pSma, pData, indexUid, querySKey, nMaxResult)) < 0) { + smaWarn("vgId:%d get tSma data failed since %s", SMA_VID(pSma), tstrerror(terrno)); + } + return code; +} diff --git a/source/dnode/vnode/src/sma/smaEnv.c b/source/dnode/vnode/src/sma/smaEnv.c index c02276f5fe..8285b74e50 100644 --- a/source/dnode/vnode/src/sma/smaEnv.c +++ b/source/dnode/vnode/src/sma/smaEnv.c @@ -122,12 +122,10 @@ static void poolFree(void *arg, void *ptr) { } int32_t tdInitSma(SSma *pSma) { - // tSma - int32_t numOfTSma = taosArrayGetSize(metaGetSmaTbUids(SMA_META(pSma), false)); + int32_t numOfTSma = taosArrayGetSize(metaGetSmaTbUids(SMA_META(pSma))); if (numOfTSma > 0) { atomic_store_16(&SMA_TSMA_NUM(pSma), (int16_t)numOfTSma); } - // TODO: rSma return TSDB_CODE_SUCCESS; } diff --git a/source/dnode/vnode/src/sma/smaRollup.c b/source/dnode/vnode/src/sma/smaRollup.c index f9cb5a1a09..88af049d0b 100644 --- a/source/dnode/vnode/src/sma/smaRollup.c +++ b/source/dnode/vnode/src/sma/smaRollup.c @@ -443,7 +443,7 @@ static int32_t tdExecuteRSma(SSma *pSma, const void *pMsg, int32_t inputType, tb if (inputType == STREAM_DATA_TYPE_SUBMIT_BLOCK) { // TODO: use the proper schema instead of 0, and cache STSchema in cache - STSchema *pTSchema = metaGetTbTSchema(SMA_META(pSma), suid, 0); + STSchema *pTSchema = metaGetTbTSchema(SMA_META(pSma), suid, 1); if (!pTSchema) { terrno = TSDB_CODE_TDB_IVD_TB_SCHEMA_VERSION; return TSDB_CODE_FAILED; diff --git a/source/dnode/vnode/src/sma/smaTimeRange.c b/source/dnode/vnode/src/sma/smaTimeRange.c index b04885c5f0..357cf710a2 100644 --- a/source/dnode/vnode/src/sma/smaTimeRange.c +++ b/source/dnode/vnode/src/sma/smaTimeRange.c @@ -70,15 +70,15 @@ static bool tdSetAndOpenTSmaFile(STSmaReadH *pReadH, TSKEY *queryKey); static int32_t tdInsertTSmaBlocks(STSmaWriteH *pSmaH, void *smaKey, int32_t keyLen, void *pData, int32_t dataLen, TXN *txn); // expired window -static int32_t tdUpdateExpiredWindowImpl(SSma *pSma, SSubmitReq *pMsg, int64_t version); + + static int32_t tdSetExpiredWindow(SSma *pSma, SHashObj *pItemsHash, int64_t indexUid, int64_t winSKey, int64_t version); static int32_t tdResetExpiredWindow(SSma *pSma, SSmaStat *pStat, int64_t indexUid, TSKEY skey); static int32_t tdDropTSmaDataImpl(SSma *pSma, int64_t indexUid); // read data -// TODO: This is the basic params, and should wrap the params to a queryHandle. -static int32_t tdGetTSmaDataImpl(SSma *pSma, char *pData, int64_t indexUid, TSKEY querySKey, int32_t nMaxResult); + // implementation @@ -713,7 +713,7 @@ static int32_t tdDropTSmaDataImpl(SSma *pSma, int64_t indexUid) { * @param nMaxResult The query invoker should control the nMaxResult need to return to avoid OOM. * @return int32_t */ -static int32_t tdGetTSmaDataImpl(SSma *pSma, char *pData, int64_t indexUid, TSKEY querySKey, int32_t nMaxResult) { +int32_t tdGetTSmaDataImpl(SSma *pSma, char *pData, int64_t indexUid, TSKEY querySKey, int32_t nMaxResult) { SSmaEnv *pEnv = atomic_load_ptr(&SMA_TSMA_ENV(pSma)); SSmaStat *pStat = NULL; @@ -834,35 +834,15 @@ static int32_t tdGetTSmaDataImpl(SSma *pSma, char *pData, int64_t indexUid, TSKE return TSDB_CODE_SUCCESS; } -int32_t tdProcessTSmaCreate(SSma *pSma, char *pMsg) { - #if 0 - SSmaCfg vCreateSmaReq = {0}; - if (!tDeserializeSVCreateTSmaReq(pMsg, &vCreateSmaReq)) { - terrno = TSDB_CODE_OUT_OF_MEMORY; - smaWarn("vgId:%d tsma create msg received but deserialize failed since %s", SMA_VID(pSma), terrstr(terrno)); - return -1; - } +int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t version, const char *pMsg) { + SSmaCfg *pCfg = (SSmaCfg *)pMsg; - smaDebug("vgId:%d tsma create msg %s:%" PRIi64 " for table %" PRIi64 " received", SMA_VID(pSma), - vCreateSmaReq.tSma.indexName, vCreateSmaReq.tSma.indexUid, vCreateSmaReq.tSma.tableUid); - - // record current timezone of server side - vCreateSmaReq.tSma.timezoneInt = tsTimezone; - - if (metaCreateTSma(SMA_META(pSma), &vCreateSmaReq) < 0) { - // TODO: handle error - smaWarn("vgId:%d tsma %s:%" PRIi64 " create failed for table %" PRIi64 " since %s", SMA_VID(pSma), - vCreateSmaReq.tSma.indexName, vCreateSmaReq.tSma.indexUid, vCreateSmaReq.tSma.tableUid, terrstr(terrno)); - tdDestroyTSma(&vCreateSmaReq.tSma); + if (metaCreateTSma(SMA_META(pSma), version, pCfg) < 0) { return -1; } tdTSmaAdd(pSma, 1); - - tdDestroyTSma(&vCreateSmaReq.tSma); - // TODO: return directly or go on follow steps? -#endif - return TSDB_CODE_SUCCESS; + return 0; } int32_t tdDropTSma(SSma *pSma, char *pMsg) { @@ -930,7 +910,7 @@ static int32_t tdSetExpiredWindow(SSma *pSma, SHashObj *pItemsHash, int64_t inde } // cache smaMeta - STSma *pTSma = metaGetSmaInfoByIndex(SMA_META(pSma), indexUid, true); + STSma *pTSma = metaGetSmaInfoByIndex(SMA_META(pSma), indexUid); if (!pTSma) { terrno = TSDB_CODE_TDB_NO_SMA_INDEX_IN_META; taosHashCleanup(pItem->expiredWindows); @@ -1031,25 +1011,25 @@ int32_t tdUpdateExpiredWindowImpl(SSma *pSma, SSubmitReq *pMsg, int64_t version) SSubmitBlkIter blkIter = {0}; if (tInitSubmitBlkIter(&msgIter, pBlock, &blkIter) < 0) { - pSW = tdFreeTSmaWrapper(pSW); + pSW = tdFreeTSmaWrapper(pSW, false); break; } while (true) { STSRow *row = tGetSubmitBlkNext(&blkIter); if (!row) { - tdFreeTSmaWrapper(pSW); + pSW = tdFreeTSmaWrapper(pSW, false); break; } - if (!pSW || (pTSma->tableUid != pBlock->suid)) { + if (!pSW || (pTSma->tableUid != msgIter.suid)) { if (pSW) { - pSW = tdFreeTSmaWrapper(pSW); + pSW = tdFreeTSmaWrapper(pSW, false); } - if (!(pSW = metaGetSmaInfoByTable(SMA_META(pSma), pBlock->suid))) { + if (!(pSW = metaGetSmaInfoByTable(SMA_META(pSma), msgIter.suid, false))) { break; } if ((pSW->number) <= 0 || !pSW->tSma) { - pSW = tdFreeTSmaWrapper(pSW); + pSW = tdFreeTSmaWrapper(pSW, false); break; } @@ -1068,6 +1048,7 @@ int32_t tdUpdateExpiredWindowImpl(SSma *pSma, SSubmitReq *pMsg, int64_t version) if (lastWinSKey != winSKey) { lastWinSKey = winSKey; if (tdSetExpiredWindow(pSma, pItemsHash, pTSma->indexUid, winSKey, version) < 0) { + pSW = tdFreeTSmaWrapper(pSW, false); tdUnRefSmaStat(pSma, pStat); return TSDB_CODE_FAILED; } @@ -1083,21 +1064,3 @@ int32_t tdUpdateExpiredWindowImpl(SSma *pSma, SSubmitReq *pMsg, int64_t version) return TSDB_CODE_SUCCESS; } - -int32_t tdUpdateExpireWindow(SSma *pSma, SSubmitReq *pMsg, int64_t version) { - int32_t code = TSDB_CODE_SUCCESS; - if ((code = tdUpdateExpiredWindowImpl(pSma, pMsg, version)) < 0) { - smaWarn("vgId:%d update expired sma window failed since %s", SMA_VID(pSma), tstrerror(terrno)); - } - return code; -} - -int32_t tdGetTSmaData(SSma *pSma, char *pData, int64_t indexUid, TSKEY querySKey, int32_t nMaxResult) { - int32_t code = TSDB_CODE_SUCCESS; - if ((code = tdGetTSmaDataImpl(pSma, pData, indexUid, querySKey, nMaxResult)) < 0) { - smaWarn("vgId:%d get tSma data failed since %s", SMA_VID(pSma), tstrerror(terrno)); - } - return code; -} - - diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index 873db62dd8..bc9893b8a0 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -861,6 +861,14 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) { } #endif +int32_t tqProcessVgDeleteReq(STQ* pTq, char* msg, int32_t msgLen) { + SMqVDeleteReq* pReq = (SMqVDeleteReq*)msg; + + int32_t code = taosHashRemove(pTq->execs, pReq->subKey, strlen(pReq->subKey)); + ASSERT(code == 0); + return 0; +} + // TODO: persist meta into tdb int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) { SMqRebVgReq req = {0}; @@ -1087,35 +1095,6 @@ int32_t tqProcessStreamTrigger2(STQ* pTq, SSubmitReq* pReq, int64_t ver) { } int32_t tqProcessTaskExec2(STQ* pTq, char* msg, int32_t msgLen) { - SStreamTaskExecReq req = {0}; - tDecodeSStreamTaskExecReq(msg, &req); - int32_t taskId = req.taskId; - - SStreamTask* pTask = taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t)); - ASSERT(pTask); - ASSERT(pTask->inputType == TASK_INPUT_TYPE__DATA_BLOCK); - - // enqueue - int32_t inputStatus = streamEnqueueDataBlk(pTask, (SStreamDataBlock*)req.data); - if (inputStatus == TASK_INPUT_STATUS__BLOCKED) { - // TODO rsp blocked - return 0; - } - - // try exec - int8_t execStatus = atomic_val_compare_exchange_8(&pTask->status, TASK_STATUS__IDLE, TASK_STATUS__EXECUTING); - if (execStatus == TASK_STATUS__IDLE) { - if (streamTaskRun(pTask) < 0) { - atomic_store_8(&pTask->status, TASK_STATUS__CLOSING); - - goto FAIL; - } - } else if (execStatus == TASK_STATUS__EXECUTING) { - return 0; - } - - // TODO rsp success + // return 0; -FAIL: - return -1; } diff --git a/source/dnode/vnode/src/tsdb/tsdbRead.c b/source/dnode/vnode/src/tsdb/tsdbRead.c index 55fe8a3945..254d452cb3 100644 --- a/source/dnode/vnode/src/tsdb/tsdbRead.c +++ b/source/dnode/vnode/src/tsdb/tsdbRead.c @@ -2760,7 +2760,7 @@ static int32_t getAllTableList(SMeta* pMeta, uint64_t uid, SArray* list) { taosArrayPush(list, &info); } - metaCloseCtbCurosr(pCur); + metaCloseCtbCursor(pCur); return TSDB_CODE_SUCCESS; } @@ -3055,7 +3055,8 @@ static bool loadDataBlockFromTableSeq(STsdbReadHandle* pTsdbReadHandle) { bool tsdbNextDataBlock(tsdbReaderT pHandle) { STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)pHandle; - for (int32_t i = 0; i < taosArrayGetSize(pTsdbReadHandle->pColumns); ++i) { + size_t numOfCols = taosArrayGetSize(pTsdbReadHandle->pColumns); + for (int32_t i = 0; i < numOfCols; ++i) { SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i); colInfoDataCleanup(pColInfo, pTsdbReadHandle->outputCapacity); } diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c index 43d68fc9fd..0f7c7df841 100644 --- a/source/dnode/vnode/src/vnd/vnodeSvr.c +++ b/source/dnode/vnode/src/vnd/vnodeSvr.c @@ -101,6 +101,11 @@ int vnodeProcessWriteReq(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRpcMsg // TODO: handle error } break; + case TDMT_VND_MQ_VG_DELETE: + if (tqProcessVgDeleteReq(pVnode->pTq, pMsg->pCont, pMsg->contLen) < 0) { + // TODO: handle error + } + break; case TDMT_VND_TASK_DEPLOY: { if (tqProcessTaskDeploy(pVnode->pTq, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)), pMsg->contLen - sizeof(SMsgHead)) < 0) { @@ -199,7 +204,7 @@ void smaHandleRes(void *pVnode, int64_t smaId, const SArray *data) { int vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) { int32_t ret = TAOS_SYNC_PROPOSE_OTHER_ERROR; - + if (syncEnvIsStart()) { SSyncNode *pSyncNode = syncNodeAcquire(pVnode->sync); assert(pSyncNode != NULL); @@ -462,7 +467,11 @@ _exit: static int vnodeProcessAlterTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) { SVAlterTbReq vAlterTbReq = {0}; + SVAlterTbRsp vAlterTbRsp = {0}; SDecoder dc = {0}; + int rcode = 0; + int ret; + SEncoder ec = {0}; pRsp->msgType = TDMT_VND_ALTER_TABLE_RSP; pRsp->pCont = NULL; @@ -473,19 +482,27 @@ static int vnodeProcessAlterTbReq(SVnode *pVnode, int64_t version, void *pReq, i // decode if (tDecodeSVAlterTbReq(&dc, &vAlterTbReq) < 0) { - pRsp->code = TSDB_CODE_INVALID_MSG; + vAlterTbRsp.code = TSDB_CODE_INVALID_MSG; tDecoderClear(&dc); - return -1; + rcode = -1; + goto _exit; } // process if (metaAlterTable(pVnode->pMeta, version, &vAlterTbReq) < 0) { - pRsp->code = terrno; + vAlterTbRsp.code = TSDB_CODE_INVALID_MSG; tDecoderClear(&dc); - return -1; + rcode = -1; + goto _exit; } - tDecoderClear(&dc); + +_exit: + tEncodeSize(tEncodeSVAlterTbRsp, &vAlterTbRsp, pRsp->contLen, ret); + pRsp->pCont = rpcMallocCont(pRsp->contLen); + tEncoderInit(&ec, pRsp->pCont, pRsp->contLen); + tEncodeSVAlterTbRsp(&ec, &vAlterTbRsp); + tEncoderClear(&ec); return 0; } @@ -711,15 +728,22 @@ static int vnodeProcessCreateTSmaReq(SVnode *pVnode, int64_t version, void *pReq goto _err; } - if (metaCreateTSma(pVnode->pMeta, version, &req) < 0) { + // record current timezone of server side + req.timezoneInt = tsTimezone; + + if (tdProcessTSmaCreate(pVnode->pSma, version, (const char *)&req) < 0) { pRsp->code = terrno; goto _err; } tDecoderClear(&coder); + vDebug("vgId:%d success to create tsma %s:%" PRIi64 " for table %" PRIi64, TD_VID(pVnode), req.indexName, + req.indexUid, req.tableUid); return 0; _err: tDecoderClear(&coder); + vError("vgId:%d failed to create tsma %s:%" PRIi64 " for table %" PRIi64 " since %s", TD_VID(pVnode), req.indexName, + req.indexUid, req.tableUid, terrstr(terrno)); return -1; } diff --git a/source/libs/executor/inc/executil.h b/source/libs/executor/inc/executil.h index e62729a051..edc21626f5 100644 --- a/source/libs/executor/inc/executil.h +++ b/source/libs/executor/inc/executil.h @@ -139,7 +139,7 @@ typedef struct { int32_t colId; } SStddevInterResult; -void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, bool sortGroupResult); +void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, int32_t order); void initMultiResInfoFromArrayList(SGroupResInfo* pGroupResInfo, SArray* pArrayList); void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo); diff --git a/source/libs/executor/src/executil.c b/source/libs/executor/src/executil.c index 763dcef790..14cff488ec 100644 --- a/source/libs/executor/src/executil.c +++ b/source/libs/executor/src/executil.c @@ -184,7 +184,7 @@ void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo) { pGroupResInfo->index = 0; } -static int32_t resultrowCompar1(const void* p1, const void* p2) { +static int32_t resultrowComparAsc(const void* p1, const void* p2) { SResKeyPos* pp1 = *(SResKeyPos**) p1; SResKeyPos* pp2 = *(SResKeyPos**) p2; @@ -202,7 +202,11 @@ static int32_t resultrowCompar1(const void* p1, const void* p2) { } } -void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, bool sortGroupResult) { +static int32_t resultrowComparDesc(const void* p1, const void* p2) { + return resultrowComparAsc(p2, p1); +} + +void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, int32_t order) { if (pGroupResInfo->pRows != NULL) { taosArrayDestroy(pGroupResInfo->pRows); } @@ -224,8 +228,9 @@ void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, boo taosArrayPush(pGroupResInfo->pRows, &p); } - if (sortGroupResult) { - qsort(pGroupResInfo->pRows->pData, taosArrayGetSize(pGroupResInfo->pRows), POINTER_BYTES, resultrowCompar1); + if (order == TSDB_ORDER_ASC || order == TSDB_ORDER_DESC) { + __compar_fn_t fn = (order == TSDB_ORDER_ASC)? resultrowComparAsc:resultrowComparDesc; + qsort(pGroupResInfo->pRows->pData, taosArrayGetSize(pGroupResInfo->pRows), POINTER_BYTES, fn); } pGroupResInfo->index = 0; diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 1cfabf2975..688901226c 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -3707,7 +3707,7 @@ static int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) { finalizeMultiTupleQueryResult(pOperator->numOfExprs, pAggInfo->aggSup.pResultBuf, &pAggInfo->binfo.resultRowInfo, pAggInfo->binfo.rowCellInfoOffset); - initGroupedResultInfo(&pAggInfo->groupResInfo, pAggInfo->aggSup.pResultRowHashTable, false); + initGroupedResultInfo(&pAggInfo->groupResInfo, pAggInfo->aggSup.pResultRowHashTable, 0); OPTR_SET_OPENED(pOperator); return TSDB_CODE_SUCCESS; } diff --git a/source/libs/executor/src/groupoperator.c b/source/libs/executor/src/groupoperator.c index 483ac67e5e..76a773a095 100644 --- a/source/libs/executor/src/groupoperator.c +++ b/source/libs/executor/src/groupoperator.c @@ -314,7 +314,7 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator) { // } blockDataEnsureCapacity(pRes, pOperator->resultInfo.capacity); - initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, false); + initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, 0); while(1) { doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf); diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 08539206a6..1a862c94f3 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -102,7 +102,7 @@ static void getNextTimeWindow(SInterval* pInterval, STimeWindow* tw, int32_t ord tw->ekey -= 1; } -static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockInfo) { +static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockInfo, int32_t order) { STimeWindow w = {0}; // 0 by default, which means it is not a interval operator of the upstream operator. @@ -110,13 +110,7 @@ static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockIn return false; } - // todo handle the time range case - TSKEY sk = INT64_MIN; - TSKEY ek = INT64_MAX; - // TSKEY sk = MIN(pQueryAttr->window.skey, pQueryAttr->window.ekey); - // TSKEY ek = MAX(pQueryAttr->window.skey, pQueryAttr->window.ekey); - - if (true) { + if (order == TSDB_ORDER_ASC) { getAlignQueryTimeWindow(pInterval, pInterval->precision, pBlockInfo->window.skey, &w); assert(w.ekey >= pBlockInfo->window.skey); @@ -124,8 +118,8 @@ static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockIn return true; } - while (1) { // todo handle the desc order scan case - getNextTimeWindow(pInterval, &w, TSDB_ORDER_ASC); + while (1) { + getNextTimeWindow(pInterval, &w, order); if (w.skey > pBlockInfo->window.ekey) { break; } @@ -136,24 +130,24 @@ static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockIn } } } else { - // getAlignQueryTimeWindow(pQueryAttr, pBlockInfo->window.ekey, sk, ek, &w); - // assert(w.skey <= pBlockInfo->window.ekey); - // - // if (w.skey > pBlockInfo->window.skey) { - // return true; - // } - // - // while(1) { - // getNextTimeWindow(pQueryAttr, &w); - // if (w.ekey < pBlockInfo->window.skey) { - // break; - // } - // - // assert(w.skey < pBlockInfo->window.skey); - // if (w.ekey < pBlockInfo->window.ekey && w.ekey >= pBlockInfo->window.skey) { - // return true; - // } - // } + getAlignQueryTimeWindow(pInterval, pInterval->precision, pBlockInfo->window.ekey, &w); + assert(w.skey <= pBlockInfo->window.ekey); + + if (w.skey > pBlockInfo->window.skey) { + return true; + } + + while(1) { + getNextTimeWindow(pInterval, &w, order); + if (w.ekey < pBlockInfo->window.skey) { + break; + } + + assert(w.skey < pBlockInfo->window.skey); + if (w.ekey < pBlockInfo->window.ekey && w.ekey >= pBlockInfo->window.skey) { + return true; + } + } } return false; @@ -172,7 +166,8 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableSca pCost->totalRows += pBlock->info.rows; *status = pInfo->dataBlockLoadFlag; - if (pTableScanInfo->pFilterNode != NULL || overlapWithTimeWindow(&pTableScanInfo->interval, &pBlock->info)) { + if (pTableScanInfo->pFilterNode != NULL || + overlapWithTimeWindow(&pTableScanInfo->interval, &pBlock->info, pTableScanInfo->cond.order)) { (*status) = FUNC_DATA_REQUIRED_DATA_LOAD; } @@ -188,6 +183,13 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableSca qDebug("%s data block skipped, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo), pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows); pCost->skipBlocks += 1; + + // clear all data in pBlock that are set when handing the previous block + for(int32_t i = 0; i < pBlockInfo->numOfCols; ++i) { + SColumnInfoData* pcol = taosArrayGet(pBlock->pDataBlock, i); + pcol->pData = NULL; + } + return TSDB_CODE_SUCCESS; } else if (*status == FUNC_DATA_REQUIRED_STATIS_LOAD) { pCost->loadBlockStatis += 1; @@ -466,6 +468,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, } pInfo->scanInfo = (SScanInfo){.numOfAsc = pTableScanNode->scanSeq[0], .numOfDesc = pTableScanNode->scanSeq[1]}; +// pInfo->scanInfo = (SScanInfo){.numOfAsc = 0, .numOfDesc = 1}; // for debug purpose pInfo->readHandle = *readHandle; pInfo->interval = extractIntervalInfo(pTableScanNode); diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c index 79e675e2df..479ce394b1 100644 --- a/source/libs/executor/src/timewindowoperator.c +++ b/source/libs/executor/src/timewindowoperator.c @@ -54,8 +54,8 @@ static TSKEY getStartTsKey(STimeWindow* win, const TSKEY* tsCols, int32_t rows, if (tsCols == NULL) { ts = ascQuery ? win->skey : win->ekey; } else { - int32_t offset = ascQuery ? 0 : rows - 1; - ts = tsCols[offset]; +// int32_t offset = ascQuery ? 0 : rows - 1; + ts = tsCols[0]; } return ts; @@ -172,14 +172,22 @@ static FORCE_INLINE int32_t getForwardStepsInBlock(int32_t numOfRows, __block_se } } } else { - int32_t end = searchFn((char*)pData, pos + 1, ekey, order); + int32_t end = searchFn((char*)&pData[pos], numOfRows - pos, ekey, order); if (end >= 0) { - forwardStep = pos - end; + forwardStep = end; - if (pData[end] == ekey) { + if (pData[end + pos] == ekey) { forwardStep += 1; } } +// int32_t end = searchFn((char*)pData, pos + 1, ekey, order); +// if (end >= 0) { +// forwardStep = pos - end; +// +// if (pData[end] == ekey) { +// forwardStep += 1; +// } +// } } assert(forwardStep >= 0); @@ -203,17 +211,25 @@ int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order) { if (order == TSDB_ORDER_DESC) { // find the first position which is smaller than the key while (1) { - if (key >= keyList[lastPos]) return lastPos; - if (key == keyList[firstPos]) return firstPos; - if (key < keyList[firstPos]) return firstPos - 1; + if (key >= keyList[firstPos]) return firstPos; + if (key == keyList[lastPos]) return lastPos; + + if (key < keyList[lastPos]) { + lastPos += 1; + if (lastPos >= num) { + return -1; + } else { + return lastPos; + } + } numOfRows = lastPos - firstPos + 1; midPos = (numOfRows >> 1) + firstPos; if (key < keyList[midPos]) { - lastPos = midPos - 1; - } else if (key > keyList[midPos]) { firstPos = midPos + 1; + } else if (key > keyList[midPos]) { + lastPos = midPos - 1; } else { break; } @@ -273,12 +289,12 @@ int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimary if (ekey > pDataBlockInfo->window.skey && pPrimaryColumn) { num = getForwardStepsInBlock(pDataBlockInfo->rows, searchFn, ekey, startPos, order, pPrimaryColumn); if (item != NULL) { - item->lastKey = pPrimaryColumn[startPos - (num - 1)] + step; + item->lastKey = pPrimaryColumn[startPos + (num - 1)] + step; } } else { - num = startPos + 1; + num = pDataBlockInfo->rows - startPos; if (item != NULL) { - item->lastKey = pDataBlockInfo->window.skey + step; + item->lastKey = pDataBlockInfo->window.ekey + step; } } } @@ -470,20 +486,17 @@ static int32_t getNextQualifiedWindow(SInterval* pInterval, STimeWindow* pNext, return -1; } - TSKEY startKey = ascQuery ? pNext->skey : pNext->ekey; + TSKEY skey = ascQuery ? pNext->skey : pNext->ekey; int32_t startPos = 0; // tumbling time window query, a special case of sliding time window query if (pInterval->sliding == pInterval->interval && prevPosition != -1) { - int32_t factor = GET_FORWARD_DIRECTION_FACTOR(order); - startPos = prevPosition + factor; + startPos = prevPosition + 1; } else { - if (startKey <= pDataBlockInfo->window.skey && ascQuery) { + if ((skey <= pDataBlockInfo->window.skey && ascQuery) || (skey >= pDataBlockInfo->window.ekey && !ascQuery)) { startPos = 0; - } else if (startKey >= pDataBlockInfo->window.ekey && !ascQuery) { - startPos = pDataBlockInfo->rows - 1; } else { - startPos = binarySearchForKey((char*)primaryKeys, pDataBlockInfo->rows, startKey, order); + startPos = binarySearchForKey((char*)primaryKeys, pDataBlockInfo->rows, skey, order); } } @@ -608,7 +621,7 @@ static void saveDataBlockLastRow(char** pRow, SArray* pDataBlock, int32_t rowInd } static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResultRowInfo, SSDataBlock* pSDataBlock, - int32_t tableGroupId) { + uint64_t tableGroupId) { SIntervalAggOperatorInfo* pInfo = (SIntervalAggOperatorInfo*)pOperatorInfo->info; SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo; @@ -620,7 +633,7 @@ static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pRe } int32_t step = 1; - bool ascScan = true; + bool ascScan = (pInfo->order == TSDB_ORDER_ASC); // int32_t prevIndex = pResultRowInfo->curPos; @@ -630,7 +643,7 @@ static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pRe tsCols = (int64_t*)pColDataInfo->pData; } - int32_t startPos = ascScan ? 0 : (pSDataBlock->info.rows - 1); + int32_t startPos = 0; TSKEY ts = getStartTsKey(&pSDataBlock->info.window, tsCols, pSDataBlock->info.rows, ascScan); STimeWindow win = getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts, &pInfo->interval, @@ -654,9 +667,10 @@ static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pRe } int32_t forwardStep = 0; - TSKEY ekey = win.ekey; + TSKEY ekey = ascScan? win.ekey:win.skey; forwardStep = - getNumOfRowsInTimeWindow(&pSDataBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, TSDB_ORDER_ASC); + getNumOfRowsInTimeWindow(&pSDataBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, pInfo->order); + ASSERT(forwardStep > 0); // prev time window not interpolation yet. // int32_t curIndex = pResultRowInfo->curPos; @@ -731,9 +745,9 @@ static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pRe taosArrayPush(pUpdated, &pos); } - ekey = nextWin.ekey; // reviseWindowEkey(pQueryAttr, &nextWin); + ekey = ascScan? nextWin.ekey:nextWin.skey; forwardStep = - getNumOfRowsInTimeWindow(&pSDataBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, TSDB_ORDER_ASC); + getNumOfRowsInTimeWindow(&pSDataBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, pInfo->order); // window start(end) key interpolation doWindowBorderInterpolation(pOperatorInfo, pSDataBlock, pInfo->binfo.pCtx, pResult, &nextWin, startPos, forwardStep, @@ -761,7 +775,8 @@ static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) { SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SIntervalAggOperatorInfo* pInfo = pOperator->info; - int32_t order = TSDB_ORDER_ASC; + int32_t scanFlag = MAIN_SCAN; + SOperatorInfo* downstream = pOperator->pDownstream[0]; while (1) { @@ -773,8 +788,10 @@ static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) { break; } + getTableScanInfo(pOperator, &pInfo->order, &scanFlag); + // the pDataBlock are always the same one, no need to call this again - setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, order, MAIN_SCAN, true); + setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, pInfo->order, scanFlag, true); STableQueryInfo* pTableQueryInfo = pInfo->pCurrent; setIntervalQueryRange(pTableQueryInfo, pBlock->info.window.skey, &pTaskInfo->window); @@ -800,7 +817,7 @@ static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) { finalizeMultiTupleQueryResult(pOperator->numOfExprs, pInfo->aggSup.pResultBuf, &pInfo->binfo.resultRowInfo, pInfo->binfo.rowCellInfoOffset); - initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, true); + initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, pInfo->order); OPTR_SET_OPENED(pOperator); return TSDB_CODE_SUCCESS; } @@ -945,7 +962,7 @@ static SSDataBlock* doStateWindowAgg(SOperatorInfo* pOperator) { finalizeMultiTupleQueryResult(pOperator->numOfExprs, pInfo->aggSup.pResultBuf, &pBInfo->resultRowInfo, pBInfo->rowCellInfoOffset); - initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, true); + initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, TSDB_ORDER_ASC); blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity); doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf); if (pBInfo->pRes->info.rows == 0 || !hasRemainDataInCurrentGroup(&pInfo->groupResInfo)) { @@ -1070,6 +1087,7 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) { doClearWindows(pInfo, pOperator->numOfExprs, pBlock); continue; } + pInfo->order = TSDB_ORDER_ASC; pUpdated = hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, 0); } @@ -1119,7 +1137,6 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pInfo->order = TSDB_ORDER_ASC; pInfo->interval = *pInterval; - // pInfo->execModel = OPTR_EXEC_MODEL_STREAM; pInfo->execModel = pTaskInfo->execModel; pInfo->win = pTaskInfo->window; pInfo->twAggSup = *pTwAggSupp; @@ -1338,7 +1355,7 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) { finalizeMultiTupleQueryResult(pOperator->numOfExprs, pInfo->aggSup.pResultBuf, &pBInfo->resultRowInfo, pBInfo->rowCellInfoOffset); - initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, true); + initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, TSDB_ORDER_ASC); blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity); doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf); if (pBInfo->pRes->info.rows == 0 || !hasRemainDataInCurrentGroup(&pInfo->groupResInfo)) { diff --git a/source/libs/function/inc/builtinsimpl.h b/source/libs/function/inc/builtinsimpl.h index c25d74911c..99313675a5 100644 --- a/source/libs/function/inc/builtinsimpl.h +++ b/source/libs/function/inc/builtinsimpl.h @@ -90,6 +90,10 @@ bool histogramFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultIn int32_t histogramFunction(SqlFunctionCtx* pCtx); int32_t histogramFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock); +bool getHLLFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv); +int32_t hllFunction(SqlFunctionCtx* pCtx); +int32_t hllFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock); + bool getStateFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv); bool stateFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo); int32_t stateCountFunction(SqlFunctionCtx* pCtx); diff --git a/source/libs/function/src/builtins.c b/source/libs/function/src/builtins.c index e41e3c7c39..48165fdd99 100644 --- a/source/libs/function/src/builtins.c +++ b/source/libs/function/src/builtins.c @@ -263,6 +263,21 @@ static int32_t translateHistogram(SFunctionNode* pFunc, char* pErrBuf, int32_t l return TSDB_CODE_SUCCESS; } +static int32_t translateHLL(SFunctionNode* pFunc, char* pErrBuf, int32_t len) { + if (1 != LIST_LENGTH(pFunc->pParameterList)) { + return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName); + } + + SNode* pPara = nodesListGetNode(pFunc->pParameterList, 0); + if (QUERY_NODE_COLUMN != nodeType(pPara)) { + return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR, + "The input parameter of HYPERLOGLOG function can only be column"); + } + + pFunc->node.resType = (SDataType){.bytes = tDataTypes[TSDB_DATA_TYPE_UBIGINT].bytes, .type = TSDB_DATA_TYPE_UBIGINT}; + return TSDB_CODE_SUCCESS; +} + static int32_t translateStateCount(SFunctionNode* pFunc, char* pErrBuf, int32_t len) { if (3 != LIST_LENGTH(pFunc->pParameterList)) { return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName); @@ -829,6 +844,16 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { .processFunc = histogramFunction, .finalizeFunc = histogramFinalize }, + { + .name = "hyperloglog", + .type = FUNCTION_TYPE_HYPERLOGLOG, + .classification = FUNC_MGT_AGG_FUNC, + .translateFunc = translateHLL, + .getEnvFunc = getHLLFuncEnv, + .initFunc = functionSetup, + .processFunc = hllFunction, + .finalizeFunc = hllFinalize + }, { .name = "state_count", .type = FUNCTION_TYPE_STATE_COUNT, diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c index 479a11ca4a..7ff3b6fb05 100644 --- a/source/libs/function/src/builtinsimpl.c +++ b/source/libs/function/src/builtinsimpl.c @@ -28,6 +28,12 @@ #define TAIL_MAX_POINTS_NUM 100 #define TAIL_MAX_OFFSET 100 +#define HLL_BUCKET_BITS 14 // The bits of the bucket +#define HLL_DATA_BITS (64-HLL_BUCKET_BITS) +#define HLL_BUCKETS (1<numOfRes; } +bool getHLLFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) { + pEnv->calcMemSize = sizeof(SHLLInfo); + return true; +} + +static uint8_t hllCountNum(void* data, int32_t bytes, int32_t *buk) { + uint64_t hash = MurmurHash3_64(data, bytes); + int32_t index = hash & HLL_BUCKET_MASK; + hash >>= HLL_BUCKET_BITS; + hash |= ((uint64_t)1 << HLL_DATA_BITS); + uint64_t bit = 1; + uint8_t count = 1; + while((hash & bit) == 0) { + count++; + bit <<= 1; + } + *buk = index; + return count; +} + +static void hllBucketHisto(uint8_t *buckets, int32_t* bucketHisto) { + uint64_t *word = (uint64_t*) buckets; + uint8_t *bytes; + + for (int32_t j = 0; j < HLL_BUCKETS>>3; j++) { + if (*word == 0) { + bucketHisto[0] += 8; + } else { + bytes = (uint8_t*) word; + bucketHisto[bytes[0]]++; + bucketHisto[bytes[1]]++; + bucketHisto[bytes[2]]++; + bucketHisto[bytes[3]]++; + bucketHisto[bytes[4]]++; + bucketHisto[bytes[5]]++; + bucketHisto[bytes[6]]++; + bucketHisto[bytes[7]]++; + } + word++; + } +} +static double hllTau(double x) { + if (x == 0. || x == 1.) return 0.; + double zPrime; + double y = 1.0; + double z = 1 - x; + do { + x = sqrt(x); + zPrime = z; + y *= 0.5; + z -= pow(1 - x, 2)*y; + } while(zPrime != z); + return z / 3; +} + +static double hllSigma(double x) { + if (x == 1.0) return INFINITY; + double zPrime; + double y = 1; + double z = x; + do { + x *= x; + zPrime = z; + z += x * y; + y += y; + } while(zPrime != z); + return z; +} + +// estimate the cardinality, the algorithm refer this paper: "New cardinality estimation algorithms for HyperLogLog sketches" +static uint64_t hllCountCnt(uint8_t *buckets) { + double m = HLL_BUCKETS; + int32_t buckethisto[64] = {0}; + hllBucketHisto(buckets,buckethisto); + + double z = m * hllTau((m-buckethisto[HLL_DATA_BITS+1])/(double)m); + for (int j = HLL_DATA_BITS; j >= 1; --j) { + z += buckethisto[j]; + z *= 0.5; + } + z += m * hllSigma(buckethisto[0]/(double)m); + double E = (double)llroundl(HLL_ALPHA_INF*m*m/z); + + return (uint64_t) E; +} + + +int32_t hllFunction(SqlFunctionCtx *pCtx) { + SHLLInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx)); + + SInputColumnInfoData* pInput = &pCtx->input; + SColumnInfoData* pCol = pInput->pData[0]; + + int32_t type = pCol->info.type; + int32_t bytes = pCol->info.bytes; + + int32_t start = pInput->startRowIndex; + int32_t numOfRows = pInput->numOfRows; + + int32_t numOfElems = 0; + for (int32_t i = start; i < numOfRows + start; ++i) { + if (pCol->hasNull && colDataIsNull_s(pCol, i)) { + continue; + } + + numOfElems++; + + char* data = colDataGetData(pCol, i); + if (IS_VAR_DATA_TYPE(type)) { + bytes = varDataLen(data); + data = varDataVal(data); + } + + int32_t index = 0; + uint8_t count = hllCountNum(data, bytes, &index); + uint8_t oldcount = pInfo->buckets[index]; + if (count > oldcount) { + pInfo->buckets[index] = count; + } + + } + + SET_VAL(GET_RES_INFO(pCtx), numOfElems, 1); + return TSDB_CODE_SUCCESS; +} + +int32_t hllFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) { + SHLLInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx)); + + pInfo->result = hllCountCnt(pInfo->buckets); + + return functionFinalize(pCtx, pBlock); +} + bool getStateFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) { pEnv->calcMemSize = sizeof(SStateInfo); return true; diff --git a/source/libs/function/src/tudf.c b/source/libs/function/src/tudf.c index 388ec28b76..03a3891a4c 100644 --- a/source/libs/function/src/tudf.c +++ b/source/libs/function/src/tudf.c @@ -795,7 +795,6 @@ int32_t convertScalarParamToDataBlock(SScalarParam *input, int32_t numOfCols, SS } output->info.hasVarCol = hasVarCol; - //TODO: free the array output->pDataBlock output->pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData)); for (int32_t i = 0; i < numOfCols; ++i) { taosArrayPush(output->pDataBlock, (input + i)->columnData); @@ -809,8 +808,12 @@ int32_t convertDataBlockToScalarParm(SSDataBlock *input, SScalarParam *output) { return -1; } output->numOfRows = input->info.rows; - //TODO: memory - output->columnData = taosArrayGet(input->pDataBlock, 0); + + output->columnData = taosMemoryMalloc(sizeof(SColumnInfoData)); + memcpy(output->columnData, + taosArrayGet(input->pDataBlock, 0), + sizeof(SColumnInfoData)); + return 0; } @@ -833,7 +836,7 @@ int32_t udfcGetUdfTaskResultFromUvTask(SClientUdfTask *task, SClientUvTaskNode * fnDebug("udfc get uv task result. task: %p, uvTask: %p", task, uvTask); if (uvTask->type == UV_TASK_REQ_RSP) { if (uvTask->rspBuf.base != NULL) { - SUdfResponse rsp; + SUdfResponse rsp = {0}; void* buf = decodeUdfResponse(uvTask->rspBuf.base, &rsp); assert(uvTask->rspBuf.len == POINTER_DISTANCE(buf, uvTask->rspBuf.base)); task->errCode = rsp.code; @@ -1427,7 +1430,10 @@ int32_t doCallUdfScalarFunc(UdfcFuncHandle handle, SScalarParam *input, int32_t int32_t err = callUdf(handle, callType, &inputBlock, NULL, NULL, &resultBlock, NULL); if (err == 0) { convertDataBlockToScalarParm(&resultBlock, output); + taosArrayDestroy(resultBlock.pDataBlock); } + + taosArrayDestroy(inputBlock.pDataBlock); return err; } @@ -1508,16 +1514,15 @@ int32_t doTeardownUdf(UdfcFuncHandle handle) { udfcRunUdfUvTask(task, UV_TASK_REQ_RSP); - SUdfTeardownResponse *rsp = &task->_teardown.rsp; int32_t err = task->errCode; udfcRunUdfUvTask(task, UV_TASK_DISCONNECT); + fnInfo("tear down udf. udf name: %s, udf func handle: %p", session->udfName, handle); + taosMemoryFree(task->session); taosMemoryFree(task); - fnInfo("tear down udf. udf name: %s, udf func handle: %p", session->udfName, handle); - return err; } @@ -1564,6 +1569,7 @@ bool udfAggInit(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResult } udfRes->interResNum = buf.numOfResult; memcpy(udfRes->interResBuf, buf.buf, buf.bufLen); + freeUdfInterBuf(&buf); return true; } @@ -1621,7 +1627,7 @@ int32_t udfAggProcess(struct SqlFunctionCtx *pCtx) { blockDataDestroy(inputBlock); taosArrayDestroy(tempBlock.pDataBlock); - taosMemoryFree(newState.buf); + freeUdfInterBuf(&newState); return udfCode; } @@ -1650,6 +1656,8 @@ int32_t udfAggFinalize(struct SqlFunctionCtx *pCtx, SSDataBlock* pBlock) { GET_RES_INFO(pCtx)->numOfRes = udfRes->finalResNum; } + freeUdfInterBuf(&resultBuf); + int32_t numOfResults = functionFinalizeWithResultBuf(pCtx, pBlock, udfRes->finalResBuf); releaseUdfFuncHandle(pCtx->udfName); return udfCallCode == 0 ? numOfResults : udfCallCode; diff --git a/source/libs/function/src/udfd.c b/source/libs/function/src/udfd.c index 706bf28be0..e644ea6172 100644 --- a/source/libs/function/src/udfd.c +++ b/source/libs/function/src/udfd.c @@ -96,10 +96,14 @@ int32_t udfdFillUdfInfoFromMNode(void *clientRpc, char *udfName, SUdf *udf); int32_t udfdLoadUdf(char *udfName, SUdf *udf) { strcpy(udf->name, udfName); + int32_t err = 0; + err = udfdFillUdfInfoFromMNode(global.clientRpc, udf->name, udf); + if (err != 0) { + fnError("can not retrieve udf from mnode. udf name %s", udfName); + return TSDB_CODE_UDF_LOAD_UDF_FAILURE; + } - udfdFillUdfInfoFromMNode(global.clientRpc, udf->name, udf); - //strcpy(udf->path, "/home/slzhou/TDengine/debug/build/lib/libudf1.so"); - int err = uv_dlopen(udf->path, &udf->lib); + err = uv_dlopen(udf->path, &udf->lib); if (err != 0) { fnError("can not load library %s. error: %s", udf->path, uv_strerror(err)); return TSDB_CODE_UDF_LOAD_UDF_FAILURE; @@ -142,7 +146,7 @@ int32_t udfdLoadUdf(char *udfName, SUdf *udf) { void udfdProcessSetupRequest(SUvUdfWork* uvUdf, SUdfRequest* request) { // TODO: tracable id from client. connect, setup, call, teardown - fnInfo("%" PRId64 " setup request. udf name: %s", request->seqNum, request->setup.udfName); + fnInfo( "setup request. seq num: %" PRId64 ", udf name: %s", request->seqNum, request->setup.udfName); SUdfSetupRequest *setup = &request->setup; int32_t code = TSDB_CODE_SUCCESS; SUdf *udf = NULL; @@ -276,7 +280,7 @@ void udfdProcessCallRequest(SUvUdfWork *uvUdf, SUdfRequest *request) { void udfdProcessTeardownRequest(SUvUdfWork* uvUdf, SUdfRequest* request) { SUdfTeardownRequest *teardown = &request->teardown; - fnInfo("teardown. %" PRId64 "handle:%" PRIx64, request->seqNum, teardown->udfHandle); + fnInfo("teardown. seq number: %" PRId64 ", handle:%" PRIx64, request->seqNum, teardown->udfHandle); SUdfcFuncHandle *handle = (SUdfcFuncHandle *)(teardown->udfHandle); SUdf *udf = handle->udf; bool unloadUdf = false; @@ -800,17 +804,11 @@ static int32_t udfdRun() { global.udfsHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK); uv_mutex_init(&global.udfsMutex); - if (udfdUvInit() != 0) { - fnError("uv init failure"); - return -2; - } - fnInfo("start the udfd"); int code = uv_run(global.loop, UV_RUN_DEFAULT); fnInfo("udfd stopped. result: %s, code: %d", uv_err_name(code), code); int codeClose = uv_loop_close(global.loop); fnDebug("uv loop close. result: %s", uv_err_name(codeClose)); - removeListeningPipe(); uv_mutex_destroy(&global.udfsMutex); taosHashCleanup(global.udfsHash); return 0; @@ -853,8 +851,14 @@ int main(int argc, char *argv[]) { return -4; } + if (udfdUvInit() != 0) { + fnError("uv init failure"); + return -5; + } + udfdRun(); - udfdCloseClientRpc(); + removeListeningPipe(); + udfdCloseClientRpc(); } diff --git a/source/libs/function/test/runUdf.c b/source/libs/function/test/runUdf.c index d7c539e5c2..9fe9269a3f 100644 --- a/source/libs/function/test/runUdf.c +++ b/source/libs/function/test/runUdf.c @@ -34,20 +34,13 @@ static int32_t initLog() { return taosCreateLog(logName, 1, configDir, NULL, NULL, NULL, NULL, 0); } -int main(int argc, char *argv[]) { - parseArgs(argc, argv); - initLog(); - if (taosInitCfg(configDir, NULL, NULL, NULL, NULL, 0) != 0) { - fnError("failed to start since read config error"); - return -1; - } - - udfcOpen(); - uv_sleep(1000); - +int scalarFuncTest() { UdfcFuncHandle handle; - doSetupUdf("udf1", &handle); + if (doSetupUdf("udf1", &handle) != 0) { + fnError("setup udf failure"); + return -1; + } SSDataBlock block = {0}; SSDataBlock *pBlock = █ @@ -74,11 +67,78 @@ int main(int argc, char *argv[]) { input.columnData = taosArrayGet(pBlock->pDataBlock, 0); SScalarParam output = {0}; doCallUdfScalarFunc(handle, &input, 1, &output); - + taosArrayDestroy(pBlock->pDataBlock); SColumnInfoData *col = output.columnData; for (int32_t i = 0; i < output.numOfRows; ++i) { fprintf(stderr, "%d\t%d\n", i, *(int32_t *)(col->pData + i * sizeof(int32_t))); } + + colDataDestroy(output.columnData); + taosMemoryFree(output.columnData); + doTeardownUdf(handle); + + return 0; +} + +int aggregateFuncTest() { + UdfcFuncHandle handle; + + if (doSetupUdf("udf2", &handle) != 0) { + fnError("setup udf failure"); + return -1; + } + + SSDataBlock block = {0}; + SSDataBlock *pBlock = █ + pBlock->pDataBlock = taosArrayInit(1, sizeof(SColumnInfoData)); + pBlock->info.numOfCols = 1; + pBlock->info.rows = 4; + char data[16] = {0}; + char bitmap[4] = {0}; + for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) { + SColumnInfoData colInfo = {0}; + colInfo.info.type = TSDB_DATA_TYPE_INT; + colInfo.info.bytes = sizeof(int32_t); + colInfo.info.colId = 1; + colInfo.pData = data; + colInfo.nullbitmap = bitmap; + for (int32_t j = 0; j < pBlock->info.rows; ++j) { + colDataAppendInt32(&colInfo, j, &j); + } + taosArrayPush(pBlock->pDataBlock, &colInfo); + } + + SUdfInterBuf buf = {0}; + SUdfInterBuf newBuf = {0}; + SUdfInterBuf resultBuf = {0}; + doCallUdfAggInit(handle, &buf); + doCallUdfAggProcess(handle, pBlock, &buf, &newBuf); + taosArrayDestroy(pBlock->pDataBlock); + + doCallUdfAggFinalize(handle, &newBuf, &resultBuf); + fprintf(stderr, "agg result: %f\n", *(double*)resultBuf.buf); + + freeUdfInterBuf(&buf); + freeUdfInterBuf(&newBuf); + freeUdfInterBuf(&resultBuf); + doTeardownUdf(handle); + + return 0; +} + +int main(int argc, char *argv[]) { + parseArgs(argc, argv); + initLog(); + if (taosInitCfg(configDir, NULL, NULL, NULL, NULL, 0) != 0) { + fnError("failed to start since read config error"); + return -1; + } + + udfcOpen(); + uv_sleep(1000); + + scalarFuncTest(); + aggregateFuncTest(); udfcClose(); } diff --git a/source/libs/stream/src/tstream.c b/source/libs/stream/src/tstream.c index 812874dafb..bdfb82071b 100644 --- a/source/libs/stream/src/tstream.c +++ b/source/libs/stream/src/tstream.c @@ -35,7 +35,7 @@ void* streamDataBlockDecode(const void* buf, SStreamDataBlock* pInput) { return (void*)buf; } -static int32_t streamBuildDispatchMsg(SStreamTask* pTask, SArray* data, SRpcMsg* pMsg, SEpSet** ppEpSet) { +static int32_t streamBuildExecMsg(SStreamTask* pTask, SArray* data, SRpcMsg* pMsg, SEpSet** ppEpSet) { SStreamTaskExecReq req = { .streamId = pTask->streamId, .data = data, @@ -107,7 +107,7 @@ static int32_t streamShuffleDispatch(SStreamTask* pTask, SMsgCb* pMsgCb, SHashOb SArray* pData = *(SArray**)pIter; SRpcMsg dispatchMsg = {0}; SEpSet* pEpSet; - if (streamBuildDispatchMsg(pTask, pData, &dispatchMsg, &pEpSet) < 0) { + if (streamBuildExecMsg(pTask, pData, &dispatchMsg, &pEpSet) < 0) { ASSERT(0); return -1; } @@ -133,7 +133,7 @@ int32_t streamEnqueueDataBlk(SStreamTask* pTask, SStreamDataBlock* input) { return inputStatus; } -int32_t streamTaskExecImpl(SStreamTask* pTask, void* data, SArray* pRes) { +static int32_t streamTaskExecImpl(SStreamTask* pTask, void* data, SArray* pRes) { void* exec = pTask->exec.runners[0].executor; // set input @@ -265,87 +265,42 @@ FAIL: return -1; } -int32_t streamTaskDispatchDown(SStreamTask* pTask, SMsgCb* pMsgCb) { - // - return 0; -} - -int32_t streamTaskSink(SStreamTask* pTask) { - // - return 0; -} - -int32_t streamTaskProcessInputReq(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDataBlock* pBlock, SRpcMsg* pRsp) { - // 1. handle input - // 1.1 enqueue - taosWriteQitem(pTask->inputQ, pBlock); - // 1.2 calc back pressure - // 1.3 rsp by input status - int8_t inputStatus = atomic_load_8(&pTask->inputStatus); - SStreamDispatchRsp* pCont = rpcMallocCont(sizeof(SStreamDispatchRsp)); - pCont->status = inputStatus; - pRsp->pCont = pCont; - pRsp->contLen = sizeof(SStreamDispatchRsp); - tmsgSendRsp(pRsp); - // 2. try exec - // 2.1. idle: exec - // 2.2. executing: return - // 2.3. closing: keep trying +int32_t streamTaskSink(SStreamTask* pTask, SMsgCb* pMsgCb) { + bool firstRun = 1; while (1) { - int8_t execStatus = atomic_val_compare_exchange_8(&pTask->status, TASK_STATUS__IDLE, TASK_STATUS__EXECUTING); - if (execStatus == TASK_STATUS__IDLE) { - void* exec = pTask->exec.runners[0].executor; - SArray* pRes = taosArrayInit(0, sizeof(void*)); - const SArray* blocks = pBlock->blocks; - qSetMultiStreamInput(exec, blocks->pData, blocks->size, STREAM_DATA_TYPE_SSDATA_BLOCK); - while (1) { - SSDataBlock* output; - uint64_t ts = 0; - if (qExecTask(exec, &output, &ts) < 0) { - ASSERT(false); - } - if (output == NULL) break; - taosArrayPush(pRes, &output); - } - // TODO: wrap destroy block - taosArrayDestroyP(pBlock->blocks, (FDelete)blockDataDestroy); - - if (taosArrayGetSize(pRes) != 0) { - SArray** resQ = taosAllocateQitem(sizeof(void**), DEF_QITEM); - *resQ = pRes; - taosWriteQitem(pTask->outputQ, resQ); - } - - } else if (execStatus == TASK_STATUS__CLOSING) { - continue; - } else if (execStatus == TASK_STATUS__EXECUTING) - break; - else { - ASSERT(0); + SStreamDataBlock* pBlock = NULL; + if (!firstRun) { + taosReadAllQitems(pTask->outputQ, pTask->outputQAll); + } + taosGetQitem(pTask->outputQAll, (void**)&pBlock); + if (pBlock == NULL) { + if (firstRun) { + firstRun = 0; + continue; + } else { + break; + } } - } - // 3. handle output - // 3.1 check and set status - // 3.2 dispatch / sink - STaosQall* qall = taosAllocateQall(); - taosReadAllQitems(pTask->outputQ, qall); - SArray** ppRes = NULL; - while (1) { - taosGetQitem(qall, (void**)&ppRes); - if (ppRes == NULL) break; - SArray* pRes = *ppRes; + SArray* pRes = pBlock->blocks; + + // sink if (pTask->sinkType == TASK_SINK__TABLE) { - pTask->tbSink.tbSinkFunc(pTask, pTask->tbSink.vnode, pBlock->sourceVer, pRes); + // blockDebugShowData(pRes); + pTask->tbSink.tbSinkFunc(pTask, pTask->tbSink.vnode, 0, pRes); } else if (pTask->sinkType == TASK_SINK__SMA) { pTask->smaSink.smaSink(pTask->ahandle, pTask->smaSink.smaId, pRes); + // + } else if (pTask->sinkType == TASK_SINK__FETCH) { + // } else { + ASSERT(pTask->sinkType == TASK_SINK__NONE); } // dispatch if (pTask->dispatchType == TASK_DISPATCH__INPLACE) { SRpcMsg dispatchMsg = {0}; - if (streamBuildDispatchMsg(pTask, pRes, &dispatchMsg, NULL) < 0) { + if (streamBuildExecMsg(pTask, pRes, &dispatchMsg, NULL) < 0) { ASSERT(0); return -1; } @@ -366,7 +321,7 @@ int32_t streamTaskProcessInputReq(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDat } else if (pTask->dispatchType == TASK_DISPATCH__FIXED) { SRpcMsg dispatchMsg = {0}; SEpSet* pEpSet = NULL; - if (streamBuildDispatchMsg(pTask, pRes, &dispatchMsg, &pEpSet) < 0) { + if (streamBuildExecMsg(pTask, pRes, &dispatchMsg, &pEpSet) < 0) { ASSERT(0); return -1; } @@ -401,12 +356,53 @@ int32_t streamTaskProcessInputReq(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDat ASSERT(pTask->dispatchType == TASK_DISPATCH__NONE); } } - // return 0; } -int32_t streamTaskProcessDispatchRsp(SStreamTask* pTask, char* msg, int32_t msgLen) { - // +int32_t streamTaskEnqueue(SStreamTask* pTask, SStreamDispatchReq* pReq, SRpcMsg* pRsp) { + SStreamDataBlock* pBlock = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM); + int8_t status; + + // 1.1 update status + // TODO cal backpressure + if (pBlock == NULL) { + atomic_store_8(&pTask->inputStatus, TASK_INPUT_STATUS__FAILED); + status = TASK_INPUT_STATUS__FAILED; + } else { + status = atomic_load_8(&pTask->inputStatus); + } + + // 1.2 enqueue + pBlock->type = STREAM_DATA_TYPE_SSDATA_BLOCK; + pBlock->sourceVg = pReq->sourceVg; + pBlock->sourceVer = pReq->sourceVer; + taosWriteQitem(pTask->inputQ, pBlock); + + // 1.3 rsp by input status + SStreamDispatchRsp* pCont = rpcMallocCont(sizeof(SStreamDispatchRsp)); + pCont->inputStatus = status; + pRsp->pCont = pCont; + pRsp->contLen = sizeof(SStreamDispatchRsp); + tmsgSendRsp(pRsp); + + return 0; +} + +int32_t streamTaskProcessDispatchReq(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDispatchReq* pReq, SRpcMsg* pRsp) { + // 1. handle input + streamTaskEnqueue(pTask, pReq, pRsp); + + // 2. try exec + // 2.1. idle: exec + // 2.2. executing: return + // 2.3. closing: keep trying + streamTaskExec2(pTask, pMsgCb); + + // 3. handle output + // 3.1 check and set status + // 3.2 dispatch / sink + streamTaskSink(pTask, pMsgCb); + return 0; } @@ -415,64 +411,6 @@ int32_t streamTaskProcessRecoverReq(SStreamTask* pTask, char* msg) { return 0; } -int32_t streamTaskRun(SStreamTask* pTask) { - SArray* pRes = NULL; - if (pTask->execType == TASK_EXEC__PIPE || pTask->execType == TASK_EXEC__MERGE) { - // TODO remove multi runner - void* exec = pTask->exec.runners[0].executor; - - int8_t status = atomic_val_compare_exchange_8(&pTask->status, TASK_STATUS__IDLE, TASK_STATUS__EXECUTING); - if (status == TASK_STATUS__IDLE) { - pRes = taosArrayInit(0, sizeof(void*)); - if (pRes == NULL) { - return -1; - } - - void* input = NULL; - taosWriteQitem(pTask->inputQ, &input); - if (input == NULL) return 0; - - // TODO: fix type - if (pTask->sourceType == TASK_SOURCE__SCAN) { - SStreamDataSubmit* pSubmit = (SStreamDataSubmit*)input; - qSetStreamInput(exec, pSubmit->data, STREAM_DATA_TYPE_SUBMIT_BLOCK); - while (1) { - SSDataBlock* output; - uint64_t ts = 0; - if (qExecTask(exec, &output, &ts) < 0) { - ASSERT(false); - } - if (output == NULL) break; - taosArrayPush(pRes, &output); - } - streamDataSubmitRefDec(pSubmit); - } else { - SStreamDataBlock* pStreamBlock = (SStreamDataBlock*)input; - const SArray* blocks = pStreamBlock->blocks; - qSetMultiStreamInput(exec, blocks->pData, blocks->size, STREAM_DATA_TYPE_SSDATA_BLOCK); - while (1) { - SSDataBlock* output; - uint64_t ts = 0; - if (qExecTask(exec, &output, &ts) < 0) { - ASSERT(false); - } - if (output == NULL) break; - taosArrayPush(pRes, &output); - } - // TODO: wrap destroy block - taosArrayDestroyP(pStreamBlock->blocks, (FDelete)blockDataDestroy); - } - - if (taosArrayGetSize(pRes) != 0) { - SArray** resQ = taosAllocateQitem(sizeof(void**), DEF_QITEM); - *resQ = pRes; - taosWriteQitem(pTask->outputQ, resQ); - } - } - } - return 0; -} - int32_t streamExecTask(SStreamTask* pTask, SMsgCb* pMsgCb, const void* input, int32_t inputType, int32_t workId) { SArray* pRes = NULL; // source @@ -545,7 +483,7 @@ int32_t streamExecTask(SStreamTask* pTask, SMsgCb* pMsgCb, const void* input, in if (pTask->dispatchType == TASK_DISPATCH__INPLACE) { SRpcMsg dispatchMsg = {0}; - if (streamBuildDispatchMsg(pTask, pRes, &dispatchMsg, NULL) < 0) { + if (streamBuildExecMsg(pTask, pRes, &dispatchMsg, NULL) < 0) { ASSERT(0); return -1; } @@ -566,7 +504,7 @@ int32_t streamExecTask(SStreamTask* pTask, SMsgCb* pMsgCb, const void* input, in } else if (pTask->dispatchType == TASK_DISPATCH__FIXED) { SRpcMsg dispatchMsg = {0}; SEpSet* pEpSet = NULL; - if (streamBuildDispatchMsg(pTask, pRes, &dispatchMsg, &pEpSet) < 0) { + if (streamBuildExecMsg(pTask, pRes, &dispatchMsg, &pEpSet) < 0) { ASSERT(0); return -1; } diff --git a/source/util/src/terror.c b/source/util/src/terror.c index b5e64242e4..58ddaa2109 100644 --- a/source/util/src/terror.c +++ b/source/util/src/terror.c @@ -285,6 +285,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_TOPIC_QUERY, "Topic with invalid qu TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_TOPIC_OPTION, "Topic with invalid option") TAOS_DEFINE_ERROR(TSDB_CODE_MND_CONSUMER_NOT_EXIST, "Consumer not exist") TAOS_DEFINE_ERROR(TSDB_CODE_MND_CONSUMER_NOT_READY, "Consumer waiting for rebalance") +TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOPIC_SUBSCRIBED, "Topic subscribed cannot be dropped") // mnode-sma TAOS_DEFINE_ERROR(TSDB_CODE_MND_SMA_ALREADY_EXIST, "SMA already exists") @@ -358,7 +359,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_TDB_TABLE_RECREATED, "Table re-created") TAOS_DEFINE_ERROR(TSDB_CODE_TDB_TDB_ENV_OPEN_ERROR, "TDB env open error") TAOS_DEFINE_ERROR(TSDB_CODE_TDB_NO_SMA_INDEX_IN_META, "No sma index in meta") TAOS_DEFINE_ERROR(TSDB_CODE_TDB_INVALID_SMA_STAT, "Invalid sma state") -TAOS_DEFINE_ERROR(TSDB_CODE_TDB_TSMA_ALREADY_EXIST, "Tsma already exists") +TAOS_DEFINE_ERROR(TSDB_CODE_TDB_TSMA_ALREADY_EXIST, "TSMA already exists") // query diff --git a/source/util/src/thashutil.c b/source/util/src/thashutil.c index d5182cb892..c2382550a6 100644 --- a/source/util/src/thashutil.c +++ b/source/util/src/thashutil.c @@ -30,7 +30,7 @@ (h) ^= (h) >> 13; \ (h) *= 0xc2b2ae35; \ (h) ^= (h) >> 16; } while (0) - + uint32_t MurmurHash3_32(const char *key, uint32_t len) { const uint8_t *data = (const uint8_t *)key; const int32_t nblocks = len >> 2u; @@ -78,18 +78,54 @@ uint32_t MurmurHash3_32(const char *key, uint32_t len) { return h1; } +uint64_t MurmurHash3_64(const char *key, uint32_t len) { + const uint64_t m = 0x87c37b91114253d5; + const int r = 47; + uint32_t seed = 0x12345678; + uint64_t h = seed ^ (len * m); + const uint8_t *data = (const uint8_t *)key; + const uint8_t *end = data + (len-(len&7)); + + while(data != end) { + uint64_t k = *((uint64_t*)data); + + k *= m; + k ^= k >> r; + k *= m; + h ^= k; + h *= m; + data += 8; + } + + switch(len & 7) { + case 7: h ^= (uint64_t)data[6] << 48; /* fall-thru */ + case 6: h ^= (uint64_t)data[5] << 40; /* fall-thru */ + case 5: h ^= (uint64_t)data[4] << 32; /* fall-thru */ + case 4: h ^= (uint64_t)data[3] << 24; /* fall-thru */ + case 3: h ^= (uint64_t)data[2] << 16; /* fall-thru */ + case 2: h ^= (uint64_t)data[1] << 8; /* fall-thru */ + case 1: h ^= (uint64_t)data[0]; + h *= m; /* fall-thru */ + }; + + h ^= h >> r; + h *= m; + h ^= h >> r; + return h; +} + uint32_t taosIntHash_32(const char *key, uint32_t UNUSED_PARAM(len)) { return *(uint32_t *)key; } uint32_t taosIntHash_16(const char *key, uint32_t UNUSED_PARAM(len)) { return *(uint16_t *)key; } uint32_t taosIntHash_8(const char *key, uint32_t UNUSED_PARAM(len)) { return *(uint8_t *)key; } uint32_t taosFloatHash(const char *key, uint32_t UNUSED_PARAM(len)) { - float f = GET_FLOAT_VAL(key); + float f = GET_FLOAT_VAL(key); if (isnan(f)) { return 0x7fc00000; } - + if (FLT_EQUAL(f, 0.0)) { return 0; - } + } if (fabs(f) < FLT_MAX/BASE - DLT) { int32_t t = (int32_t)(round(BASE * (f + DLT))); return (uint32_t)t; @@ -98,27 +134,27 @@ uint32_t taosFloatHash(const char *key, uint32_t UNUSED_PARAM(len)) { } } uint32_t taosDoubleHash(const char *key, uint32_t UNUSED_PARAM(len)) { - double f = GET_DOUBLE_VAL(key); + double f = GET_DOUBLE_VAL(key); if (isnan(f)) { return 0x7fc00000; } if (FLT_EQUAL(f, 0.0)) { return 0; - } + } if (fabs(f) < DBL_MAX/BASE - DLT) { int32_t t = (int32_t)(round(BASE * (f + DLT))); return (uint32_t)t; } else { return 0x7fc00000; - } + } } uint32_t taosIntHash_64(const char *key, uint32_t UNUSED_PARAM(len)) { uint64_t val = *(uint64_t *)key; uint64_t hash = val >> 16U; hash += (val & 0xFFFFU); - + return (uint32_t)hash; } @@ -127,39 +163,39 @@ _hash_fn_t taosGetDefaultHashFunction(int32_t type) { switch(type) { case TSDB_DATA_TYPE_TIMESTAMP: case TSDB_DATA_TYPE_UBIGINT: - case TSDB_DATA_TYPE_BIGINT: + case TSDB_DATA_TYPE_BIGINT: fn = taosIntHash_64; break; - case TSDB_DATA_TYPE_BINARY: + case TSDB_DATA_TYPE_BINARY: fn = MurmurHash3_32; break; - case TSDB_DATA_TYPE_NCHAR: + case TSDB_DATA_TYPE_NCHAR: fn = MurmurHash3_32; break; case TSDB_DATA_TYPE_UINT: - case TSDB_DATA_TYPE_INT: - fn = taosIntHash_32; + case TSDB_DATA_TYPE_INT: + fn = taosIntHash_32; break; - case TSDB_DATA_TYPE_SMALLINT: - case TSDB_DATA_TYPE_USMALLINT: - fn = taosIntHash_16; + case TSDB_DATA_TYPE_SMALLINT: + case TSDB_DATA_TYPE_USMALLINT: + fn = taosIntHash_16; break; case TSDB_DATA_TYPE_BOOL: case TSDB_DATA_TYPE_UTINYINT: - case TSDB_DATA_TYPE_TINYINT: - fn = taosIntHash_8; + case TSDB_DATA_TYPE_TINYINT: + fn = taosIntHash_8; break; - case TSDB_DATA_TYPE_FLOAT: - fn = taosFloatHash; - break; - case TSDB_DATA_TYPE_DOUBLE: - fn = taosDoubleHash; - break; - default: + case TSDB_DATA_TYPE_FLOAT: + fn = taosFloatHash; + break; + case TSDB_DATA_TYPE_DOUBLE: + fn = taosDoubleHash; + break; + default: fn = taosIntHash_32; break; } - + return fn; } diff --git a/tests/script/jenkins/basic.txt b/tests/script/jenkins/basic.txt index f6bc9f8306..18fe3b9afe 100644 --- a/tests/script/jenkins/basic.txt +++ b/tests/script/jenkins/basic.txt @@ -104,7 +104,8 @@ ./test.sh -f tsim/mnode/basic1.sim -m # --- sma -# ./test.sh -f tsim/sma/tsmaCreateInsertData.sim +./test.sh -f tsim/sma/tsmaCreateInsertData.sim +./test.sh -f tsim/sma/rsmaCreateInsertQuery.sim # --- valgrind ./test.sh -f tsim/valgrind/checkError.sim -v diff --git a/tests/script/tsim/sma/rsmaCreateInsertQuery.sim b/tests/script/tsim/sma/rsmaCreateInsertQuery.sim new file mode 100644 index 0000000000..38ae0dc0a2 --- /dev/null +++ b/tests/script/tsim/sma/rsmaCreateInsertQuery.sim @@ -0,0 +1,89 @@ +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 + +print =============== create database with retentions +sql create database d0 retentions 15s:7d,1m:21d,15m:365d; +sql use d0 + +print =============== create super table and register rsma +sql create table if not exists stb (ts timestamp, c1 int) tags (city binary(20),district binary(20)) rollup(min) file_factor 0.1 delay 2; + +sql show stables +if $rows != 1 then + return -1 +endi + +print =============== create child table +sql create table ct1 using stb tags("BeiJing", "ChaoYang"); + +sql show tables +if $rows != 1 then + return -1 +endi + +print =============== insert data and trigger rollup +sql insert into ct1 values(now, 10); +sql insert into ct1 values(now+1s, 1); +sql insert into ct1 values(now+2s, 100); + + +print =============== select * from retention level 2 from memory +sql select * from ct1; +print $data00 $data01 +if $rows > 2 then + print retention level 2 file rows $rows > 2 + return -1 +endi +print =============== select * from retention level 1 from memory +sql select * from ct1 where ts > now-8d; +print $data00 $data01 +if $rows > 2 then + print retention level 1 file rows $rows > 2 + return -1 +endi +print =============== select * from retention level 0 from memory +sql select * from ct1 where ts > now-3d; +print $data00 $data01 +print $data10 $data11 +print $data20 $data21 +if $rows < 1 then + print retention level 0 file rows $rows < 1 + return -1 +endi +#=================================================================== + + +#==================== reboot to trigger commit data to file +system sh/exec.sh -n dnode1 -s stop -x SIGINT +system sh/exec.sh -n dnode1 -s start + +print =============== select * from retention level 2 from file +sql select * from ct1; +print $data00 $data01 +if $rows > 2 then + print retention level 2 file rows $rows > 2 + return -1 +endi + +print =============== select * from retention level 1 from file +sql select * from ct1 where ts > now-8d; +print $data00 $data01 +if $rows > 2 then + print retention level 1 file rows $rows > 2 + return -1 +endi + +print =============== select * from retention level 0 from file +sql select * from ct1 where ts > now-3d; +print $data00 $data01 +print $data10 $data11 +print $data20 $data21 +if $rows < 1 then + print retention level 0 file rows $rows < 1 + return -1 +endi + +system sh/exec.sh -n dnode1 -s stop -x SIGINT \ No newline at end of file diff --git a/tests/system-test/7-tmq/basic5.py b/tests/system-test/7-tmq/basic5.py index c2fe25efc4..4a29cacd97 100644 --- a/tests/system-test/7-tmq/basic5.py +++ b/tests/system-test/7-tmq/basic5.py @@ -360,7 +360,7 @@ class TDTestCase: # wait db ready while 1: tdSql.query("show databases") - if tdSql.getRows() == 4: + if tdSql.getRows() == 5: print (tdSql.getData(0,0), tdSql.getData(1,0),tdSql.getData(2,0),) break else: diff --git a/tests/system-test/7-tmq/subscribeDb.py b/tests/system-test/7-tmq/subscribeDb.py index ec4c9e9d9b..d2cccd0532 100644 --- a/tests/system-test/7-tmq/subscribeDb.py +++ b/tests/system-test/7-tmq/subscribeDb.py @@ -251,7 +251,7 @@ class TDTestCase: expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"] topicList = topicName1 ifcheckdata = 0 - ifManualCommit = 0 + ifManualCommit = 1 keyList = 'group.id:cgrp1,\ enable.auto.commit:false,\ auto.commit.interval.ms:6000,\ @@ -410,7 +410,7 @@ class TDTestCase: expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"] + parameterDict2["rowsPerTbl"] * parameterDict2["ctbNum"] topicList = topicName1 ifcheckdata = 0 - ifManualCommit = 0 + ifManualCommit = 1 keyList = 'group.id:cgrp1,\ enable.auto.commit:false,\ auto.commit.interval.ms:6000,\ @@ -487,7 +487,7 @@ class TDTestCase: expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"] + parameterDict2["rowsPerTbl"] * parameterDict2["ctbNum"] topicList = topicName1 ifcheckdata = 0 - ifManualCommit = 0 + ifManualCommit = 1 keyList = 'group.id:cgrp1,\ enable.auto.commit:false,\ auto.commit.interval.ms:6000,\ @@ -659,7 +659,7 @@ class TDTestCase: expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"] + parameterDict2["rowsPerTbl"] * parameterDict2["ctbNum"] topicList = topicName1 + ',' + topicName2 ifcheckdata = 0 - ifManualCommit = 0 + ifManualCommit = 1 keyList = 'group.id:cgrp1,\ enable.auto.commit:false,\ auto.commit.interval.ms:6000,\ @@ -708,9 +708,9 @@ class TDTestCase: cfgPath = buildPath + "/../sim/psim/cfg" tdLog.info("cfgPath: %s" % cfgPath) - #self.tmqCase1(cfgPath, buildPath) - #self.tmqCase2(cfgPath, buildPath) - #self.tmqCase3(cfgPath, buildPath) + self.tmqCase1(cfgPath, buildPath) + self.tmqCase2(cfgPath, buildPath) + self.tmqCase3(cfgPath, buildPath) self.tmqCase4(cfgPath, buildPath) self.tmqCase5(cfgPath, buildPath) self.tmqCase6(cfgPath, buildPath)