diff --git a/include/common/tmsg.h b/include/common/tmsg.h index fd7a843e7fbf02e291a01ce5cda4fbc4c555e5a4..da48846a8f24c63187d9dde1f379a2bf26eadd09 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -1521,8 +1521,8 @@ typedef struct { char* qmsg2; // pAst2:qmsg2:SRetention2 => trigger aggr task2 } SRSmaParam; -int tEncodeSRSmaParam(SCoder* pCoder, const SRSmaParam* pRSmaParam); -int tDecodeSRSmaParam(SCoder* pCoder, SRSmaParam* pRSmaParam); +int32_t tEncodeSRSmaParam(SCoder* pCoder, const SRSmaParam* pRSmaParam); +int32_t tDecodeSRSmaParam(SCoder* pCoder, SRSmaParam* pRSmaParam); typedef struct SVCreateStbReq { const char* name; diff --git a/include/common/trow.h b/include/common/trow.h index 8732497dbb7bd585be4c1268fc5876f0a14fc737..ab956f1db7bf8e175cc73f8c9f91a17d277346e4 100644 --- a/include/common/trow.h +++ b/include/common/trow.h @@ -214,6 +214,14 @@ STSRow *tdRowDup(STSRow *row); static FORCE_INLINE SKvRowIdx *tdKvRowColIdxAt(STSRow *pRow, col_id_t idx) { return (SKvRowIdx *)TD_ROW_COL_IDX(pRow) + idx; } +static FORCE_INLINE int16_t tdKvRowColIdAt(STSRow *pRow, col_id_t idx) { + ASSERT(idx >= 0); + if (idx == 0) { + return PRIMARYKEY_TIMESTAMP_COL_ID; + } + + return ((SKvRowIdx *)TD_ROW_COL_IDX(pRow) + idx - 1)->colId; +} static FORCE_INLINE void *tdKVRowColVal(STSRow *pRow, SKvRowIdx *pIdx) { return POINTER_SHIFT(pRow, pIdx->offset); } #define TD_ROW_OFFSET(p) ((p)->toffset); // During ParseInsert when without STSchema, how to get the offset for STpRow? diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index 8c1cfa510129fba002f832c4938c0418e0cc26df..afab703e773cb19180fbbbe232ecac3a42f510e8 100644 --- a/source/common/src/tmsg.c +++ b/source/common/src/tmsg.c @@ -3621,6 +3621,43 @@ void tFreeSCMCreateStreamReq(SCMCreateStreamReq *pReq) { taosMemoryFreeClear(pReq->ast); } +int32_t tEncodeSRSmaParam(SCoder *pCoder, const SRSmaParam *pRSmaParam) { + if (tEncodeFloat(pCoder, pRSmaParam->xFilesFactor) < 0) return -1; + if (tEncodeI32v(pCoder, pRSmaParam->delay) < 0) return -1; + if (tEncodeI32v(pCoder, pRSmaParam->qmsg1Len) < 0) return -1; + if (tEncodeI32v(pCoder, pRSmaParam->qmsg2Len) < 0) return -1; + if (pRSmaParam->qmsg1Len > 0) { + if (tEncodeBinary(pCoder, pRSmaParam->qmsg1, (uint64_t)pRSmaParam->qmsg1Len) < 0) // qmsg1Len contains len of '\0' + return -1; + } + if (pRSmaParam->qmsg2Len > 0) { + if (tEncodeBinary(pCoder, pRSmaParam->qmsg2, (uint64_t)pRSmaParam->qmsg2Len) < 0) // qmsg2Len contains len of '\0' + return -1; + } + + return 0; +} + +int32_t tDecodeSRSmaParam(SCoder *pCoder, SRSmaParam *pRSmaParam) { + if (tDecodeFloat(pCoder, &pRSmaParam->xFilesFactor) < 0) return -1; + if (tDecodeI32v(pCoder, &pRSmaParam->delay) < 0) return -1; + if (tDecodeI32v(pCoder, &pRSmaParam->qmsg1Len) < 0) return -1; + if (tDecodeI32v(pCoder, &pRSmaParam->qmsg2Len) < 0) return -1; + if (pRSmaParam->qmsg1Len > 0) { + uint64_t len; + if (tDecodeBinaryAlloc(pCoder, (void **)&pRSmaParam->qmsg1, &len) < 0) return -1; // qmsg1Len contains len of '\0' + } else { + pRSmaParam->qmsg1 = NULL; + } + if (pRSmaParam->qmsg2Len > 0) { + uint64_t len; + if (tDecodeBinaryAlloc(pCoder, (void **)&pRSmaParam->qmsg2, &len) < 0) return -1; // qmsg2Len contains len of '\0' + } else { + pRSmaParam->qmsg2 = NULL; + } + return 0; +} + int tEncodeSVCreateStbReq(SCoder *pCoder, const SVCreateStbReq *pReq) { if (tStartEncode(pCoder) < 0) return -1; @@ -3629,9 +3666,9 @@ int tEncodeSVCreateStbReq(SCoder *pCoder, const SVCreateStbReq *pReq) { if (tEncodeI8(pCoder, pReq->rollup) < 0) return -1; if (tEncodeSSchemaWrapper(pCoder, &pReq->schema) < 0) return -1; if (tEncodeSSchemaWrapper(pCoder, &pReq->schemaTag) < 0) return -1; - // if (pReq->rollup) { - // if (tEncodeSRSmaParam(pCoder, pReq->pRSmaParam) < 0) return -1; - // } + if (pReq->rollup) { + if (tEncodeSRSmaParam(pCoder, &pReq->pRSmaParam) < 0) return -1; + } tEndEncode(pCoder); return 0; @@ -3645,9 +3682,9 @@ int tDecodeSVCreateStbReq(SCoder *pCoder, SVCreateStbReq *pReq) { if (tDecodeI8(pCoder, &pReq->rollup) < 0) return -1; if (tDecodeSSchemaWrapper(pCoder, &pReq->schema) < 0) return -1; if (tDecodeSSchemaWrapper(pCoder, &pReq->schemaTag) < 0) return -1; - // if (pReq->rollup) { - // if (tDecodeSRSmaParam(pCoder, pReq->pRSmaParam) < 0) return -1; - // } + if (pReq->rollup) { + if (tDecodeSRSmaParam(pCoder, &pReq->pRSmaParam) < 0) return -1; + } tEndDecode(pCoder); return 0; diff --git a/source/common/src/trow.c b/source/common/src/trow.c index c73f26e6dab78828965788e78a21ece9231d9aff..7157c1e0f0c2ec3953eacc73d07ae979b6b694b7 100644 --- a/source/common/src/trow.c +++ b/source/common/src/trow.c @@ -220,7 +220,7 @@ static uint8_t tdGetMergedBitmapByte(uint8_t byte) { } /** - * @brief Merge bitmap from 2 bits to 1 bits, and the memory buffer should be guaranteed by the invoker. + * @brief Merge bitmap from 2 bits to 1 bit, and the memory buffer should be guaranteed by the invoker. * * @param srcBitmap * @param nBits diff --git a/source/dnode/vnode/src/inc/tsdbSma.h b/source/dnode/vnode/src/inc/tsdbSma.h index 8fb18ddfea868f10cc226dc8e891fe5ae56aff2f..162d733cc3654bb2d743334ff6758508c7c98683 100644 --- a/source/dnode/vnode/src/inc/tsdbSma.h +++ b/source/dnode/vnode/src/inc/tsdbSma.h @@ -22,13 +22,13 @@ extern "C" { #endif -typedef int32_t (*__tb_ddl_fn_t)(void *ahandle, void **result, void *p1, void *p2); +// typedef int32_t (*__tb_ddl_fn_t)(void *ahandle, void **result, void *p1, void *p2); -struct STbDdlH { - void *ahandle; - void *result; - __tb_ddl_fn_t fp; -}; +// struct STbDdlH { +// void *ahandle; +// void *result; +// __tb_ddl_fn_t fp; +// }; static FORCE_INLINE int32_t tsdbUidStoreInit(STbUidStore **pStore) { ASSERT(*pStore == NULL); @@ -40,14 +40,6 @@ static FORCE_INLINE int32_t tsdbUidStoreInit(STbUidStore **pStore) { return TSDB_CODE_SUCCESS; } -int32_t tsdbUidStorePut(STbUidStore *pStore, tb_uid_t suid, tb_uid_t *uid); -void tsdbUidStoreDestory(STbUidStore *pStore); -void *tsdbUidStoreFree(STbUidStore *pStore); - -int32_t tsdbRegisterRSma(STsdb *pTsdb, SMeta *pMeta, SVCreateTbReq *pReq); -int32_t tsdbFetchTbUidList(void *pTsdb, void **result, void *suid, void *uid); -int32_t tsdbUpdateTbUidList(STsdb *pTsdb, STbUidStore *pUidStore); -int32_t tsdbTriggerRSma(STsdb *pTsdb, SMeta *pMeta, void *pMsg, int32_t inputType); #ifdef __cplusplus } diff --git a/source/dnode/vnode/src/inc/vnodeInt.h b/source/dnode/vnode/src/inc/vnodeInt.h index 884f86a3bb3dab85fd9bc4b2af47a5c6de8beb94..b5066799ca172711c3511e93cbe59520fc5093db 100644 --- a/source/dnode/vnode/src/inc/vnodeInt.h +++ b/source/dnode/vnode/src/inc/vnodeInt.h @@ -107,6 +107,15 @@ int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen); int32_t tqProcessStreamTrigger(STQ* pTq, void* data, int32_t dataLen, int32_t workerId); int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId); +// sma + +int32_t tsdbRegisterRSma(STsdb* pTsdb, SMeta* pMeta, SVCreateStbReq* pReq); +int32_t tsdbFetchTbUidList(STsdb* pTsdb, STbUidStore** ppStore, tb_uid_t suid, tb_uid_t uid); +int32_t tsdbUpdateTbUidList(STsdb* pTsdb, STbUidStore* pUidStore); +void tsdbUidStoreDestory(STbUidStore* pStore); +void* tsdbUidStoreFree(STbUidStore* pStore); +int32_t tsdbTriggerRSma(STsdb* pTsdb, SMeta* pMeta, void* pMsg, int32_t inputType); + typedef struct { int8_t streamType; // sma or other int8_t dstType; @@ -162,7 +171,7 @@ struct STbUidStore { #define TD_VID(PVNODE) (PVNODE)->config.vgId -typedef struct STbDdlH STbDdlH; +// typedef struct STbDdlH STbDdlH; // sma void smaHandleRes(void* pVnode, int64_t smaId, const SArray* data); diff --git a/source/dnode/vnode/src/tsdb/tsdbRead.c b/source/dnode/vnode/src/tsdb/tsdbRead.c index 1b23d12c2f6a42279376f79f1c9e0f3bcc364d8d..6243eb48910de0e991c38064aab94d9cbe34beb2 100644 --- a/source/dnode/vnode/src/tsdb/tsdbRead.c +++ b/source/dnode/vnode/src/tsdb/tsdbRead.c @@ -1519,8 +1519,7 @@ static void mergeTwoRowFromMem(STsdbReadHandle* pTsdbReadHandle, int32_t capacit } else if (isRow1DataRow) { colIdOfRow1 = pSchema1->columns[j].colId; } else { - SKvRowIdx* pColIdx = tdKvRowColIdxAt(row1, j); - colIdOfRow1 = pColIdx->colId; + colIdOfRow1 = tdKvRowColIdAt(row1, j); } int32_t colIdOfRow2; @@ -1529,8 +1528,7 @@ static void mergeTwoRowFromMem(STsdbReadHandle* pTsdbReadHandle, int32_t capacit } else if (isRow2DataRow) { colIdOfRow2 = pSchema2->columns[k].colId; } else { - SKvRowIdx* pColIdx = tdKvRowColIdxAt(row2, k); - colIdOfRow2 = pColIdx->colId; + colIdOfRow2 = tdKvRowColIdAt(row2, j); } if (colIdOfRow1 == colIdOfRow2) { diff --git a/source/dnode/vnode/src/tsdb/tsdbSma.c b/source/dnode/vnode/src/tsdb/tsdbSma.c index c9e33cefc7c87f6601e53026252c89a0a629c5a1..1abca21d34f9235511b477ee3e484d32e68558d3 100644 --- a/source/dnode/vnode/src/tsdb/tsdbSma.c +++ b/source/dnode/vnode/src/tsdb/tsdbSma.c @@ -173,6 +173,7 @@ static void tsdbGetSmaDir(int32_t vgId, ETsdbSmaType smaType, char dirName[]) static int32_t tsdbInsertTSmaDataImpl(STsdb *pTsdb, int64_t indexUid, const char *msg); static int32_t tsdbInsertRSmaDataImpl(STsdb *pTsdb, const char *msg); +static FORCE_INLINE int32_t tsdbUidStorePut(STbUidStore *pStore, tb_uid_t suid, tb_uid_t *uid); static FORCE_INLINE int32_t tsdbUpdateTbUidListImpl(STsdb *pTsdb, tb_uid_t *suid, SArray *tbUids); // mgmt interface static int32_t tsdbDropTSmaDataImpl(STsdb *pTsdb, int64_t indexUid); @@ -1692,18 +1693,16 @@ int32_t tsdbDropTSma(STsdb *pTsdb, char *pMsg) { * @param pReq * @return int32_t */ -int32_t tsdbRegisterRSma(STsdb *pTsdb, SMeta *pMeta, SVCreateTbReq *pReq) { -#if 0 - SRSmaParam *param = pReq->stbCfg.pRSmaParam; - - if (!param) { - tsdbDebug("vgId:%d return directly since no rollup for stable %s %" PRIi64, REPO_ID(pTsdb), pReq->name, - pReq->stbCfg.suid); +int32_t tsdbRegisterRSma(STsdb *pTsdb, SMeta *pMeta, SVCreateStbReq *pReq) { + if (!pReq->rollup) { + tsdbDebug("vgId:%d return directly since no rollup for stable %s %" PRIi64, REPO_ID(pTsdb), pReq->name, pReq->suid); return TSDB_CODE_SUCCESS; } + SRSmaParam *param = &pReq->pRSmaParam; + if ((param->qmsg1Len == 0) && (param->qmsg2Len == 0)) { - tsdbWarn("vgId:%d no qmsg1/qmsg2 for rollup stable %s %" PRIi64, REPO_ID(pTsdb), pReq->name, pReq->stbCfg.suid); + tsdbWarn("vgId:%d no qmsg1/qmsg2 for rollup stable %s %" PRIi64, REPO_ID(pTsdb), pReq->name, pReq->suid); return TSDB_CODE_SUCCESS; } @@ -1716,9 +1715,9 @@ int32_t tsdbRegisterRSma(STsdb *pTsdb, SMeta *pMeta, SVCreateTbReq *pReq) { SSmaStat *pStat = SMA_ENV_STAT(pEnv); SRSmaInfo *pRSmaInfo = NULL; - pRSmaInfo = taosHashGet(SMA_STAT_INFO_HASH(pStat), &pReq->stbCfg.suid, sizeof(tb_uid_t)); + pRSmaInfo = taosHashGet(SMA_STAT_INFO_HASH(pStat), &pReq->suid, sizeof(tb_uid_t)); if (pRSmaInfo) { - tsdbWarn("vgId:%d rsma info already exists for stb: %s, %" PRIi64, REPO_ID(pTsdb), pReq->name, pReq->stbCfg.suid); + tsdbWarn("vgId:%d rsma info already exists for stb: %s, %" PRIi64, REPO_ID(pTsdb), pReq->name, pReq->suid); return TSDB_CODE_SUCCESS; } @@ -1758,14 +1757,13 @@ int32_t tsdbRegisterRSma(STsdb *pTsdb, SMeta *pMeta, SVCreateTbReq *pReq) { } } - if (taosHashPut(SMA_STAT_INFO_HASH(pStat), &pReq->stbCfg.suid, sizeof(tb_uid_t), &pRSmaInfo, sizeof(pRSmaInfo)) != + if (taosHashPut(SMA_STAT_INFO_HASH(pStat), &pReq->suid, sizeof(tb_uid_t), &pRSmaInfo, sizeof(pRSmaInfo)) != TSDB_CODE_SUCCESS) { return TSDB_CODE_FAILED; } else { - tsdbDebug("vgId:%d register rsma info succeed for suid:%" PRIi64, REPO_ID(pTsdb), pReq->stbCfg.suid); + tsdbDebug("vgId:%d register rsma info succeed for suid:%" PRIi64, REPO_ID(pTsdb), pReq->suid); } -#endif return TSDB_CODE_SUCCESS; } @@ -1777,7 +1775,7 @@ int32_t tsdbRegisterRSma(STsdb *pTsdb, SMeta *pMeta, SVCreateTbReq *pReq) { * @param uid * @return int32_t */ -int32_t tsdbUidStorePut(STbUidStore *pStore, tb_uid_t suid, tb_uid_t *uid) { +static int32_t tsdbUidStorePut(STbUidStore *pStore, tb_uid_t suid, tb_uid_t *uid) { // prefer to store suid/uids in array if ((suid == pStore->suid) || (pStore->suid == 0)) { if (pStore->suid == 0) { @@ -1833,6 +1831,7 @@ void tsdbUidStoreDestory(STbUidStore *pStore) { if (pStore) { if (pStore->uidHash) { if (pStore->tbUids) { + // When pStore->tbUids not NULL, the pStore->uidHash has k/v; otherwise pStore->uidHash only has keys. void *pIter = taosHashIterate(pStore->uidHash, NULL); while (pIter) { SArray *arr = *(SArray **)pIter; @@ -1847,8 +1846,10 @@ void tsdbUidStoreDestory(STbUidStore *pStore) { } void *tsdbUidStoreFree(STbUidStore *pStore) { - tsdbUidStoreDestory(pStore); - taosMemoryFree(pStore); + if (pStore) { + tsdbUidStoreDestory(pStore); + taosMemoryFree(pStore); + } return NULL; } @@ -1861,7 +1862,7 @@ void *tsdbUidStoreFree(STbUidStore *pStore) { * @param uid * @return int32_t */ -int32_t tsdbFetchTbUidList(void *pTsdb, void **ppStore, void *suid, void *uid) { +int32_t tsdbFetchTbUidList(STsdb *pTsdb, STbUidStore **ppStore, tb_uid_t suid, tb_uid_t uid) { SSmaEnv *pEnv = REPO_RSMA_ENV((STsdb *)pTsdb); // only applicable to rollup SMA ctables @@ -1877,7 +1878,7 @@ int32_t tsdbFetchTbUidList(void *pTsdb, void **ppStore, void *suid, void *uid) { } // info cached when create rsma stable and return directly for non-rsma ctables - if (!taosHashGet(infoHash, suid, sizeof(tb_uid_t))) { + if (!taosHashGet(infoHash, &suid, sizeof(tb_uid_t))) { return TSDB_CODE_SUCCESS; } @@ -1887,7 +1888,7 @@ int32_t tsdbFetchTbUidList(void *pTsdb, void **ppStore, void *suid, void *uid) { } } - if (tsdbUidStorePut(*ppStore, *(tb_uid_t *)suid, (tb_uid_t *)uid) != 0) { + if (tsdbUidStorePut(*ppStore, suid, &uid) != 0) { *ppStore = tsdbUidStoreFree(*ppStore); return TSDB_CODE_FAILED; } @@ -1935,12 +1936,10 @@ static FORCE_INLINE int32_t tsdbUpdateTbUidListImpl(STsdb *pTsdb, tb_uid_t *suid int32_t tsdbUpdateTbUidList(STsdb *pTsdb, STbUidStore *pStore) { if (!pStore || (taosArrayGetSize(pStore->tbUids) == 0)) { tsdbDebug("vgId:%d no need to update tbUids since empty uidStore", REPO_ID(pTsdb)); - tsdbUidStoreFree(pStore); return TSDB_CODE_SUCCESS; } if (tsdbUpdateTbUidListImpl(pTsdb, &pStore->suid, pStore->tbUids) != TSDB_CODE_SUCCESS) { - tsdbUidStoreFree(pStore); return TSDB_CODE_FAILED; } @@ -1951,15 +1950,11 @@ int32_t tsdbUpdateTbUidList(STsdb *pTsdb, STbUidStore *pStore) { if (tsdbUpdateTbUidListImpl(pTsdb, pTbSuid, pTbUids) != TSDB_CODE_SUCCESS) { taosHashCancelIterate(pStore->uidHash, pIter); - tsdbUidStoreFree(pStore); return TSDB_CODE_FAILED; } pIter = taosHashIterate(pStore->uidHash, pIter); } - - tsdbUidStoreFree(pStore); - return TSDB_CODE_SUCCESS; } @@ -1971,8 +1966,6 @@ static int32_t tsdbFetchSubmitReqSuids(SSubmitReq *pMsg, STbUidStore *pStore) { STSRow *row = NULL; terrno = TSDB_CODE_SUCCESS; - // pMsg->length = htonl(pMsg->length); - // pMsg->numOfBlocks = htonl(pMsg->numOfBlocks); if (tInitSubmitMsgIterEx(pMsg, &msgIter) < 0) return -1; while (true) { diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c index 4a5be466983bbe564bf28a04a59a67bbe63c69be..7f7023fccd676d9ed61341b64f2625cd7bd53238 100644 --- a/source/dnode/vnode/src/vnd/vnodeSvr.c +++ b/source/dnode/vnode/src/vnd/vnodeSvr.c @@ -300,13 +300,13 @@ static int vnodeProcessCreateStbReq(SVnode *pVnode, int64_t version, void *pReq, goto _err; } - // tsdbRegisterRSma(pVnode->pTsdb, pVnode->pMeta, &vCreateTbReq); - if (metaCreateSTable(pVnode->pMeta, version, &req) < 0) { pRsp->code = terrno; goto _err; } + tsdbRegisterRSma(pVnode->pTsdb, pVnode->pMeta, &req); + tCoderClear(&coder); return 0; @@ -323,6 +323,7 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq, SVCreateTbBatchRsp rsp = {0}; SVCreateTbRsp cRsp = {0}; char tbName[TSDB_TABLE_FNAME_LEN]; + STbUidStore *pStore = NULL; pRsp->msgType = TDMT_VND_CREATE_TABLE_RSP; pRsp->code = TSDB_CODE_SUCCESS; @@ -361,6 +362,7 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq, cRsp.code = terrno; } else { cRsp.code = TSDB_CODE_SUCCESS; + tsdbFetchTbUidList(pVnode->pTsdb, &pStore, pCreateReq->ctb.suid, pCreateReq->uid); } taosArrayPush(rsp.pArray, &cRsp); @@ -368,6 +370,9 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq, tCoderClear(&coder); + tsdbUpdateTbUidList(pVnode->pTsdb, pStore); + tsdbUidStoreFree(pStore); + // prepare rsp int32_t ret = 0; tEncodeSize(tEncodeSVCreateTbBatchRsp, &rsp, pRsp->contLen, ret); @@ -426,7 +431,7 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in SSubmitRsp rsp = {0}; pRsp->code = 0; - + tsdbTriggerRSma(pVnode->pTsdb, pVnode->pMeta, pReq, STREAM_DATA_TYPE_SUBMIT_BLOCK); // handle the request if (tsdbInsertData(pVnode->pTsdb, version, pSubmitReq, &rsp) < 0) { pRsp->code = terrno; @@ -435,7 +440,7 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in // pRsp->msgType = TDMT_VND_SUBMIT_RSP; // vnodeProcessSubmitReq(pVnode, ptr, pRsp); - // tsdbTriggerRSma(pVnode->pTsdb, pVnode->pMeta, ptr, STREAM_DATA_TYPE_SUBMIT_BLOCK); + // tsdbTriggerRSma(pVnode->pTsdb, pVnode->pMeta, pReq, STREAM_DATA_TYPE_SUBMIT_BLOCK); // encode the response (TODO) pRsp->pCont = rpcMallocCont(sizeof(SSubmitRsp));