From f3d7d6671b08f916e477907300f13024c70ba669 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Mon, 18 Jul 2022 20:06:29 +0800 Subject: [PATCH 1/9] fix(query):filter invisible columns --- source/libs/executor/src/scanoperator.c | 41 ++++++++++++++++++++++--- 1 file changed, 37 insertions(+), 4 deletions(-) diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index e60f6f8a5b..4796f0405e 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -1196,7 +1196,6 @@ static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock if (pResCol->info.colId == pColMatchInfo->colId) { SColumnInfoData* pDst = taosArrayGet(pInfo->pRes->pDataBlock, pColMatchInfo->targetSlotId); colDataAssign(pDst, pResCol, pBlock->info.rows, &pInfo->pRes->info); - // taosArraySet(pInfo->pRes->pDataBlock, pColMatchInfo->targetSlotId, pResCol); colExists = true; break; } @@ -1225,6 +1224,40 @@ static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock return 0; } +// todo opt perf +static SSDataBlock* doDropInvisibleCol(SSDataBlock* pBlock, SArray* pColMatchInfo) { + size_t numOfMatchInfo = taosArrayGetSize(pColMatchInfo); + + bool ignoreCols = false; + for (int32_t j = 0; j < numOfMatchInfo; ++j) { + SColMatchInfo* pInfo = taosArrayGet(pColMatchInfo, j); + if (!pInfo->output) { + ignoreCols = true; + break; + } + } + + SSDataBlock* pRes = createOneDataBlock(pBlock, false); + if (ignoreCols) { + int32_t i = 0; + while(i < taosArrayGetSize(pRes->pDataBlock)) { + SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, i); + for (int32_t j = 0; j < numOfMatchInfo; ++j) { + SColMatchInfo* pInfo = taosArrayGet(pColMatchInfo, j); + if (pInfo->colId == pCol->info.colId && !pInfo->output) { + taosArrayRemove(pBlock->pDataBlock, i); + i -= 1; + break; + } + } + + i += 1; + } + } + + return pRes; +} + static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { // NOTE: this operator does never check if current status is done or not SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; @@ -1435,8 +1468,9 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { } } } + qDebug("scan rows: %d", pBlockInfo->rows); - return (pBlockInfo->rows == 0) ? NULL : pInfo->pRes; + return (pBlockInfo->rows == 0) ? NULL : doDropInvisibleCol(pInfo->pRes, pInfo->pColMatchInfo); } else { ASSERT(0); @@ -1507,8 +1541,7 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys goto _error; } - SScanPhysiNode* pScanPhyNode = &pTableScanNode->scan; - + SScanPhysiNode* pScanPhyNode = &pTableScanNode->scan; SDataBlockDescNode* pDescNode = pScanPhyNode->node.pOutputDataBlockDesc; pInfo->pTagCond = pTagCond; From 484d043b4418af5e2cc9443b3645dd116c2c8d4a Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Mon, 18 Jul 2022 20:58:52 +0800 Subject: [PATCH 2/9] fix(query):filter invisible columns --- source/libs/executor/src/scanoperator.c | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 4796f0405e..4ff0e6b936 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -1237,7 +1237,7 @@ static SSDataBlock* doDropInvisibleCol(SSDataBlock* pBlock, SArray* pColMatchInf } } - SSDataBlock* pRes = createOneDataBlock(pBlock, false); + SSDataBlock* pRes = createOneDataBlock(pBlock, true); if (ignoreCols) { int32_t i = 0; while(i < taosArrayGetSize(pRes->pDataBlock)) { @@ -1255,10 +1255,11 @@ static SSDataBlock* doDropInvisibleCol(SSDataBlock* pBlock, SArray* pColMatchInf } } + pRes->info.rows = pBlock->info.rows; return pRes; } -static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { +static SSDataBlock* doStreamScanImpl(SOperatorInfo* pOperator) { // NOTE: this operator does never check if current status is done or not SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SStreamScanInfo* pInfo = pOperator->info; @@ -1470,7 +1471,7 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { } qDebug("scan rows: %d", pBlockInfo->rows); - return (pBlockInfo->rows == 0) ? NULL : doDropInvisibleCol(pInfo->pRes, pInfo->pColMatchInfo); + return (pBlockInfo->rows == 0) ? NULL : pInfo->pRes; } else { ASSERT(0); @@ -1478,6 +1479,16 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { } } +static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { + SSDataBlock* pBlock = doStreamScanImpl(pOperator); + if (pBlock != NULL) { + SStreamScanInfo* pInfo = (SStreamScanInfo*) pOperator->info; + return doDropInvisibleCol(pInfo->pRes, pInfo->pColMatchInfo); + } else { + return NULL; + } +} + static SSDataBlock* doRawScan(SOperatorInfo* pInfo) { // return NULL; From 2f199bd80b7a112b4093abb56fcbeb0592a62f92 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Tue, 19 Jul 2022 10:28:29 +0800 Subject: [PATCH 3/9] fix(query):revoke remove operation --- source/libs/executor/src/scanoperator.c | 47 +------------------------ 1 file changed, 1 insertion(+), 46 deletions(-) diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 4ff0e6b936..946057e0d9 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -1224,42 +1224,7 @@ static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock return 0; } -// todo opt perf -static SSDataBlock* doDropInvisibleCol(SSDataBlock* pBlock, SArray* pColMatchInfo) { - size_t numOfMatchInfo = taosArrayGetSize(pColMatchInfo); - - bool ignoreCols = false; - for (int32_t j = 0; j < numOfMatchInfo; ++j) { - SColMatchInfo* pInfo = taosArrayGet(pColMatchInfo, j); - if (!pInfo->output) { - ignoreCols = true; - break; - } - } - - SSDataBlock* pRes = createOneDataBlock(pBlock, true); - if (ignoreCols) { - int32_t i = 0; - while(i < taosArrayGetSize(pRes->pDataBlock)) { - SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, i); - for (int32_t j = 0; j < numOfMatchInfo; ++j) { - SColMatchInfo* pInfo = taosArrayGet(pColMatchInfo, j); - if (pInfo->colId == pCol->info.colId && !pInfo->output) { - taosArrayRemove(pBlock->pDataBlock, i); - i -= 1; - break; - } - } - - i += 1; - } - } - - pRes->info.rows = pBlock->info.rows; - return pRes; -} - -static SSDataBlock* doStreamScanImpl(SOperatorInfo* pOperator) { +static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { // NOTE: this operator does never check if current status is done or not SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SStreamScanInfo* pInfo = pOperator->info; @@ -1479,16 +1444,6 @@ static SSDataBlock* doStreamScanImpl(SOperatorInfo* pOperator) { } } -static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { - SSDataBlock* pBlock = doStreamScanImpl(pOperator); - if (pBlock != NULL) { - SStreamScanInfo* pInfo = (SStreamScanInfo*) pOperator->info; - return doDropInvisibleCol(pInfo->pRes, pInfo->pColMatchInfo); - } else { - return NULL; - } -} - static SSDataBlock* doRawScan(SOperatorInfo* pInfo) { // return NULL; From 5e754a7662e8efa77ee806eafb9b8d808e800095 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Tue, 19 Jul 2022 14:00:45 +0800 Subject: [PATCH 4/9] fix(query): add check for deleting record --- source/dnode/vnode/src/tsdb/tsdbRead.c | 69 +++++++++++++++++++------- 1 file changed, 51 insertions(+), 18 deletions(-) diff --git a/source/dnode/vnode/src/tsdb/tsdbRead.c b/source/dnode/vnode/src/tsdb/tsdbRead.c index 0b05bb7224..2801e09f06 100644 --- a/source/dnode/vnode/src/tsdb/tsdbRead.c +++ b/source/dnode/vnode/src/tsdb/tsdbRead.c @@ -1453,38 +1453,71 @@ static bool keyOverlapFileBlock(TSDBKEY key, SBlock* pBlock, SVersionRange* pVer (pBlock->minVersion <= pVerRange->maxVer); } -static bool overlapWithDelSkyline(STableBlockScanInfo* pBlockScanInfo, const SBlock* pBlock, int32_t order) { - if (pBlockScanInfo->delSkyline == NULL) { - return false; - } - TSDBKEY* pFirst = taosArrayGet(pBlockScanInfo->delSkyline, 0); - TSDBKEY* pLast = taosArrayGetLast(pBlockScanInfo->delSkyline); - - // ts is not overlap - if (pBlock->minKey.ts > pLast->ts || pBlock->maxKey.ts < pFirst->ts) { - return false; - } - - int32_t step = ASCENDING_TRAVERSE(order) ? 1 : -1; - - // version is not overlap +static bool doCheckforDatablockOverlap(STableBlockScanInfo* pBlockScanInfo, const SBlock* pBlock) { size_t num = taosArrayGetSize(pBlockScanInfo->delSkyline); - for (int32_t i = pBlockScanInfo->fileDelIndex; i < num; i += step) { + + for (int32_t i = pBlockScanInfo->fileDelIndex; i < num; i += 1) { TSDBKEY* p = taosArrayGet(pBlockScanInfo->delSkyline, i); if (p->ts >= pBlock->minKey.ts && p->ts <= pBlock->maxKey.ts) { if (p->version >= pBlock->minVersion) { return true; } - } else if (p->ts > pBlock->maxKey.ts) { + } else if (p->ts < pBlock->minKey.ts) { // p->ts < pBlock->minKey.ts + if (p->version >= pBlock->minVersion) { + if (i < num - 1) { + TSDBKEY* pnext = taosArrayGet(pBlockScanInfo->delSkyline, i + 1); + if (i + 1 == num - 1) { // pnext is the last point + if (pnext->ts >= pBlock->minKey.ts) { + return true; + } + } else { + if (pnext->ts >= pBlock->minKey.ts && pnext->version >= pBlock->minVersion) { + return true; + } + } + } else { // it must be the last point + ASSERT(p->version == 0); + } + } + } else { // (p->ts > pBlock->maxKey.ts) { return false; } } - ASSERT(0); return false; } +static bool overlapWithDelSkyline(STableBlockScanInfo* pBlockScanInfo, const SBlock* pBlock, int32_t order) { + if (pBlockScanInfo->delSkyline == NULL) { + return false; + } + + // ts is not overlap + TSDBKEY* pFirst = taosArrayGet(pBlockScanInfo->delSkyline, 0); + TSDBKEY* pLast = taosArrayGetLast(pBlockScanInfo->delSkyline); + if (pBlock->minKey.ts > pLast->ts || pBlock->maxKey.ts < pFirst->ts) { + return false; + } + + // version is not overlap + if (ASCENDING_TRAVERSE(order)) { + return doCheckforDatablockOverlap(pBlockScanInfo, pBlock); + } else { + int32_t index = pBlockScanInfo->fileDelIndex; + while(1) { + TSDBKEY* p = taosArrayGet(pBlockScanInfo->delSkyline, index); + if (p->ts > pBlock->minKey.ts && index > 0) { + index -= 1; + } else { // find the first point that is smaller than the minKey.ts of dataBlock. + break; + } + } + + return doCheckforDatablockOverlap(pBlockScanInfo, pBlock); + } +} + // 1. the version of all rows should be less than the endVersion // 2. current block should not overlap with next neighbor block // 3. current timestamp should not be overlap with each other From 329ba3cee87d0d604d3a8052b125f901d6fd8612 Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Tue, 19 Jul 2022 14:34:39 +0800 Subject: [PATCH 5/9] refactor(sync): add trace log --- source/libs/sync/src/syncRespMgr.c | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/source/libs/sync/src/syncRespMgr.c b/source/libs/sync/src/syncRespMgr.c index 97e5816038..3c63b7692a 100644 --- a/source/libs/sync/src/syncRespMgr.c +++ b/source/libs/sync/src/syncRespMgr.c @@ -120,15 +120,18 @@ void syncRespClean(SSyncRespMgr *pObj) { void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl) { SRespStub *pStub = (SRespStub *)taosHashIterate(pObj->pRespHash, NULL); int cnt = 0; + int sum = 0; SSyncNode *pSyncNode = pObj->data; SArray *delIndexArray = taosArrayInit(0, sizeof(uint64_t)); ASSERT(delIndexArray != NULL); + sDebug("vgId:%d, resp mgr begin clean by ttl", pSyncNode->vgId); while (pStub) { size_t len; - void * key = taosHashGetKey(pStub, &len); + void *key = taosHashGetKey(pStub, &len); uint64_t *pSeqNum = (uint64_t *)key; + sum++; int64_t nowMS = taosGetTimestampMs(); if (nowMS - pStub->createTime > ttl) { @@ -155,7 +158,7 @@ void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl) { } int32_t arraySize = taosArrayGetSize(delIndexArray); - sDebug("vgId:%d, resp mgr clean by ttl, cnt:%d, array-size:%d", pSyncNode->vgId, cnt, arraySize); + sDebug("vgId:%d, resp mgr end clean by ttl, sum:%d, cnt:%d, array-size:%d", pSyncNode->vgId, sum, cnt, arraySize); for (int32_t i = 0; i < arraySize; ++i) { uint64_t *pSeqNum = taosArrayGet(delIndexArray, i); From ed51c8e329aa8c23887813ddb012fe9a8a5f961f Mon Sep 17 00:00:00 2001 From: Ganlin Zhao Date: Tue, 19 Jul 2022 14:54:18 +0800 Subject: [PATCH 6/9] fix(query): fix mode function processing all null column TD-17024 --- source/libs/function/src/builtinsimpl.c | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c index 9d40b15354..25648fdea4 100644 --- a/source/libs/function/src/builtinsimpl.c +++ b/source/libs/function/src/builtinsimpl.c @@ -5098,12 +5098,7 @@ bool modeFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResInfo) { return true; } -static void doModeAdd(SModeInfo* pInfo, char* data, bool isNull) { - // ignore null elements - if (isNull) { - return; - } - +static void doModeAdd(SModeInfo* pInfo, char* data) { int32_t hashKeyBytes = IS_VAR_DATA_TYPE(pInfo->colType) ? varDataTLen(data) : pInfo->colBytes; SModeItem** pHashItem = taosHashGet(pInfo->pHash, data, hashKeyBytes); if (pHashItem == NULL) { @@ -5128,10 +5123,16 @@ int32_t modeFunction(SqlFunctionCtx* pCtx) { SColumnInfoData* pInputCol = pInput->pData[0]; SColumnInfoData* pOutput = (SColumnInfoData*)pCtx->pOutput; + int32_t numOfElems = 0; int32_t startOffset = pCtx->offset; for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; ++i) { char* data = colDataGetData(pInputCol, i); - doModeAdd(pInfo, data, colDataIsNull_s(pInputCol, i)); + if (colDataIsNull_s(pInputCol, i)) { + continue; + } + + numOfElems++; + doModeAdd(pInfo, data); if (sizeof(SModeInfo) + pInfo->numOfPoints * (sizeof(SModeItem) + pInfo->colBytes) >= MODE_MAX_RESULT_SIZE) { taosHashCleanup(pInfo->pHash); @@ -5139,7 +5140,7 @@ int32_t modeFunction(SqlFunctionCtx* pCtx) { } } - SET_VAL(pResInfo, 1, 1); + SET_VAL(pResInfo, numOfElems, 1); return TSDB_CODE_SUCCESS; } From 622ab28e6e967cea8417a6cd6262e5b49839ac09 Mon Sep 17 00:00:00 2001 From: afwerar <1296468573@qq.com> Date: Tue, 19 Jul 2022 14:59:07 +0800 Subject: [PATCH 7/9] os: backend run fail error --- source/common/src/tglobal.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index c4f3ca0bdf..fcc27e440c 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -556,7 +556,7 @@ static void taosSetSystemCfg(SConfig *pCfg) { osSetSystemLocale(locale, charset); bool enableCore = cfgGetItem(pCfg, "enableCoreFile")->bval; - taosSetConsoleEcho(enableCore); + taosSetCoreDump(enableCore); // todo tsVersion = 30000000; @@ -675,7 +675,7 @@ int32_t taosSetCfg(SConfig *pCfg, char *name) { case 'e': { if (strcasecmp("enableCoreFile", name) == 0) { bool enableCore = cfgGetItem(pCfg, "enableCoreFile")->bval; - taosSetConsoleEcho(enableCore); + taosSetCoreDump(enableCore); } break; } From 0d33e7363faa355ea500bfa4397844aa9677fe18 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Tue, 19 Jul 2022 15:41:31 +0800 Subject: [PATCH 8/9] fix: read sdb when restore from raft module may cause crash --- source/dnode/mnode/sdb/src/sdbHash.c | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/source/dnode/mnode/sdb/src/sdbHash.c b/source/dnode/mnode/sdb/src/sdbHash.c index 71792a2354..c579f82a9d 100644 --- a/source/dnode/mnode/sdb/src/sdbHash.c +++ b/source/dnode/mnode/sdb/src/sdbHash.c @@ -153,23 +153,22 @@ static int32_t sdbInsertRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow * return terrno; } - taosThreadRwlockUnlock(pLock); - int32_t code = 0; SdbInsertFp insertFp = pSdb->insertFps[pRow->type]; if (insertFp != NULL) { code = (*insertFp)(pSdb, pRow->pObj); if (code != 0) { code = terrno; - taosThreadRwlockWrlock(pLock); taosHashRemove(hash, pRow->pObj, keySize); - taosThreadRwlockUnlock(pLock); sdbFreeRow(pSdb, pRow, false); terrno = code; + taosThreadRwlockUnlock(pLock); return terrno; } } + taosThreadRwlockUnlock(pLock); + if (pSdb->keyTypes[pRow->type] == SDB_KEY_INT32) { pSdb->maxId[pRow->type] = TMAX(pSdb->maxId[pRow->type], *((int32_t *)pRow->pObj)); } @@ -194,7 +193,6 @@ static int32_t sdbUpdateRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow * SSdbRow *pOldRow = *ppOldRow; pOldRow->status = pRaw->status; sdbPrintOper(pSdb, pOldRow, "update"); - taosThreadRwlockUnlock(pLock); int32_t code = 0; SdbUpdateFp updateFp = pSdb->updateFps[pNewRow->type]; @@ -202,6 +200,7 @@ static int32_t sdbUpdateRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow * code = (*updateFp)(pSdb, pOldRow->pObj, pNewRow->pObj); } + taosThreadRwlockUnlock(pLock); sdbFreeRow(pSdb, pNewRow, false); pSdb->tableVer[pOldRow->type]++; From c13e638de775f8794c34640dc46a0d1231ad07cb Mon Sep 17 00:00:00 2001 From: Hui Li <52318143+plum-lihui@users.noreply.github.com> Date: Tue, 19 Jul 2022 16:04:13 +0800 Subject: [PATCH 9/9] Delete stbTagFilter.py --- tests/system-test/7-tmq/stbTagFilter.py | 260 ------------------------ 1 file changed, 260 deletions(-) delete mode 100644 tests/system-test/7-tmq/stbTagFilter.py diff --git a/tests/system-test/7-tmq/stbTagFilter.py b/tests/system-test/7-tmq/stbTagFilter.py deleted file mode 100644 index 1bb3d24bde..0000000000 --- a/tests/system-test/7-tmq/stbTagFilter.py +++ /dev/null @@ -1,260 +0,0 @@ - -import taos -import sys -import time -import socket -import os -import threading -from enum import Enum - -from util.log import * -from util.sql import * -from util.cases import * -from util.dnodes import * -sys.path.append("./7-tmq") -from tmqCommon import * - -class TDTestCase: - def __init__(self): - self.snapshot = 0 - self.vgroups = 4 - self.ctbNum = 1 - self.rowsPerTbl = 10000 - - def init(self, conn, logSql): - tdLog.debug(f"start to excute {__file__}") - tdSql.init(conn.cursor(), False) - - def prepareTestEnv(self): - tdLog.printNoPrefix("======== prepare test env include database, stable, ctables, and insert data: ") - paraDict = {'dbName': 'dbt', - 'dropFlag': 1, - 'event': '', - 'vgroups': 4, - 'stbName': 'stb', - 'colPrefix': 'c', - 'tagPrefix': 't', - 'colSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1},{'type': 'TIMESTAMP', 'count':1}], - 'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}], - 'ctbPrefix': 'ctb', - 'ctbStartIdx': 0, - 'ctbNum': 1, - 'rowsPerTbl': 100000, - 'batchNum': 1200, - 'startTs': 1640966400000, # 2022-01-01 00:00:00.000 - 'pollDelay': 3, - 'showMsg': 1, - 'showRow': 1, - 'snapshot': 0} - - paraDict['vgroups'] = self.vgroups - paraDict['ctbNum'] = self.ctbNum - paraDict['rowsPerTbl'] = self.rowsPerTbl - - tmqCom.initConsumerTable() - tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=1) - tdLog.info("create stb") - tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"]) - tdLog.info("create ctb") - tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict['ctbPrefix'], - ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx']) - tdLog.info("insert data") - tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"], - ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"], - startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx']) - # tmqCom.insert_data_with_autoCreateTbl(tsql=tdSql,dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix="ctbx", - # ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"], - # startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx']) - - # tdLog.info("restart taosd to ensure that the data falls into the disk") - # tdSql.query("flush database %s"%(paraDict['dbName'])) - return - - def tmqCase1(self): - tdLog.printNoPrefix("======== test case 1: ") - paraDict = {'dbName': 'dbt', - 'dropFlag': 1, - 'event': '', - 'vgroups': 4, - 'stbName': 'stb', - 'colPrefix': 'c', - 'tagPrefix': 't', - 'colSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1},{'type': 'TIMESTAMP', 'count':1}], - 'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}], - 'ctbPrefix': 'ctb', - 'ctbStartIdx': 0, - 'ctbNum': 1, - 'rowsPerTbl': 100000, - 'batchNum': 3000, - 'startTs': 1640966400000, # 2022-01-01 00:00:00.000 - 'pollDelay': 5, - 'showMsg': 1, - 'showRow': 1, - 'snapshot': 0} - paraDict['snapshot'] = self.snapshot - paraDict['vgroups'] = self.vgroups - paraDict['ctbNum'] = self.ctbNum - paraDict['rowsPerTbl'] = self.rowsPerTbl - - # update to half tables - # paraDict['rowsPerTbl'] = int(self.rowsPerTbl / 2) - # tmqCom.insert_data_with_autoCreateTbl(tsql=tdSql,dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix="ctbx", - # ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"], - # startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx']) - # tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"], - # ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"], - # startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx']) - - tdLog.info("create topics from stb1") - topicFromStb1 = 'topic_stb1' - queryString = "select ts, c1, c2 from %s.%s where t4 == 'shanghai' or t4 == 'changsha'"%(paraDict['dbName'], paraDict['stbName']) - # queryString = "select ts, c1, c2, t4 from %s.%s where t4 == 'shanghai' or t4 == 'changsha'"%(paraDict['dbName'], paraDict['stbName']) - sqlString = "create topic %s as %s" %(topicFromStb1, queryString) - tdLog.info("create topic sql: %s"%sqlString) - tdSql.execute(sqlString) - - # paraDict['ctbNum'] = self.ctbNum - paraDict['rowsPerTbl'] = self.rowsPerTbl - consumerId = 0 - expectrowcnt = int(paraDict["rowsPerTbl"] * paraDict["ctbNum"] * 2) - topicList = topicFromStb1 - ifcheckdata = 1 - ifManualCommit = 1 - keyList = 'group.id:cgrp1,\ - enable.auto.commit:true,\ - auto.commit.interval.ms:1000,\ - auto.offset.reset:earliest' - tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit) - - tdLog.info("start consume processor") - tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot']) - - tdLog.info("insert process end, and start to check consume result") - expectRows = 1 - resultList = tmqCom.selectConsumeResult(expectRows) - totalConsumeRows = 0 - for i in range(expectRows): - totalConsumeRows += resultList[i] - - tdLog.info("run select sql from db") - tdSql.query(queryString) - expectrowcnt = tdSql.getRows() - - tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt)) - if totalConsumeRows != expectrowcnt: - tdLog.exit("tmq consume rows error!") - - tmqCom.checkFileContent(consumerId, queryString) - - tdSql.query("drop topic %s"%topicFromStb1) - tdLog.printNoPrefix("======== test case 1 end ...... ") - - def tmqCase2(self): - tdLog.printNoPrefix("======== test case 2: ") - paraDict = {'dbName': 'dbt', - 'dropFlag': 1, - 'event': '', - 'vgroups': 4, - 'stbName': 'stb', - 'colPrefix': 'c', - 'tagPrefix': 't', - 'colSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1},{'type': 'TIMESTAMP', 'count':1}], - 'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}], - 'ctbPrefix': 'ctb', - 'ctbStartIdx': 0, - 'ctbNum': 1, - 'rowsPerTbl': 10000, - 'batchNum': 5000, - 'startTs': 1640966400000, # 2022-01-01 00:00:00.000 - 'pollDelay': 5, - 'showMsg': 1, - 'showRow': 1, - 'snapshot': 0} - - paraDict['snapshot'] = self.snapshot - paraDict['vgroups'] = self.vgroups - paraDict['ctbNum'] = self.ctbNum - paraDict['rowsPerTbl'] = self.rowsPerTbl - - tdLog.info("restart taosd to ensure that the data falls into the disk") - tdSql.query("flush database %s"%(paraDict['dbName'])) - - # update to half tables - paraDict['startTs'] = paraDict['startTs'] + int(self.rowsPerTbl / 2) - paraDict['rowsPerTbl'] = int(self.rowsPerTbl / 2) - tmqCom.insert_data_with_autoCreateTbl(tsql=tdSql,dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict["ctbPrefix"], - ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"], - startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx']) - # tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"], - # ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"], - # startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx']) - - tmqCom.initConsumerTable() - tdLog.info("create topics from stb1") - topicFromStb1 = 'topic_stb1' - queryString = "select ts, c1, c2 from %s.%s"%(paraDict['dbName'], paraDict['stbName']) - sqlString = "create topic %s as %s" %(topicFromStb1, queryString) - tdLog.info("create topic sql: %s"%sqlString) - tdSql.execute(sqlString) - - # paraDict['ctbNum'] = self.ctbNum - paraDict['rowsPerTbl'] = self.rowsPerTbl - consumerId = 1 - expectrowcnt = int(paraDict["rowsPerTbl"] * paraDict["ctbNum"] * 2) - topicList = topicFromStb1 - ifcheckdata = 1 - ifManualCommit = 1 - keyList = 'group.id:cgrp1,\ - enable.auto.commit:true,\ - auto.commit.interval.ms:1000,\ - auto.offset.reset:earliest' - tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit) - - tdLog.info("start consume processor") - tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot']) - - tdLog.info("insert process end, and start to check consume result") - expectRows = 1 - resultList = tmqCom.selectConsumeResult(expectRows) - totalConsumeRows = 0 - for i in range(expectRows): - totalConsumeRows += resultList[i] - - tdSql.query(queryString) - totalRowsInserted = tdSql.getRows() - - tdLog.info("act consume rows: %d, act insert rows: %d, expect consume rows: %d, "%(totalConsumeRows, totalRowsInserted, expectrowcnt)) - - if totalConsumeRows != expectrowcnt: - tdLog.exit("tmq consume rows error!") - - # tmqCom.checkFileContent(consumerId, queryString) - - tdSql.query("drop topic %s"%topicFromStb1) - - tdLog.printNoPrefix("======== test case 2 end ...... ") - - def run(self): - tdSql.prepare() - self.prepareTestEnv() - tdLog.printNoPrefix("=============================================") - tdLog.printNoPrefix("======== snapshot is 0: only consume from wal") - self.tmqCase1() - # self.tmqCase2() - - # self.prepareTestEnv() - # tdLog.printNoPrefix("====================================================================") - # tdLog.printNoPrefix("======== snapshot is 1: firstly consume from tsbs, and then from wal") - # self.snapshot = 1 - # self.tmqCase1() - # self.tmqCase2() - - - def stop(self): - tdSql.close() - tdLog.success(f"{__file__} successfully executed") - -event = threading.Event() - -tdCases.addLinux(__file__, TDTestCase()) -tdCases.addWindows(__file__, TDTestCase())