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; } 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]++; 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 diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 505d023492..bc526f8a31 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; } @@ -1435,6 +1434,7 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { } } } + qDebug("scan rows: %d", pBlockInfo->rows); return (pBlockInfo->rows == 0) ? NULL : pInfo->pRes; @@ -1507,8 +1507,7 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys goto _error; } - SScanPhysiNode* pScanPhyNode = &pTableScanNode->scan; - + SScanPhysiNode* pScanPhyNode = &pTableScanNode->scan; SDataBlockDescNode* pDescNode = pScanPhyNode->node.pOutputDataBlockDesc; pInfo->pTagCond = pTagCond; 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; } 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); 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())