diff --git a/examples/c/stream_demo.c b/examples/c/stream_demo.c index 1c9d11b755..46c8297fba 100644 --- a/examples/c/stream_demo.c +++ b/examples/c/stream_demo.c @@ -49,28 +49,28 @@ int32_t init_env() { } taos_free_result(pRes); - pRes = taos_query(pConn, "create stable if not exists st1 (ts timestamp, k int) tags(a int)"); + pRes = taos_query(pConn, "create stable if not exists st1 (ts timestamp, k int, j varchar(20)) tags(a varchar(20))"); if (taos_errno(pRes) != 0) { printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes)); return -1; } taos_free_result(pRes); - pRes = taos_query(pConn, "create table if not exists tu1 using st1 tags(1)"); + pRes = taos_query(pConn, "create table if not exists tu1 using st1 tags('c1')"); if (taos_errno(pRes) != 0) { printf("failed to create child table tu1, reason:%s\n", taos_errstr(pRes)); return -1; } taos_free_result(pRes); - pRes = taos_query(pConn, "create table if not exists tu2 using st1 tags(2)"); + pRes = taos_query(pConn, "create table if not exists tu2 using st1 tags('c2')"); if (taos_errno(pRes) != 0) { printf("failed to create child table tu2, reason:%s\n", taos_errstr(pRes)); return -1; } taos_free_result(pRes); - pRes = taos_query(pConn, "create table if not exists tu3 using st1 tags(3)"); + pRes = taos_query(pConn, "create table if not exists tu3 using st1 tags('c3')"); if (taos_errno(pRes) != 0) { printf("failed to create child table tu3, reason:%s\n", taos_errstr(pRes)); return -1; @@ -96,7 +96,8 @@ int32_t create_stream() { taos_free_result(pRes); pRes = taos_query(pConn, - "create stream stream1 trigger at_once watermark 10s into outstb as select _wstart start, avg(k) from st1 partition by tbname interval(10s)"); + /*"create stream stream1 trigger at_once watermark 10s into outstb as select _wstart start, avg(k) from st1 partition by tbname interval(10s)");*/ + "create stream stream2 into outstb subtable(concat(concat(concat('prefix_', tname), '_suffix_'), cast(k1 as varchar(20)))) as select _wstart wstart, avg(k) from st1 partition by tbname tname, a k1 interval(10s);"); if (taos_errno(pRes) != 0) { printf("failed to create stream stream1, reason:%s\n", taos_errstr(pRes)); return -1; diff --git a/source/dnode/mnode/impl/src/mndConsumer.c b/source/dnode/mnode/impl/src/mndConsumer.c index c853828bf2..7ffa3cef99 100644 --- a/source/dnode/mnode/impl/src/mndConsumer.c +++ b/source/dnode/mnode/impl/src/mndConsumer.c @@ -117,6 +117,7 @@ static int32_t mndProcessConsumerLostMsg(SRpcMsg *pMsg) { if (mndTransPrepare(pMnode, pTrans) != 0) goto FAIL; mndTransDrop(pTrans); + tDeleteSMqConsumerObj(pConsumerNew); return 0; FAIL: tDeleteSMqConsumerObj(pConsumerNew); diff --git a/source/dnode/mnode/impl/src/mndOffset.c b/source/dnode/mnode/impl/src/mndOffset.c index 2ada3e00bb..8d779f0021 100644 --- a/source/dnode/mnode/impl/src/mndOffset.c +++ b/source/dnode/mnode/impl/src/mndOffset.c @@ -182,6 +182,11 @@ static int32_t mndProcessCommitOffsetReq(SRpcMsg *pMsg) { tDecodeSMqCMCommitOffsetReq(&decoder, &commitOffsetReq); STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pMsg, "commit-offset"); + if (pTrans == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + tDecoderClear(&decoder); + return -1; + } for (int32_t i = 0; i < commitOffsetReq.num; i++) { SMqOffset *pOffset = &commitOffsetReq.offsets[i]; diff --git a/source/dnode/mnode/impl/src/mndTopic.c b/source/dnode/mnode/impl/src/mndTopic.c index 5d3a2be79a..ae259b95be 100644 --- a/source/dnode/mnode/impl/src/mndTopic.c +++ b/source/dnode/mnode/impl/src/mndTopic.c @@ -222,6 +222,7 @@ SSdbRow *mndTopicActionDecode(SSdbRaw *pRaw) { } SDB_GET_BINARY(pRaw, dataPos, buf, len, TOPIC_DECODE_OVER); if (taosDecodeSSchemaWrapper(buf, &pTopic->schema) == NULL) { + taosMemoryFree(buf); goto TOPIC_DECODE_OVER; } taosMemoryFree(buf); diff --git a/source/dnode/vnode/src/tq/tqMeta.c b/source/dnode/vnode/src/tq/tqMeta.c index 1a57a391b1..1c5eee7378 100644 --- a/source/dnode/vnode/src/tq/tqMeta.c +++ b/source/dnode/vnode/src/tq/tqMeta.c @@ -170,6 +170,7 @@ int32_t tqMetaRestoreCheckInfo(STQ* pTq) { tDecoderInit(&decoder, (uint8_t*)pVal, vLen); if (tDecodeSTqCheckInfo(&decoder, &info) < 0) { terrno = TSDB_CODE_OUT_OF_MEMORY; + tdbFree(pKey); tdbTbcClose(pCur); return -1; } diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index df7e5ff06f..0cfdd2b68e 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -1341,7 +1341,8 @@ static void calBlockTbName(SExprSupp* pTbNameCalSup, SSDataBlock* pBlock) { void* pData = colDataGetData(pCol, 0); // TODO check tbname validation if (pData != (void*)-1 && pData != NULL) { - memcpy(pBlock->info.parTbName, varDataVal(pData), varDataLen(pData)); + memcpy(pBlock->info.parTbName, varDataVal(pData), TMIN(varDataLen(pData), TSDB_TABLE_NAME_LEN)); + pBlock->info.parTbName[TSDB_TABLE_NAME_LEN - 1] = 0; } else { pBlock->info.parTbName[0] = 0; } diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c index c968f82702..560a1d319a 100644 --- a/source/libs/executor/src/timewindowoperator.c +++ b/source/libs/executor/src/timewindowoperator.c @@ -3611,6 +3611,7 @@ void destroyStreamSessionAggOperatorInfo(void* param) { blockDataDestroy(pInfo->pUpdateRes); destroySqlFunctionCtx(pInfo->pDummyCtx, 0); taosHashCleanup(pInfo->pStDeleted); + taosHashCleanup(pInfo->pGroupIdTbNameMap); taosMemoryFreeClear(param); } @@ -4670,6 +4671,7 @@ void destroyStreamStateOperatorInfo(void* param) { colDataDestroy(&pInfo->twAggSup.timeWindowData); blockDataDestroy(pInfo->pDelRes); taosHashCleanup(pInfo->pSeDeleted); + taosHashCleanup(pInfo->pGroupIdTbNameMap); destroySqlFunctionCtx(pInfo->pDummyCtx, 0); taosMemoryFreeClear(param); diff --git a/source/libs/wal/src/walMeta.c b/source/libs/wal/src/walMeta.c index 233d6a87b8..e49a963191 100644 --- a/source/libs/wal/src/walMeta.c +++ b/source/libs/wal/src/walMeta.c @@ -35,8 +35,8 @@ int64_t FORCE_INLINE walGetCommittedVer(SWal* pWal) { return pWal->vers.commitVe int64_t FORCE_INLINE walGetAppliedVer(SWal* pWal) { return pWal->vers.appliedVer; } -static FORCE_INLINE int walBuildMetaName(SWal* pWal, int metaVer, char* buf) { - return sprintf(buf, "%s/meta-ver%d", pWal->path, metaVer); +static FORCE_INLINE void walBuildMetaName(SWal* pWal, int metaVer, char* buf) { + sprintf(buf, "%s/meta-ver%d", pWal->path, metaVer); } static FORCE_INLINE int64_t walScanLogGetLastVer(SWal* pWal) { @@ -150,6 +150,7 @@ int walCheckAndRepairMeta(SWal* pWal) { const char* idxPattern = "^[0-9]+.idx$"; regex_t logRegPattern; regex_t idxRegPattern; + bool fixed = false; regcomp(&logRegPattern, logPattern, REG_EXTENDED); regcomp(&idxRegPattern, idxPattern, REG_EXTENDED); @@ -206,6 +207,77 @@ int walCheckAndRepairMeta(SWal* pWal) { actualFileNum = taosArrayGetSize(pLogInfoArray); #endif + { + int32_t i = 0, j = 0; + while (i < actualFileNum && j < metaFileNum) { + SWalFileInfo* pActualFile = taosArrayGet(actualLog, i); + SWalFileInfo* pMetaFile = taosArrayGet(pWal->fileInfoSet, j); + if (pActualFile->firstVer < pMetaFile->firstVer) { + char fNameStr[WAL_FILE_LEN]; + walBuildLogName(pWal, pActualFile->firstVer, fNameStr); + taosRemoveFile(fNameStr); + walBuildIdxName(pWal, pActualFile->firstVer, fNameStr); + taosRemoveFile(fNameStr); + i++; + } else if (pActualFile->firstVer > pMetaFile->firstVer) { + taosArrayRemove(pWal->fileInfoSet, j); + metaFileNum--; + } else { + i++; + j++; + } + } + if (i == actualFileNum && j == metaFileNum) { + if (j > 0) { + SWalFileInfo* pLastInfo = taosArrayGet(pWal->fileInfoSet, j - 1); + int64_t fsize = 0; + char fNameStr[WAL_FILE_LEN]; + walBuildLogName(pWal, pLastInfo->firstVer, fNameStr); + taosStatFile(fNameStr, &fsize, NULL); + if (pLastInfo->fileSize != fsize) { + fixed = true; + pLastInfo->fileSize = fsize; + pLastInfo->lastVer = walScanLogGetLastVer(pWal); + } + } + } else { + fixed = true; + while (i < actualFileNum) { + SWalFileInfo* pActualFile = taosArrayGet(actualLog, i); + char fNameStr[WAL_FILE_LEN]; + walBuildLogName(pWal, pActualFile->firstVer, fNameStr); + taosStatFile(fNameStr, &pActualFile->fileSize, NULL); + + if (pActualFile->fileSize == 0) { + ASSERT(i == actualFileNum - 1); + taosRemoveFile(fNameStr); + + walBuildIdxName(pWal, pActualFile->firstVer, fNameStr); + taosRemoveFile(fNameStr); + break; + } + + if (i < actualFileNum - 1) { + pActualFile->lastVer = ((SWalFileInfo*)taosArrayGet(actualLog, i + 1))->firstVer - 1; + taosArrayPush(pWal->fileInfoSet, pActualFile); + i++; + } else { + pActualFile = taosArrayPush(pWal->fileInfoSet, pActualFile); + pActualFile->lastVer = walScanLogGetLastVer(pWal); + if (pActualFile->lastVer == -1) { + taosRemoveFile(fNameStr); + + walBuildIdxName(pWal, pActualFile->firstVer, fNameStr); + taosRemoveFile(fNameStr); + taosArrayPop(pWal->fileInfoSet); + } + break; + } + } + } + } + +#if 0 if (metaFileNum > actualFileNum) { taosArrayPopFrontBatch(pWal->fileInfoSet, metaFileNum - actualFileNum); } else if (metaFileNum < actualFileNum) { @@ -214,30 +286,28 @@ int walCheckAndRepairMeta(SWal* pWal) { taosArrayPush(pWal->fileInfoSet, pFileInfo); } } +#endif + taosArrayDestroy(actualLog); + actualFileNum = taosArrayGetSize(pWal->fileInfoSet); pWal->writeCur = actualFileNum - 1; + if (actualFileNum > 0) { - pWal->vers.firstVer = ((SWalFileInfo*)taosArrayGet(pWal->fileInfoSet, 0))->firstVer; - - SWalFileInfo* pLastFileInfo = taosArrayGet(pWal->fileInfoSet, actualFileNum - 1); - char fnameStr[WAL_FILE_LEN]; - walBuildLogName(pWal, pLastFileInfo->firstVer, fnameStr); - int64_t fileSize = 0; - taosStatFile(fnameStr, &fileSize, NULL); - /*ASSERT(fileSize != 0);*/ - - if (metaFileNum != actualFileNum || pLastFileInfo->fileSize != fileSize) { - pLastFileInfo->fileSize = fileSize; - pWal->vers.lastVer = walScanLogGetLastVer(pWal); - ((SWalFileInfo*)taosArrayGetLast(pWal->fileInfoSet))->lastVer = pWal->vers.lastVer; - ASSERT(pWal->vers.lastVer != -1); - - int code = walSaveMeta(pWal); - if (code < 0) { - return -1; - } + int64_t fLastVer = ((SWalFileInfo*)taosArrayGet(pWal->fileInfoSet, pWal->writeCur))->lastVer; + if (fLastVer != -1 && pWal->vers.lastVer != fLastVer) { + fixed = true; + pWal->vers.lastVer = fLastVer; } + int64_t fFirstVer = ((SWalFileInfo*)taosArrayGet(pWal->fileInfoSet, 0))->firstVer; + if (fFirstVer != pWal->vers.firstVer) { + fixed = true; + pWal->vers.firstVer = fFirstVer; + } + } + + if (fixed) { + walSaveMeta(pWal); } return 0; @@ -530,6 +600,11 @@ int walLoadMeta(SWal* pWal) { // read metafile int64_t fileSize = 0; taosStatFile(fnameStr, &fileSize, NULL); + if (fileSize == 0) { + taosRemoveFile(fnameStr); + wDebug("vgId:%d wal find empty meta ver %d", pWal->cfg.vgId, metaVer); + return -1; + } int size = (int)fileSize; char* buf = taosMemoryMalloc(size + 5); if (buf == NULL) { @@ -540,6 +615,7 @@ int walLoadMeta(SWal* pWal) { TdFilePtr pFile = taosOpenFile(fnameStr, TD_FILE_READ); if (pFile == NULL) { terrno = TSDB_CODE_WAL_FILE_CORRUPTED; + taosMemoryFree(buf); return -1; } if (taosReadFile(pFile, buf, size) != size) { diff --git a/source/libs/wal/src/walRead.c b/source/libs/wal/src/walRead.c index cc6f827b8e..1acaf5e7f3 100644 --- a/source/libs/wal/src/walRead.c +++ b/source/libs/wal/src/walRead.c @@ -399,7 +399,7 @@ int32_t walFetchBody(SWalReader *pRead, SWalCkHead **ppHead) { terrno = TSDB_CODE_WAL_OUT_OF_MEMORY; return -1; } - *ppHead = ptr; + *ppHead = (SWalCkHead *)ptr; pReadHead = &((*ppHead)->head); pRead->capacity = pReadHead->bodyLen; }