xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge branch 'fix/TD-22951' into feature/3_liaohj

This commit is contained in:
Shengliang Guan 2023-03-03 18:21:55 +08:00
parent 051e4796f3 2624ea65d8
commit 25e0ed8cb5
11 changed files with 254 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
source/dnode/mnode/impl/inc/mndDef.h
View File

@ -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 {

52
source/dnode/mnode/impl/src/mndTrans.c
View File

@ -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;

1
source/dnode/mnode/sdb/src/sdbHash.c
View File

@ -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);

38
source/dnode/vnode/src/vnd/vnodeSvr.c
View File

@ -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);

1
source/libs/executor/src/eventwindowoperator.c
View File

@ -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;

4
source/libs/parser/src/parTranslater.c
View File

@ -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;
}

13
source/libs/wal/src/walWrite.c
View File

@ -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;

72
tests/script/tsim/trans/create_stb.sim Normal file
View File

@ -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

90
tests/system-test/0-others/show.py
View File

@ -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())

1
tests/system-test/2-query/projectionDesc.py
View File

@ -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()

2
tests/system-test/7-tmq/tmqConsFromTsdb1-1ctb.py
View File

@ -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()