diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 1e7847b4e89d402e2fc99935615243dcd78bc5cb..0b4e604a9db014e313e95df9e83aa20b068b65c5 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -1474,7 +1474,6 @@ _err: // this message is sent from mnode to mnode(read thread to write thread), so there is no need for serialization or // deserialization typedef struct { - int8_t* mqInReb; SHashObj* rebSubHash; // SHashObj } SMqDoRebalanceMsg; diff --git a/include/libs/function/function.h b/include/libs/function/function.h index 0580f3acba9e2563af70dcdc92a9cdb08fb5557b..094ce801062764cbe8e7e83a7cb1d52e21295c75 100644 --- a/include/libs/function/function.h +++ b/include/libs/function/function.h @@ -207,7 +207,7 @@ typedef struct SqlFunctionCtx { struct SSDataBlock *pSrcBlock; int32_t curBufPage; - char* udfName[TSDB_FUNC_NAME_LEN]; + char udfName[TSDB_FUNC_NAME_LEN]; } SqlFunctionCtx; enum { diff --git a/include/libs/function/functionMgt.h b/include/libs/function/functionMgt.h index 126a2ccf99c02daf1360d6fa551e3ed38726de77..0572262bfc99a2b64f381c6b148ef79fda6d4c0f 100644 --- a/include/libs/function/functionMgt.h +++ b/include/libs/function/functionMgt.h @@ -162,6 +162,7 @@ EFuncDataRequired fmFuncDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWin int32_t fmGetFuncExecFuncs(int32_t funcId, SFuncExecFuncs* pFpSet); int32_t fmGetScalarFuncExecFuncs(int32_t funcId, SScalarFuncExecFuncs* pFpSet); +int32_t fmGetUdafExecFuncs(int32_t funcId, SFuncExecFuncs* pFpSet); #ifdef __cplusplus } diff --git a/include/libs/function/tudf.h b/include/libs/function/tudf.h index 985bf6fa6fabec9161be7cfa10e9c8fd819edcd9..e49f5cac45d1563b620004f106fad51481483039 100644 --- a/include/libs/function/tudf.h +++ b/include/libs/function/tudf.h @@ -140,7 +140,7 @@ typedef int32_t (*TUdfFreeUdfColumnFunc)(SUdfColumn* column); typedef int32_t (*TUdfScalarProcFunc)(SUdfDataBlock* block, SUdfColumn *resultCol); typedef int32_t (*TUdfAggStartFunc)(SUdfInterBuf *buf); -typedef int32_t (*TUdfAggProcessFunc)(SUdfDataBlock* block, SUdfInterBuf *interBuf); +typedef int32_t (*TUdfAggProcessFunc)(SUdfDataBlock* block, SUdfInterBuf *interBuf, SUdfInterBuf *newInterBuf); typedef int32_t (*TUdfAggFinishFunc)(SUdfInterBuf* buf, SUdfInterBuf *resultData); diff --git a/source/dnode/mnode/impl/inc/mndConsumer.h b/source/dnode/mnode/impl/inc/mndConsumer.h index 19e202dddc0c56cc7679d7a3d319ab7918e36528..a8bfe91cbf8edfb225510253b9536115b21cea30 100644 --- a/source/dnode/mnode/impl/inc/mndConsumer.h +++ b/source/dnode/mnode/impl/inc/mndConsumer.h @@ -44,6 +44,11 @@ SSdbRow *mndConsumerActionDecode(SSdbRaw *pRaw); int32_t mndSetConsumerCommitLogs(SMnode *pMnode, STrans *pTrans, SMqConsumerObj *pConsumer); +bool mndRebTryStart(); +void mndRebEnd(); +void mndRebCntInc(); +void mndRebCntDec(); + #ifdef __cplusplus } #endif diff --git a/source/dnode/mnode/impl/inc/mndTrans.h b/source/dnode/mnode/impl/inc/mndTrans.h index 5b5aff7c86ef204a1e75d6b0d20a295c454ab57c..7644ec3c4cab709e10ed89b7443c5bca466f6f7e 100644 --- a/source/dnode/mnode/impl/inc/mndTrans.h +++ b/source/dnode/mnode/impl/inc/mndTrans.h @@ -36,8 +36,8 @@ typedef struct { typedef enum { TEST_TRANS_START_FUNC = 1, TEST_TRANS_STOP_FUNC = 2, - CONSUME_TRANS_START_FUNC = 3, - CONSUME_TRANS_STOP_FUNC = 4, + MQ_REB_TRANS_START_FUNC = 3, + MQ_REB_TRANS_STOP_FUNC = 4, } ETrnFuncType; typedef void (*TransCbFp)(SMnode *pMnode, void *param, int32_t paramLen); diff --git a/source/dnode/mnode/impl/src/mndConsumer.c b/source/dnode/mnode/impl/src/mndConsumer.c index 025f61dc87694d2a6f872f6200470d83445e8b7c..b17772bdb2184c32b92dd87911490222967b24fb 100644 --- a/source/dnode/mnode/impl/src/mndConsumer.c +++ b/source/dnode/mnode/impl/src/mndConsumer.c @@ -35,7 +35,7 @@ #define MND_CONSUMER_LOST_HB_CNT 3 -static int8_t mqInRebFlag = 0; +static int8_t mqRebLock = 0; static const char *mndConsumerStatusName(int status); @@ -75,6 +75,17 @@ int32_t mndInitConsumer(SMnode *pMnode) { void mndCleanupConsumer(SMnode *pMnode) {} +bool mndRebTryStart() { + int8_t old = atomic_val_compare_exchange_8(&mqRebLock, 0, 1); + return old == 0; +} + +void mndRebEnd() { atomic_sub_fetch_8(&mqRebLock, 1); } + +void mndRebCntInc() { atomic_add_fetch_8(&mqRebLock, 1); } + +void mndRebCntDec() { atomic_sub_fetch_8(&mqRebLock, 1); } + static int32_t mndProcessConsumerLostMsg(SNodeMsg *pMsg) { SMnode *pMnode = pMsg->pNode; SMqConsumerLostMsg *pLostMsg = pMsg->rpcMsg.pCont; @@ -143,8 +154,7 @@ static int32_t mndProcessMqTimerMsg(SNodeMsg *pMsg) { void *pIter = NULL; // rebalance cannot be parallel - int8_t old = atomic_val_compare_exchange_8(&mqInRebFlag, 0, 1); - if (old != 0) { + if (!mndRebTryStart()) { mInfo("mq rebalance already in progress, do nothing"); return 0; } @@ -152,7 +162,6 @@ static int32_t mndProcessMqTimerMsg(SNodeMsg *pMsg) { SMqDoRebalanceMsg *pRebMsg = rpcMallocCont(sizeof(SMqDoRebalanceMsg)); pRebMsg->rebSubHash = taosHashInit(64, MurmurHash3_32, true, HASH_NO_LOCK); // TODO set cleanfp - pRebMsg->mqInReb = &mqInRebFlag; // iterate all consumers, find all modification while (1) { @@ -223,7 +232,7 @@ static int32_t mndProcessMqTimerMsg(SNodeMsg *pMsg) { taosHashCleanup(pRebMsg->rebSubHash); rpcFreeCont(pRebMsg); mTrace("mq rebalance finished, no modification"); - atomic_store_8(&mqInRebFlag, 0); + mndRebEnd(); } return 0; } diff --git a/source/dnode/mnode/impl/src/mndScheduler.c b/source/dnode/mnode/impl/src/mndScheduler.c index 3f4b2fe145ea5121cf42a56275daf285de16279d..b760dd6aa8d584accd2569ca4605f61934cab12d 100644 --- a/source/dnode/mnode/impl/src/mndScheduler.c +++ b/source/dnode/mnode/impl/src/mndScheduler.c @@ -308,8 +308,7 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) { // sink part if (level == 0) { // only for inplace - pTask->sinkType = TASK_SINK__SHOW; - pTask->showSink.reserved = 0; + pTask->sinkType = TASK_SINK__NONE; if (!hasExtraSink) { #if 1 if (pStream->createdBy == STREAM_CREATED_BY__SMA) { @@ -368,8 +367,7 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) { pTask->sourceType = TASK_SOURCE__PIPE; // sink part - pTask->sinkType = TASK_SINK__SHOW; - /*pTask->sinkType = TASK_SINK__NONE;*/ + pTask->sinkType = TASK_SINK__NONE; // dispatch part ASSERT(hasExtraSink); @@ -456,7 +454,7 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) { pTask->sourceType = TASK_SOURCE__MERGE; // sink part - pTask->sinkType = TASK_SINK__SHOW; + pTask->sinkType = TASK_SINK__NONE; // dispatch part pTask->dispatchType = TASK_DISPATCH__NONE; diff --git a/source/dnode/mnode/impl/src/mndShow.c b/source/dnode/mnode/impl/src/mndShow.c index 94366a241d697025f0d0682d8b2bab24df12ad76..0a3789402b4a1e5b7ca0b1d7d7c85da72214dc09 100644 --- a/source/dnode/mnode/impl/src/mndShow.c +++ b/source/dnode/mnode/impl/src/mndShow.c @@ -47,7 +47,7 @@ void mndCleanupShow(SMnode *pMnode) { } } -static int32_t convertToRetrieveType(char* name, int32_t len) { +static int32_t convertToRetrieveType(char *name, int32_t len) { int32_t type = -1; if (strncasecmp(name, TSDB_INS_TABLE_DNODES, len) == 0) { @@ -72,8 +72,6 @@ static int32_t convertToRetrieveType(char* name, int32_t len) { // type = TSDB_MGMT_TABLE_INDEX; } else if (strncasecmp(name, TSDB_INS_TABLE_USER_STABLES, len) == 0) { type = TSDB_MGMT_TABLE_STB; - } else if (strncasecmp(name, TSDB_INS_TABLE_USER_STREAMS, len) == 0) { - type = TSDB_MGMT_TABLE_STREAMS; } else if (strncasecmp(name, TSDB_INS_TABLE_USER_TABLES, len) == 0) { type = TSDB_MGMT_TABLE_TABLE; } else if (strncasecmp(name, TSDB_INS_TABLE_USER_TABLE_DISTRIBUTED, len) == 0) { @@ -98,12 +96,14 @@ static int32_t convertToRetrieveType(char* name, int32_t len) { type = TSDB_MGMT_TABLE_CONNS; } else if (strncasecmp(name, TSDB_INS_TABLE_QUERIES, len) == 0) { type = TSDB_MGMT_TABLE_QUERIES; - } else if (strncasecmp(name, TSDB_INS_TABLE_VNODES, len) == 0) { + } else if (strncasecmp(name, TSDB_INS_TABLE_VNODES, len) == 0) { type = TSDB_MGMT_TABLE_VNODES; } else if (strncasecmp(name, TSDB_PERFS_TABLE_TOPICS, len) == 0) { type = TSDB_MGMT_TABLE_TOPICS; + } else if (strncasecmp(name, TSDB_PERFS_TABLE_STREAMS, len) == 0) { + type = TSDB_MGMT_TABLE_STREAMS; } else { -// ASSERT(0); + // ASSERT(0); } return type; @@ -115,12 +115,12 @@ static SShowObj *mndCreateShowObj(SMnode *pMnode, SRetrieveTableReq *pReq) { int64_t showId = atomic_add_fetch_64(&pMgmt->showId, 1); if (showId == 0) atomic_add_fetch_64(&pMgmt->showId, 1); - int32_t size = sizeof(SShowObj); + int32_t size = sizeof(SShowObj); SShowObj showObj = {0}; - showObj.id = showId; + showObj.id = showId; showObj.pMnode = pMnode; - showObj.type = convertToRetrieveType(pReq->tb, tListLen(pReq->tb)); + showObj.type = convertToRetrieveType(pReq->tb, tListLen(pReq->tb)); memcpy(showObj.db, pReq->db, TSDB_DB_FNAME_LEN); int32_t keepTime = tsShellActivityTimer * 6 * 1000; diff --git a/source/dnode/mnode/impl/src/mndSubscribe.c b/source/dnode/mnode/impl/src/mndSubscribe.c index aeecab34cb0e61a3251dc7506ec5b174a69109d9..2b3af85066b8658d23ca05a7ef1220299c5db807 100644 --- a/source/dnode/mnode/impl/src/mndSubscribe.c +++ b/source/dnode/mnode/impl/src/mndSubscribe.c @@ -452,7 +452,10 @@ static int32_t mndPersistRebResult(SMnode *pMnode, SNodeMsg *pMsg, const SMqRebO } // 4. TODO commit log: modification log - // 5. execution + // 5. set cb + mndTransSetCb(pTrans, MQ_REB_TRANS_START_FUNC, MQ_REB_TRANS_STOP_FUNC, NULL, 0); + + // 6. execution if (mndTransPrepare(pMnode, pTrans) != 0) goto REB_FAIL; mndTransDrop(pTrans); @@ -518,9 +521,9 @@ static int32_t mndProcessRebalanceReq(SNodeMsg *pMsg) { } // reset flag - atomic_store_8(pReq->mqInReb, 0); mInfo("mq rebalance completed successfully"); taosHashCleanup(pReq->rebSubHash); + mndRebEnd(); return 0; } diff --git a/source/dnode/mnode/impl/src/mndTrans.c b/source/dnode/mnode/impl/src/mndTrans.c index c08b0f6db94b19f270f608690bca875c0664a763..2d10c4a7a5623cb35fc161b882883c690b917461 100644 --- a/source/dnode/mnode/impl/src/mndTrans.c +++ b/source/dnode/mnode/impl/src/mndTrans.c @@ -16,6 +16,7 @@ #define _DEFAULT_SOURCE #include "mndTrans.h" #include "mndAuth.h" +#include "mndConsumer.h" #include "mndDb.h" #include "mndShow.h" #include "mndSync.h" @@ -442,6 +443,10 @@ static TransCbFp mndTransGetCbFp(ETrnFuncType ftype) { return mndTransTestStartFunc; case TEST_TRANS_STOP_FUNC: return mndTransTestStopFunc; + case MQ_REB_TRANS_START_FUNC: + return mndRebCntInc; + case MQ_REB_TRANS_STOP_FUNC: + return mndRebCntDec; default: return NULL; } diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index f1995b723d906f6698a69024364e1b5702db8538..943a016c2741c885d8c7b4bdb40a7967dcda67b1 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -1897,7 +1897,14 @@ SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput, pCtx->functionId = pExpr->pExpr->_function.pFunctNode->funcId; if (fmIsAggFunc(pCtx->functionId) || fmIsNonstandardSQLFunc(pCtx->functionId)) { - fmGetFuncExecFuncs(pCtx->functionId, &pCtx->fpSet); + bool isUdaf = fmIsUserDefinedFunc(pCtx->functionId); + if (!isUdaf) { + fmGetFuncExecFuncs(pCtx->functionId, &pCtx->fpSet); + } else { + char *udfName = pExpr->pExpr->_function.pFunctNode->functionName; + strncpy(pCtx->udfName, udfName, strlen(udfName)); + fmGetUdafExecFuncs(pCtx->functionId, &pCtx->fpSet); + } pCtx->fpSet.getEnv(pExpr->pExpr->_function.pFunctNode, &env); } else { fmGetScalarFuncExecFuncs(pCtx->functionId, &pCtx->sfp); diff --git a/source/libs/function/src/functionMgt.c b/source/libs/function/src/functionMgt.c index 585eb57a56e5954d756e10587ff5fb551f8257b6..0113da94eb08a419ff05d2c48768438d9acf5e0b 100644 --- a/source/libs/function/src/functionMgt.c +++ b/source/libs/function/src/functionMgt.c @@ -124,7 +124,10 @@ int32_t fmGetFuncExecFuncs(int32_t funcId, SFuncExecFuncs* pFpSet) { return TSDB_CODE_SUCCESS; } -int32_t fmGetUdafExecFuncs(SFuncExecFuncs* pFpSet) { +int32_t fmGetUdafExecFuncs(int32_t funcId, SFuncExecFuncs* pFpSet) { + if (!fmIsUserDefinedFunc(funcId)) { + return TSDB_CODE_FAILED; + } pFpSet->getEnv = udfAggGetEnv; pFpSet->init = udfAggInit; pFpSet->process = udfAggProcess; diff --git a/source/libs/function/src/udfd.c b/source/libs/function/src/udfd.c index 896ebd3763ec8b73de883c4db16dd632ecd53ba1..ae24a832c8cea8d72d4f621c4802060aecfc96f9 100644 --- a/source/libs/function/src/udfd.c +++ b/source/libs/function/src/udfd.c @@ -232,7 +232,7 @@ void udfdProcessRequest(uv_work_t *req) { SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize), .bufLen= udf->bufSize, .numOfResult = 0}; - udf->aggProcFunc(&input, &outBuf); + udf->aggProcFunc(&input, &call->interBuf, &outBuf); subRsp->resultBuf = outBuf; break; diff --git a/source/libs/function/test/udf2.c b/source/libs/function/test/udf2.c index 250c20ba88d90e612f76fa466b115b4697e3c885..83187c5855650437a215901126b22975baa6216d 100644 --- a/source/libs/function/test/udf2.c +++ b/source/libs/function/test/udf2.c @@ -24,7 +24,7 @@ int32_t udf2_start(SUdfInterBuf *buf) { return 0; } -int32_t udf2(SUdfDataBlock* block, SUdfInterBuf *interBuf) { +int32_t udf2(SUdfDataBlock* block, SUdfInterBuf *interBuf, SUdfInterBuf *newInterBuf) { int64_t sumSquares = *(int64_t*)interBuf->buf; for (int32_t i = 0; i < block->numOfCols; ++i) { for (int32_t j = 0; j < block->numOfRows; ++i) { @@ -35,10 +35,10 @@ int32_t udf2(SUdfDataBlock* block, SUdfInterBuf *interBuf) { } } - *(int64_t*)interBuf = sumSquares; - interBuf->bufLen = sizeof(int64_t); + *(int64_t*)newInterBuf = sumSquares; + newInterBuf->bufLen = sizeof(int64_t); //TODO: if all null value, numOfResult = 0; - interBuf->numOfResult = 1; + newInterBuf->numOfResult = 1; return 0; } diff --git a/source/libs/stream/src/tstream.c b/source/libs/stream/src/tstream.c index 8aaaa414ca265f31f09f26f877b2ec3dea7ce5b3..b06c774ed3cb289afb5b08073f1d3dc9602b6bda 100644 --- a/source/libs/stream/src/tstream.c +++ b/source/libs/stream/src/tstream.c @@ -158,8 +158,6 @@ int32_t streamExecTask(SStreamTask* pTask, SMsgCb* pMsgCb, const void* input, in // } else if (pTask->sinkType == TASK_SINK__FETCH) { // - } else if (pTask->sinkType == TASK_SINK__SHOW) { - blockDebugShowData(pRes); } else { ASSERT(pTask->sinkType == TASK_SINK__NONE); } @@ -280,8 +278,6 @@ int32_t tEncodeSStreamTask(SCoder* pEncoder, const SStreamTask* pTask) { if (tEncodeI64(pEncoder, pTask->smaSink.smaId) < 0) return -1; } else if (pTask->sinkType == TASK_SINK__FETCH) { if (tEncodeI8(pEncoder, pTask->fetchSink.reserved) < 0) return -1; - } else if (pTask->sinkType == TASK_SINK__SHOW) { - if (tEncodeI8(pEncoder, pTask->showSink.reserved) < 0) return -1; } else { ASSERT(pTask->sinkType == TASK_SINK__NONE); } @@ -326,8 +322,6 @@ int32_t tDecodeSStreamTask(SCoder* pDecoder, SStreamTask* pTask) { if (tDecodeI64(pDecoder, &pTask->smaSink.smaId) < 0) return -1; } else if (pTask->sinkType == TASK_SINK__FETCH) { if (tDecodeI8(pDecoder, &pTask->fetchSink.reserved) < 0) return -1; - } else if (pTask->sinkType == TASK_SINK__SHOW) { - if (tDecodeI8(pDecoder, &pTask->showSink.reserved) < 0) return -1; } else { ASSERT(pTask->sinkType == TASK_SINK__NONE); } diff --git a/tests/script/tsim/db/create_all_options.sim b/tests/script/tsim/db/create_all_options.sim index ebb371688273427c9114c2eccca0c789eb359d8c..8a3826fe21bfefb5afc3b4d0f096855fbc1f1e48 100644 --- a/tests/script/tsim/db/create_all_options.sim +++ b/tests/script/tsim/db/create_all_options.sim @@ -346,8 +346,8 @@ sql drop database db sql_error create database db PRECISION 'as' sql_error create database db PRECISION -1 -print ====> QUORUM value [1 | 2, default: 1] -#sql create database db QUORUM 2 +print ====> QUORUM value [1 | 2, default: 1] 3.0 not support this item +#sql_error create database db QUORUM 2 #sql show databases #print $data0_db $data1_db $data2_db $data3_db $data4_db $data5_db $data6_db $data7_db $data8_db $data9_db $data10_db $data11_db $data12_db $data13_db $data14_db $data15_db $data16_db $data17_db #if $data5_db != 2 then @@ -362,9 +362,11 @@ print ====> QUORUM value [1 | 2, default: 1] # return -1 #endi #sql drop database db -#sql_error create database db QUORUM 3 -#sql_error create database db QUORUM 0 -#sql_error create database db QUORUM -1 +sql_error create database db QUORUM 1 +sql_error create database db QUORUM 2 +sql_error create database db QUORUM 3 +sql_error create database db QUORUM 0 +sql_error create database db QUORUM -1 print ====> REPLICA value [1 | 3, default: 1] sql create database db REPLICA 3 diff --git a/tests/system-test/0-others/taosShell.py b/tests/system-test/0-others/taosShell.py index 51b00bd66a4876b6e89122edffda77d0d46d9d3f..4e9a1155002fa0decd1cb6fc114f8183ca766c94 100644 --- a/tests/system-test/0-others/taosShell.py +++ b/tests/system-test/0-others/taosShell.py @@ -57,12 +57,12 @@ def taos_command (buildPath, key, value, expectString, cfgDir, sqlString='', key else: return "TAOS_FAIL" else: - if key == 'A' or key1 == 'A' or key == 'C' or key1 == 'C': + if key == 'A' or key1 == 'A' or key == 'C' or key1 == 'C' or key == 'V' or key1 == 'V': return "TAOS_OK", retResult else: return "TAOS_OK" else: - if key == 'A' or key1 == 'A' or key == 'C' or key1 == 'C': + if key == 'A' or key1 == 'A' or key == 'C' or key1 == 'C' or key == 'V' or key1 == 'V': return "TAOS_OK", retResult else: return "TAOS_FAIL" @@ -311,7 +311,7 @@ class TDTestCase: tdSql.query('drop database %s'%newDbName) tdLog.printNoPrefix("================================ parameter: -C") - newDbName="dbcc" + #newDbName="dbcc" retCode, retVal = taos_command(buildPath, "C", keyDict['C'], "buildinfo", keyDict['c'], '', '', '') if retCode != "TAOS_OK": tdLog.exit("taos -C fail") @@ -336,6 +336,86 @@ class TDTestCase: if (totalCfgItem["numOfCores"][2] != count) and (totalCfgItem["numOfCores"][0] != 'default'): tdLog.exit("taos -C return numOfCores error!") + version = totalCfgItem["version"][2] + + tdLog.printNoPrefix("================================ parameter: -V") + #newDbName="dbvv" + retCode, retVal = taos_command(buildPath, "V", keyDict['V'], "", keyDict['c'], '', '', '') + if retCode != "TAOS_OK": + tdLog.exit("taos -V fail") + + version = 'version: ' + version + retVal = retVal.replace("\n", "") + retVal = retVal.replace("\r", "") + if retVal != version: + print ("return version: [%s]"%retVal) + print ("dict version: [%s]"%version) + tdLog.exit("taos -V version not match") + + tdLog.printNoPrefix("================================ parameter: -d") + newDbName="dbd" + sqlString = 'create database ' + newDbName + ';' + retCode = taos_command(buildPath, "d", keyDict['d'], "taos>", keyDict['c'], sqlString, '', '') + if retCode != "TAOS_OK": + tdLog.exit("taos -d %s fail"%(keyDict['d'])) + else: + tdSql.query("show databases") + for i in range(tdSql.queryRows): + if tdSql.getData(i, 0) == newDbName: + break + else: + tdLog.exit("create db fail after taos -d %s fail"%(keyDict['d'])) + + tdSql.query('drop database %s'%newDbName) + + retCode = taos_command(buildPath, "d", 'dbno', "taos>", keyDict['c'], sqlString, '', '') + if retCode != "TAOS_FAIL": + tdLog.exit("taos -d dbno fail") + + tdLog.printNoPrefix("================================ parameter: -w") + newDbName="dbw" + keyDict['s'] = "\"create database " + newDbName + "\"" + retCode = taos_command(buildPath, "s", keyDict['s'], "Query OK", keyDict['c'], '', '', '') + if retCode != "TAOS_OK": + tdLog.exit("taos -w fail") + + keyDict['s'] = "\"create table " + newDbName + ".ntb (ts timestamp, c binary(128))\"" + retCode = taos_command(buildPath, "s", keyDict['s'], "Query OK", keyDict['c'], '', '', '') + if retCode != "TAOS_OK": + tdLog.exit("taos -w create table fail") + + keyDict['s'] = "\"insert into " + newDbName + ".ntb values('2021-04-01 08:00:00.001', 'abcd0123456789')('2021-04-01 08:00:00.002', 'abcd012345678901234567890123456789') \"" + retCode = taos_command(buildPath, "s", keyDict['s'], "Query OK", keyDict['c'], '', '', '') + if retCode != "TAOS_OK": + tdLog.exit("taos -w insert data fail") + + keyDict['s'] = "\"insert into " + newDbName + ".ntb values('2021-04-01 08:00:00.003', 'aaaaaaaaaaaaaaaaaaaa')('2021-04-01 08:00:01.004', 'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb') \"" + retCode = taos_command(buildPath, "s", keyDict['s'], "Query OK", keyDict['c'], '', '', '') + if retCode != "TAOS_OK": + tdLog.exit("taos -w insert data fail") + + keyDict['s'] = "\"insert into " + newDbName + ".ntb values('2021-04-01 08:00:00.005', 'cccccccccccccccccccc')('2021-04-01 08:00:01.006', 'dddddddddddddddddddddddddddddddddddddddd') \"" + retCode = taos_command(buildPath, "s", keyDict['s'], "Query OK", keyDict['c'], '', '', '') + if retCode != "TAOS_OK": + tdLog.exit("taos -w insert data fail") + + keyDict['s'] = "\"select * from " + newDbName + ".ntb \"" + retCode = taos_command(buildPath, "s", keyDict['s'], "aaaaaaaaaaaaaaaaaaaa", keyDict['c'], '', '', '') + if retCode != "TAOS_OK": + tdLog.exit("taos -w insert data fail") + + keyDict['s'] = "\"select * from " + newDbName + ".ntb \"" + retCode = taos_command(buildPath, "s", keyDict['s'], "dddddddddddddddddddddddddddddddddddddddd", keyDict['c'], '', '', '') + if retCode != "TAOS_FAIL": + tdLog.exit("taos -w insert data fail") + + keyDict['s'] = "\"select * from " + newDbName + ".ntb \"" + retCode = taos_command(buildPath, "s", keyDict['s'], "dddddddddddddddddddddddddddddddddddddddd", keyDict['c'], '', 'w', '60') + if retCode != "TAOS_OK": + tdLog.exit("taos -w insert data fail") + + tdSql.query('drop database %s'%newDbName) + def stop(self): tdSql.close() tdLog.success(f"{__file__} successfully executed")