diff --git a/source/dnode/mnode/impl/inc/mndDef.h b/source/dnode/mnode/impl/inc/mndDef.h index 345418801b27df5422b67ce3ebcf00344c2c7e4b..fe2c58e29e1251e084d97f95e39efdf511f67e66 100644 --- a/source/dnode/mnode/impl/inc/mndDef.h +++ b/source/dnode/mnode/impl/inc/mndDef.h @@ -108,7 +108,8 @@ typedef enum { TRN_STAGE_UNDO_ACTION = 3, TRN_STAGE_COMMIT = 4, TRN_STAGE_COMMIT_ACTION = 5, - TRN_STAGE_FINISHED = 6 + TRN_STAGE_FINISHED = 6, + TRN_STAGE_PRE_FINISH = 7 } ETrnStage; typedef enum { diff --git a/source/dnode/mnode/impl/src/mndTrans.c b/source/dnode/mnode/impl/src/mndTrans.c index 05368d344520c81ef00081273d40bcda2deb4f8f..56fce0507f1d461a77586d309be1bdc83c494992 100644 --- a/source/dnode/mnode/impl/src/mndTrans.c +++ b/source/dnode/mnode/impl/src/mndTrans.c @@ -460,6 +460,8 @@ static const char *mndTransStr(ETrnStage stage) { return "commitAction"; case TRN_STAGE_FINISHED: return "finished"; + case TRN_STAGE_PRE_FINISH: + return "pre-finish"; default: return "invalid"; } @@ -600,10 +602,15 @@ static int32_t mndTransActionUpdate(SSdb *pSdb, STrans *pOld, STrans *pNew) { } if (pOld->stage == TRN_STAGE_ROLLBACK) { - pOld->stage = TRN_STAGE_REDO_ACTION; + pOld->stage = TRN_STAGE_UNDO_ACTION; mTrace("trans:%d, stage from rollback to undoAction since perform update action", pNew->id); } + if (pOld->stage == TRN_STAGE_PRE_FINISH) { + pOld->stage = TRN_STAGE_FINISHED; + mTrace("trans:%d, stage from pre-finish to finished since perform update action", pNew->id); + } + return 0; } @@ -933,6 +940,16 @@ static int32_t mndTransRollback(SMnode *pMnode, STrans *pTrans) { return 0; } +static int32_t mndTransPreFinish(SMnode *pMnode, STrans *pTrans) { + mInfo("trans:%d, pre-finish transaction", pTrans->id); + if (mndTransSync(pMnode, pTrans) != 0) { + mError("trans:%d, failed to pre-finish since %s", pTrans->id, terrstr()); + return -1; + } + mInfo("trans:%d, pre-finish finished", pTrans->id); + return 0; +} + static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) { bool sendRsp = false; int32_t code = pTrans->code; @@ -1439,7 +1456,7 @@ static bool mndTransPerformCommitActionStage(SMnode *pMnode, STrans *pTrans) { if (code == 0) { pTrans->code = 0; - pTrans->stage = TRN_STAGE_FINISHED; + pTrans->stage = TRN_STAGE_FINISHED; // TRN_STAGE_PRE_FINISH is not necessary mInfo("trans:%d, stage from commitAction to finished", pTrans->id); continueExec = true; } else { @@ -1457,8 +1474,8 @@ static bool mndTransPerformUndoActionStage(SMnode *pMnode, STrans *pTrans) { int32_t code = mndTransExecuteUndoActions(pMnode, pTrans); if (code == 0) { - pTrans->stage = TRN_STAGE_FINISHED; - mInfo("trans:%d, stage from undoAction to finished", pTrans->id); + pTrans->stage = TRN_STAGE_PRE_FINISH; + mInfo("trans:%d, stage from undoAction to pre-finish", pTrans->id); continueExec = true; } else if (code == TSDB_CODE_ACTION_IN_PROGRESS) { mInfo("trans:%d, stage keep on undoAction since %s", pTrans->id, tstrerror(code)); @@ -1491,6 +1508,25 @@ static bool mndTransPerformRollbackStage(SMnode *pMnode, STrans *pTrans) { return continueExec; } +static bool mndTransPerfromPreFinishedStage(SMnode *pMnode, STrans *pTrans) { + if (mndCannotExecuteTransAction(pMnode)) return false; + + bool continueExec = true; + int32_t code = mndTransPreFinish(pMnode, pTrans); + + if (code == 0) { + pTrans->stage = TRN_STAGE_FINISHED; + mInfo("trans:%d, stage from pre-finish to finish", pTrans->id); + continueExec = true; + } else { + pTrans->failedTimes++; + mError("trans:%d, stage keep on pre-finish since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes); + continueExec = false; + } + + return continueExec; +} + static bool mndTransPerfromFinishedStage(SMnode *pMnode, STrans *pTrans) { bool continueExec = false; @@ -1547,6 +1583,14 @@ void mndTransExecute(SMnode *pMnode, STrans *pTrans, bool isLeader) { case TRN_STAGE_UNDO_ACTION: continueExec = mndTransPerformUndoActionStage(pMnode, pTrans); break; + case TRN_STAGE_PRE_FINISH: + if (isLeader) { + continueExec = mndTransPerfromPreFinishedStage(pMnode, pTrans); + } else { + mInfo("trans:%d, can not pre-finish since not leader", pTrans->id); + continueExec = false; + } + break; case TRN_STAGE_FINISHED: continueExec = mndTransPerfromFinishedStage(pMnode, pTrans); break; diff --git a/source/dnode/mnode/sdb/src/sdbHash.c b/source/dnode/mnode/sdb/src/sdbHash.c index cc3dba2e07d64f68c017d1b6377fc81f99cad435..569c78a68c0c560e0272521fd2512089f63ca593 100644 --- a/source/dnode/mnode/sdb/src/sdbHash.c +++ b/source/dnode/mnode/sdb/src/sdbHash.c @@ -160,6 +160,7 @@ static int32_t sdbInsertRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow * if (insertFp != NULL) { code = (*insertFp)(pSdb, pRow->pObj); if (code != 0) { + if (terrno == 0) terrno = TSDB_CODE_MND_TRANS_UNKNOW_ERROR; code = terrno; taosHashRemove(hash, pRow->pObj, keySize); sdbFreeRow(pSdb, pRow, false); diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c index 5712a8ecb8d020628eff44780469f3e12f6939c8..2b275f5122bef650812fd3041acc03d4714f1cbe 100644 --- a/source/dnode/vnode/src/vnd/vnodeSvr.c +++ b/source/dnode/vnode/src/vnd/vnodeSvr.c @@ -1211,6 +1211,44 @@ static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq tDecoderClear(&dc); } + // scan + TSKEY now = taosGetTimestamp(pVnode->config.tsdbCfg.precision); + TSKEY minKey = now - tsTickPerMin[pVnode->config.tsdbCfg.precision] * pVnode->config.tsdbCfg.keep2; + TSKEY maxKey = tsMaxKeyByPrecision[pVnode->config.tsdbCfg.precision]; + for (int32_t i = 0; i < TARRAY_SIZE(pSubmitReq->aSubmitTbData); ++i) { + SSubmitTbData *pSubmitTbData = taosArrayGet(pSubmitReq->aSubmitTbData, i); + + if (pSubmitTbData->flags & SUBMIT_REQ_COLUMN_DATA_FORMAT) { + if (TARRAY_SIZE(pSubmitTbData->aCol) <= 0) { + code = TSDB_CODE_INVALID_MSG; + goto _exit; + } + + SColData *pColData = (SColData *)taosArrayGet(pSubmitTbData->aCol, 0); + TSKEY *aKey = (TSKEY *)(pColData->pData); + + for (int32_t iRow = 0; iRow < pColData->nVal; iRow++) { + if (aKey[iRow] < minKey || aKey[iRow] > maxKey || (iRow > 0 && aKey[iRow] <= aKey[iRow - 1])) { + code = TSDB_CODE_INVALID_MSG; + vError("vgId:%d %s failed since %s, version:%" PRId64, TD_VID(pVnode), __func__, tstrerror(terrno), version); + goto _exit; + } + } + + } else { + int32_t nRow = TARRAY_SIZE(pSubmitTbData->aRowP); + SRow **aRow = (SRow **)TARRAY_DATA(pSubmitTbData->aRowP); + + for (int32_t iRow = 0; iRow < nRow; ++iRow) { + if (aRow[iRow]->ts < minKey || aRow[iRow]->ts > maxKey || (iRow > 0 && aRow[iRow]->ts <= aRow[iRow - 1]->ts)) { + code = TSDB_CODE_INVALID_MSG; + vError("vgId:%d %s failed since %s, version:%" PRId64, TD_VID(pVnode), __func__, tstrerror(terrno), version); + goto _exit; + } + } + } + } + for (int32_t i = 0; i < TARRAY_SIZE(pSubmitReq->aSubmitTbData); ++i) { SSubmitTbData *pSubmitTbData = taosArrayGet(pSubmitReq->aSubmitTbData, i); diff --git a/source/libs/executor/src/eventwindowoperator.c b/source/libs/executor/src/eventwindowoperator.c index 49e2d5bc4ac0c85fabd8ffde93aa527bbe3c85e0..559fce52249facc32213789eee9bf2655b36ee16 100644 --- a/source/libs/executor/src/eventwindowoperator.c +++ b/source/libs/executor/src/eventwindowoperator.c @@ -29,7 +29,7 @@ typedef struct SEventWindowOperatorInfo { SWindowRowsSup winSup; int32_t tsSlotId; // primary timestamp column slot id STimeWindowAggSupp twAggSup; - uint64_t groupId; // current group id, used to identify the data block from different groups + uint64_t groupId; // current group id, used to identify the data block from different groups SFilterInfo* pStartCondInfo; SFilterInfo* pEndCondInfo; bool inWindow; @@ -310,6 +310,7 @@ int32_t eventWindowAggImpl(SOperatorInfo* pOperator, SEventWindowOperatorInfo* p pSup->rowEntryInfoOffset, pTaskInfo); pRes->info.rows += pInfo->pRow->numOfRows; + pInfo->pRow->numOfRows = 0; pInfo->inWindow = false; rowIndex += 1; diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c index 69d48a99d755d3e6a021c2755a2b94eb9d8502be..cdc5dad8ecce3ee5aa924d447a13fe502f3d5f66 100644 --- a/source/libs/parser/src/parTranslater.c +++ b/source/libs/parser/src/parTranslater.c @@ -2797,6 +2797,10 @@ static int32_t translateOrderBy(STranslateContext* pCxt, SSelectStmt* pSelect) { bool other; int32_t code = translateOrderByPosition(pCxt, pSelect->pProjectionList, pSelect->pOrderByList, &other); if (TSDB_CODE_SUCCESS == code) { + if (0 == LIST_LENGTH(pSelect->pOrderByList)) { + NODES_DESTORY_LIST(pSelect->pOrderByList); + return TSDB_CODE_SUCCESS; + } if (!other) { return TSDB_CODE_SUCCESS; } diff --git a/source/libs/wal/src/walWrite.c b/source/libs/wal/src/walWrite.c index a992c951fab0a5b9d2c0889d3fe4eb8588374ab3..b74bbdd15f25b1781f1b2a361fa21a9fee4e8f0a 100644 --- a/source/libs/wal/src/walWrite.c +++ b/source/libs/wal/src/walWrite.c @@ -387,20 +387,33 @@ END: int32_t walRollImpl(SWal *pWal) { int32_t code = 0; + if (pWal->pIdxFile != NULL) { + code = taosFsyncFile(pWal->pIdxFile); + if (code != 0) { + terrno = TAOS_SYSTEM_ERROR(errno); + goto END; + } code = taosCloseFile(&pWal->pIdxFile); if (code != 0) { terrno = TAOS_SYSTEM_ERROR(errno); goto END; } } + if (pWal->pLogFile != NULL) { + code = taosFsyncFile(pWal->pLogFile); + if (code != 0) { + terrno = TAOS_SYSTEM_ERROR(errno); + goto END; + } code = taosCloseFile(&pWal->pLogFile); if (code != 0) { terrno = TAOS_SYSTEM_ERROR(errno); goto END; } } + TdFilePtr pIdxFile, pLogFile; // create new file int64_t newFileFirstVer = pWal->vers.lastVer + 1; diff --git a/tests/script/tsim/trans/create_stb.sim b/tests/script/tsim/trans/create_stb.sim new file mode 100644 index 0000000000000000000000000000000000000000..94041645d570296480b2557a3ec9bb21aca7241f --- /dev/null +++ b/tests/script/tsim/trans/create_stb.sim @@ -0,0 +1,72 @@ +system sh/stop_dnodes.sh +system sh/deploy.sh -n dnode1 -i 1 +system sh/deploy.sh -n dnode2 -i 2 +system sh/deploy.sh -n dnode3 -i 3 +system sh/deploy.sh -n dnode4 -i 4 +system sh/cfg.sh -n dnode1 -c transPullupInterval -v 1 +system sh/cfg.sh -n dnode2 -c transPullupInterval -v 1 +system sh/cfg.sh -n dnode3 -c transPullupInterval -v 1 +system sh/cfg.sh -n dnode4 -c transPullupInterval -v 1 +system sh/exec.sh -n dnode1 -s start +system sh/exec.sh -n dnode2 -s start +system sh/exec.sh -n dnode3 -s start +system sh/exec.sh -n dnode4 -s start +sql connect + +print =============== step1: create dnodes +sql create dnode $hostname port 7200 +sql create dnode $hostname port 7300 +sql create dnode $hostname port 7400 + +$x = 0 +step1: + $x = $x + 1 + sleep 1000 + if $x == 10 then + return -1 + endi +sql select * from information_schema.ins_dnodes -x step1 +if $data(1)[4] != ready then + goto step1 +endi +if $data(2)[4] != ready then + goto step1 +endi +if $data(3)[4] != ready then + goto step1 +endi + +print =============== step2: create mnode 2 and 3 +sql create mnode on dnode 2 +sql create mnode on dnode 3 +sql create database db vgroups 2 + +print =============== step3: kill dnode4 +system sh/exec.sh -n dnode4 -s stop -x SIGKILL +sql use db +sql_error create table stb (ts timestamp, i int) tags (j int) + +print =============== step4: create database +sql show transactions +if $rows != 1 then + return -1 +endi + +sleep 3000 +system sh/exec.sh -n dnode4 -s start + +$wt = 0 +step4: +$wt = $wt + 1 +sleep 1000 +if $wt == 200 then + print ====> transaction already running + return -1 +endi + +sql show transactions +if $rows != 0 then + print wait 1 seconds to alter + goto step4 +endi + diff --git a/tests/system-test/0-others/show.py b/tests/system-test/0-others/show.py index 82f9cb9d7ef8488027059873b7220c2ecc39ba62..3e176fe25162b2053f5d86d83b97bc61e6a354d8 100644 --- a/tests/system-test/0-others/show.py +++ b/tests/system-test/0-others/show.py @@ -11,10 +11,6 @@ # -*- coding: utf-8 -*- - - - - import re from util.log import * from util.cases import * @@ -31,10 +27,35 @@ class TDTestCase: self.ins_param_list = ['dnodes','mnodes','qnodes','cluster','functions','users','grants','topics','subscriptions','streams'] self.perf_param = ['apps','connections','consumers','queries','transactions'] self.perf_param_list = ['apps','connections','consumers','queries','trans'] - + self.dbname = "db" + self.vgroups = 10 + self.stbname = f'`{tdCom.getLongName(5)}`' + self.tbname = f'`{tdCom.getLongName(3)}`' + self.db_param = { + "database":f"{self.dbname}", + "buffer":100, + "cachemodel":"'none'", + "cachesize":1, + "comp":2, + "maxrows":1000, + "minrows":200, + "pages":512, + "pagesize":16, + "precision":"'ms'", + "replica":1, + "wal_level":1, + "wal_fsync_period":6000, + "wal_roll_period":0, + "wal_segment_size":1024, + "vgroups":self.vgroups, + "stt_trigger":1, + "tsdb_pagesize":16 + } def ins_check(self): tdSql.prepare() for param in self.ins_param_list: + if param.lower() == 'qnodes': + tdSql.execute('create qnode on dnode 1') tdSql.query(f'show {param}') show_result = tdSql.queryResult tdSql.query(f'select * from information_schema.ins_{param}') @@ -62,11 +83,32 @@ class TDTestCase: tag_sql += f"{k} {v}, " create_stb_sql = f'create stable {stbname} ({column_sql[:-2]}) tags ({tag_sql[:-2]})' return create_stb_sql - def show_sql(self): - tdSql.prepare() - tdSql.execute('use db') - stbname = f'`{tdCom.getLongName(5)}`' - tbname = f'`{tdCom.getLongName(3)}`' + + def set_create_database_sql(self,sql_dict): + create_sql = 'create' + for key,value in sql_dict.items(): + create_sql += f' {key} {value}' + return create_sql + def show_create_sql(self): + create_db_sql = self.set_create_database_sql(self.db_param) + print(create_db_sql) + tdSql.execute(create_db_sql) + tdSql.query(f'show create database {self.dbname}') + tdSql.checkEqual(self.dbname,tdSql.queryResult[0][0]) + for key,value in self.db_param.items(): + if key == 'database': + continue + else: + param = f'{key} {value}' + if param in tdSql.queryResult[0][1].lower(): + tdLog.info(f'show create database check success with {key} {value}') + continue + else: + tdLog.exit(f"show create database check failed with {key} {value}") + tdSql.query('show vnodes 1') + tdSql.checkRows(self.vgroups) + tdSql.execute(f'use {self.dbname}') + column_dict = { '`ts`': 'timestamp', '`col1`': 'tinyint', @@ -101,21 +143,21 @@ class TDTestCase: '`t14`': 'timestamp' } - create_table_sql = self.set_stb_sql(stbname,column_dict,tag_dict) + create_table_sql = self.set_stb_sql(self.stbname,column_dict,tag_dict) tdSql.execute(create_table_sql) - tdSql.query(f'show create table {stbname}') + tdSql.query(f'show create stable {self.stbname}') query_result = tdSql.queryResult tdSql.checkEqual(query_result[0][1].lower(),create_table_sql) - tdSql.execute(f'create table {tbname} using {stbname} tags(1,1,1,1,1,1,1,1,1.000000e+00,1.000000e+00,true,"abc","abc123",0)') + tdSql.execute(f'create table {self.tbname} using {self.stbname} tags(1,1,1,1,1,1,1,1,1.000000e+00,1.000000e+00,true,"abc","abc123",0)') tag_sql = '(' for tag_keys in tag_dict.keys(): tag_sql += f'{tag_keys}, ' tags = f'{tag_sql[:-2]})' - sql = f'create table {tbname} using {stbname} {tags} tags (1, 1, 1, 1, 1, 1, 1, 1, 1.000000e+00, 1.000000e+00, true, "abc", "abc123", 0)' - tdSql.query(f'show create table {tbname}') + sql = f'create table {self.tbname} using {self.stbname} {tags} tags (1, 1, 1, 1, 1, 1, 1, 1, 1.000000e+00, 1.000000e+00, true, "abc", "abc123", 0)' + tdSql.query(f'show create table {self.tbname}') query_result = tdSql.queryResult tdSql.checkEqual(query_result[0][1].lower(),sql) - tdSql.execute('drop database db') + tdSql.execute(f'drop database {self.dbname}') def check_gitinfo(self): taosd_gitinfo_sql = '' tdSql.query('show dnode 1 variables') @@ -133,11 +175,24 @@ class TDTestCase: taosd_info = os.popen('taosd -V').read() taosd_gitinfo = re.findall("^gitinfo.*",taosd_info,re.M) tdSql.checkEqual(taosd_gitinfo_sql,taosd_gitinfo[0]) + + def show_base(self): + for sql in ['dnodes','mnodes','cluster']: + tdSql.query(f'show {sql}') + print(tdSql.queryResult) + tdSql.checkRows(1) + tdSql.query('show grants') + grants_info = tdSql.queryResult + tdSql.query('show licences') + licences_info = tdSql.queryResult + tdSql.checkEqual(grants_info,licences_info) + def run(self): self.check_gitinfo() + self.show_base() self.ins_check() self.perf_check() - self.show_sql() + self.show_create_sql() def stop(self): tdSql.close() @@ -145,4 +200,3 @@ class TDTestCase: tdCases.addWindows(__file__, TDTestCase()) tdCases.addLinux(__file__, TDTestCase()) - diff --git a/tests/system-test/2-query/projectionDesc.py b/tests/system-test/2-query/projectionDesc.py index 6bb9aff4f3c8707a9fda16f4e240961ee91eaccf..1edceaa6f2ced20595132149c8d8f135111ce097 100644 --- a/tests/system-test/2-query/projectionDesc.py +++ b/tests/system-test/2-query/projectionDesc.py @@ -38,6 +38,7 @@ class TDTestCase: #tdSql.checkData(0,0,1537146000000) tdSql.checkData(0,1,10) + tdSql.query(f"select * from {dbname}.stb_1 order by 'aaa' desc") def stop(self): tdSql.close() diff --git a/tests/system-test/7-tmq/tmqConsFromTsdb1-1ctb.py b/tests/system-test/7-tmq/tmqConsFromTsdb1-1ctb.py index 4dcc0b963f8e883326132f5db3bc04a6edd349d1..c11159c6e551d60b90aab16bb7c90e4b26021ac2 100644 --- a/tests/system-test/7-tmq/tmqConsFromTsdb1-1ctb.py +++ b/tests/system-test/7-tmq/tmqConsFromTsdb1-1ctb.py @@ -200,7 +200,7 @@ class TDTestCase: tdLog.info("pkill consume processor") tdCom.killProcessor("tmq_sim") - # time.sleep(10) + time.sleep(10) # reinit consume info, and start tmq_sim, then check consume result tmqCom.initConsumerTable()