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

Merge branch '3.0' of https://github.com/taosdata/TDengine into 3.0

This commit is contained in:
chenhaoran 2024-01-02 10:25:34 +08:00
parent f90694e726 d5a9d0de46
commit 762f1eac1e
24 changed files with 551 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
source/client/src/clientTmq.c
View File

@ -1028,11 +1028,16 @@ static void tmqMgmtInit(void) {
}
}
#define SET_ERROR_MSG(MSG) if(errstr!=NULL)snprintf(errstr,errstrLen,MSG);
tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
if(conf == NULL) return NULL;
if(conf == NULL) {
SET_ERROR_MSG("configure is null")
return NULL;
}
taosThreadOnce(&tmqInit, tmqMgmtInit);
if (tmqInitRes != 0) {
terrno = tmqInitRes;
SET_ERROR_MSG("tmq timer init error")
return NULL;
}
@ -1040,6 +1045,7 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
if (pTmq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
tscError("failed to create consumer, groupId:%s, code:%s", conf->groupId, terrstr());
SET_ERROR_MSG("malloc tmq failed")
return NULL;
}
@ -1055,6 +1061,7 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
conf->groupId[0] == 0) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
tscError("consumer:0x%" PRIx64 " setup failed since %s, groupId:%s", pTmq->consumerId, terrstr(), pTmq->groupId);
SET_ERROR_MSG("malloc tmq element failed or group is empty")
goto _failed;
}
@ -1086,6 +1093,7 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
if (tsem_init(&pTmq->rspSem, 0, 0) != 0) {
tscError("consumer:0x %" PRIx64 " setup failed since %s, consumer group %s", pTmq->consumerId, terrstr(),
pTmq->groupId);
SET_ERROR_MSG("init t_sem failed")
goto _failed;
}
@ -1094,11 +1102,13 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
if (pTmq->pTscObj == NULL) {
tscError("consumer:0x%" PRIx64 " setup failed since %s, groupId:%s", pTmq->consumerId, terrstr(), pTmq->groupId);
tsem_destroy(&pTmq->rspSem);
SET_ERROR_MSG("init tscObj failed")
goto _failed;
}
pTmq->refId = taosAddRef(tmqMgmt.rsetId, pTmq);
if (pTmq->refId < 0) {
SET_ERROR_MSG("add tscObj ref failed")
goto _failed;
}

2
source/common/src/tglobal.c
View File

@ -789,7 +789,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
if (cfgAddInt32(pCfg, "s3PageCacheSize", tsS3PageCacheSize, 4, 1024 * 1024 * 1024, CFG_SCOPE_SERVER,
CFG_DYN_ENT_SERVER) != 0)
return -1;
if (cfgAddInt32(pCfg, "s3UploadDelaySec", tsS3UploadDelaySec, 60 * 10, 60 * 60 * 24 * 30, CFG_SCOPE_SERVER,
if (cfgAddInt32(pCfg, "s3UploadDelaySec", tsS3UploadDelaySec, 60 * 1, 60 * 60 * 24 * 30, CFG_SCOPE_SERVER,
CFG_DYN_ENT_SERVER) != 0)
return -1;

2
source/dnode/mnode/impl/src/mndStream.c
View File

@ -792,7 +792,7 @@ static int32_t checkForNumOfStreams(SMnode *pMnode, SStreamObj *pStreamObj) { /
if (pStream->targetStbUid == pStreamObj->targetStbUid) {
mError("Cannot write the same stable as other stream:%s", pStream->name);
sdbCancelFetch(pMnode->pSdb, pIter);
terrno = TSDB_CODE_MND_TOO_MANY_STREAMS;
terrno = TSDB_CODE_MND_INVALID_TARGET_TABLE;
return terrno;
}
}

6
source/dnode/vnode/src/tsdb/tsdbFS2.c
View File

@ -805,6 +805,12 @@ int64_t tsdbFSAllocEid(STFileSystem *fs) {
return cid;
}
void tsdbFSUpdateEid(STFileSystem *fs, int64_t cid) {
taosThreadMutexLock(&fs->tsdb->mutex);
fs->neid = TMAX(fs->neid, cid);
taosThreadMutexUnlock(&fs->tsdb->mutex);
}
int32_t tsdbFSEditBegin(STFileSystem *fs, const TFileOpArray *opArray, EFEditT etype) {
int32_t code = 0;
int32_t lino;

1
source/dnode/vnode/src/tsdb/tsdbFS2.h
View File

@ -52,6 +52,7 @@ int32_t tsdbFSCreateRefRangedSnapshot(STFileSystem *fs, int64_t sver, int64_t ev
int32_t tsdbFSDestroyRefRangedSnapshot(TFileSetRangeArray **fsrArr);
// txn
int64_t tsdbFSAllocEid(STFileSystem *fs);
void tsdbFSUpdateEid(STFileSystem *fs, int64_t cid);
int32_t tsdbFSEditBegin(STFileSystem *fs, const TFileOpArray *opArray, EFEditT etype);
int32_t tsdbFSEditCommit(STFileSystem *fs);
int32_t tsdbFSEditAbort(STFileSystem *fs);

6
source/dnode/vnode/src/tsdb/tsdbFSetRAW.c
View File

@ -14,6 +14,7 @@
*/
#include "tsdbFSetRAW.h"
#include "tsdbFS2.h"
// SFSetRAWWriter ==================================================
typedef struct SFSetRAWWriter {
@ -76,7 +77,7 @@ static int32_t tsdbFSetRAWWriteFileDataBegin(SFSetRAWWriter *writer, STsdbDataRA
.szPage = writer->config->szPage,
.fid = bHdr->file.fid,
.did = writer->config->did,
.cid = writer->config->cid,
.cid = bHdr->file.cid,
.level = writer->config->level,
.file =
@ -84,7 +85,7 @@ static int32_t tsdbFSetRAWWriteFileDataBegin(SFSetRAWWriter *writer, STsdbDataRA
.type = bHdr->file.type,
.fid = bHdr->file.fid,
.did = writer->config->did,
.cid = writer->config->cid,
.cid = bHdr->file.cid,
.size = bHdr->file.size,
.minVer = bHdr->file.minVer,
.maxVer = bHdr->file.maxVer,
@ -94,6 +95,7 @@ static int32_t tsdbFSetRAWWriteFileDataBegin(SFSetRAWWriter *writer, STsdbDataRA
},
};
tsdbFSUpdateEid(config.tsdb->pFS, config.cid);
writer->ctx->offset = 0;
writer->ctx->file = config.file;

28
source/libs/stream/src/streamBackendRocksdb.c
View File

@ -665,6 +665,7 @@ void streamBackendHandleCleanup(void* arg) {
return;
}
#ifdef BUILD_NO_CALL
int32_t getLatestCheckpoint(void* arg, int64_t* checkpoint) {
SStreamMeta* pMeta = arg;
taosWLockLatch(&pMeta->chkpDirLock);
@ -738,6 +739,7 @@ int32_t delObsoleteCheckpoint(void* arg, const char* path) {
taosArrayDestroy(chkpDel);
return 0;
}
#endif
/*
* checkpointSave |--cp1--|--cp2--|--cp3--|--cp4--|--cp5--|
* chkpInUse: |--cp2--|--cp4--|
@ -855,6 +857,7 @@ int32_t streamBackendLoadCheckpointInfo(void* arg) {
return 0;
}
#ifdef BUILD_NO_CALL
int32_t chkpGetAllDbCfHandle(SStreamMeta* pMeta, rocksdb_column_family_handle_t*** ppHandle, SArray* refs) {
return 0;
// SArray* pHandle = taosArrayInit(16, POINTER_BYTES);
@ -891,6 +894,7 @@ int32_t chkpGetAllDbCfHandle(SStreamMeta* pMeta, rocksdb_column_family_handle_t*
// *ppHandle = ppCf;
// return nCf;
}
#endif
int32_t chkpGetAllDbCfHandle2(STaskDbWrapper* pBackend, rocksdb_column_family_handle_t*** ppHandle) {
SArray* pHandle = taosArrayInit(8, POINTER_BYTES);
@ -1006,6 +1010,7 @@ int32_t taskDbBuildSnap(void* arg, SArray* pSnap) {
return code;
}
#ifdef BUILD_NO_CALL
int32_t streamBackendAddInUseChkp(void* arg, int64_t chkpId) {
// if (arg == NULL) return 0;
@ -1029,6 +1034,7 @@ int32_t streamBackendDelInUseChkp(void* arg, int64_t chkpId) {
// }
// taosWUnLockLatch(&pMeta->chkpDirLock);
}
#endif
/*
0
@ -1099,7 +1105,9 @@ void streamBackendDelCompare(void* backend, void* arg) {
taosMemoryFree(node);
}
}
#ifdef BUILD_NO_CALL
void streamStateDestroy_rocksdb(SStreamState* pState, bool remove) { streamStateCloseBackend(pState, remove); }
#endif
void destroyRocksdbCfInst(RocksdbCfInst* inst) {
int cfLen = sizeof(ginitDict) / sizeof(ginitDict[0]);
if (inst->pHandle) {
@ -2170,6 +2178,7 @@ int32_t streamStateOpenBackendCf(void* backend, char* name, char** cfs, int32_t
taosMemoryFree(cfOpts);
return 0;
}
#ifdef BUILD_NO_CALL
int streamStateOpenBackend(void* backend, SStreamState* pState) {
taosAcquireRef(streamBackendId, pState->streamBackendRid);
SBackendWrapper* handle = backend;
@ -2279,6 +2288,7 @@ void streamStateCloseBackend(SStreamState* pState, bool remove) {
wrapper->remove |= remove; // update by other pState
taosReleaseRef(streamBackendCfWrapperId, pState->pTdbState->backendCfWrapperId);
}
#endif
void streamStateDestroyCompar(void* arg) {
SCfComparator* comp = (SCfComparator*)arg;
for (int i = 0; i < comp->numOfComp; i++) {
@ -2669,7 +2679,7 @@ SStreamStateCur* streamStateSeekToLast_rocksdb(SStreamState* pState) {
STREAM_STATE_DEL_ROCKSDB(pState, "state", &maxStateKey);
return pCur;
}
#ifdef BUILD_NO_CALL
SStreamStateCur* streamStateGetCur_rocksdb(SStreamState* pState, const SWinKey* key) {
qDebug("streamStateGetCur_rocksdb");
STaskDbWrapper* wrapper = pState->pTdbState->pOwner->pBackend;
@ -2719,6 +2729,7 @@ int32_t streamStateFuncDel_rocksdb(SStreamState* pState, const STupleKey* key) {
STREAM_STATE_DEL_ROCKSDB(pState, "func", key);
return 0;
}
#endif
// session cf
int32_t streamStateSessionPut_rocksdb(SStreamState* pState, const SSessionKey* key, const void* value, int32_t vLen) {
@ -3128,6 +3139,7 @@ SStreamStateCur* streamStateFillSeekKeyPrev_rocksdb(SStreamState* pState, const
streamStateFreeCur(pCur);
return NULL;
}
#ifdef BUILD_NO_CALL
int32_t streamStateSessionGetKeyByRange_rocksdb(SStreamState* pState, const SSessionKey* key, SSessionKey* curKey) {
stDebug("streamStateSessionGetKeyByRange_rocksdb");
STaskDbWrapper* wrapper = pState->pTdbState->pOwner->pBackend;
@ -3185,6 +3197,7 @@ int32_t streamStateSessionGetKeyByRange_rocksdb(SStreamState* pState, const SSes
streamStateFreeCur(pCur);
return -1;
}
#endif
int32_t streamStateSessionAddIfNotExist_rocksdb(SStreamState* pState, SSessionKey* key, TSKEY gap, void** pVal,
int32_t* pVLen) {
@ -3317,6 +3330,7 @@ _end:
return res;
}
#ifdef BUILD_NO_CALL
// partag cf
int32_t streamStatePutParTag_rocksdb(SStreamState* pState, int64_t groupId, const void* tag, int32_t tagLen) {
int code = 0;
@ -3329,6 +3343,7 @@ int32_t streamStateGetParTag_rocksdb(SStreamState* pState, int64_t groupId, void
STREAM_STATE_GET_ROCKSDB(pState, "partag", &groupId, tagVal, tagLen);
return code;
}
#endif
// parname cfg
int32_t streamStatePutParName_rocksdb(SStreamState* pState, int64_t groupId, const char tbname[TSDB_TABLE_NAME_LEN]) {
int code = 0;
@ -3342,11 +3357,13 @@ int32_t streamStateGetParName_rocksdb(SStreamState* pState, int64_t groupId, voi
return code;
}
#ifdef BUILD_NO_CALL
int32_t streamDefaultPut_rocksdb(SStreamState* pState, const void* key, void* pVal, int32_t pVLen) {
int code = 0;
STREAM_STATE_PUT_ROCKSDB(pState, "default", key, pVal, pVLen);
return code;
}
#endif
int32_t streamDefaultGet_rocksdb(SStreamState* pState, const void* key, void** pVal, int32_t* pVLen) {
int code = 0;
STREAM_STATE_GET_ROCKSDB(pState, "default", key, pVal, pVLen);
@ -3400,6 +3417,7 @@ int32_t streamDefaultIterGet_rocksdb(SStreamState* pState, const void* start, co
rocksdb_iter_destroy(pIter);
return code;
}
#ifdef BUILD_NO_CALL
void* streamDefaultIterCreate_rocksdb(SStreamState* pState) {
SStreamStateCur* pCur = createStreamStateCursor();
STaskDbWrapper* wrapper = pState->pTdbState->pOwner->pBackend;
@ -3443,6 +3461,7 @@ char* streamDefaultIterVal_rocksdb(void* iter, int32_t* len) {
return ret;
}
#endif
// batch func
void* streamStateCreateBatch() {
rocksdb_writebatch_t* pBatch = rocksdb_writebatch_create();
@ -3796,11 +3815,12 @@ void dbChkpDestroy(SDbChkp* pChkp) {
taosMemoryFree(pChkp->pManifest);
taosMemoryFree(pChkp);
}
#ifdef BUILD_NO_CALL
int32_t dbChkpInit(SDbChkp* p) {
if (p == NULL) return 0;
return 0;
}
#endif
int32_t dbChkpDumpTo(SDbChkp* p, char* dname, SArray* list) {
taosThreadRwlockRdlock(&p->rwLock);
int32_t code = -1;
@ -3945,6 +3965,7 @@ int32_t bkdMgtGetDelta(SBkdMgt* bm, char* taskId, int64_t chkpId, SArray* list,
return code;
}
#ifdef BUILD_NO_CALL
int32_t bkdMgtAddChkp(SBkdMgt* bm, char* task, char* path) {
int32_t code = -1;
@ -3974,4 +3995,5 @@ int32_t bkdMgtDumpTo(SBkdMgt* bm, char* taskId, char* dname) {
taosThreadRwlockUnlock(&bm->rwLock);
return code;
}
}
#endif

2
source/libs/stream/src/streamMeta.c
View File

@ -419,6 +419,7 @@ _err:
}
// todo refactor: the lock shoud be restricted in one function
#ifdef BUILD_NO_CALL
void streamMetaInitBackend(SStreamMeta* pMeta) {
pMeta->streamBackend = streamBackendInit(pMeta->path, pMeta->chkpId, pMeta->vgId);
if (pMeta->streamBackend == NULL) {
@ -440,6 +441,7 @@ void streamMetaInitBackend(SStreamMeta* pMeta) {
pMeta->streamBackendRid = taosAddRef(streamBackendId, pMeta->streamBackend);
streamBackendLoadCheckpointInfo(pMeta);
}
#endif
void streamMetaClear(SStreamMeta* pMeta) {
// remove all existed tasks in this vnode

60
tests/army/community/cluster/snapshot.json Normal file
View File

@ -0,0 +1,60 @@
{
"filetype": "insert",
"cfgdir": "/etc/taos",
"host": "127.0.0.1",
"port": 6030,
"user": "root",
"password": "taosdata",
"connection_pool_size": 8,
"num_of_records_per_req": 2000,
"thread_count": 2,
"create_table_thread_count": 1,
"confirm_parameter_prompt": "no",
"databases": [
{
"dbinfo": {
"name": "db",
"drop": "yes",
"vgroups": 2,
"replica": 3,
"duration":"1d",
"wal_retention_period": 1,
"wal_retention_size": 1,
"keep": "3d,6d,30d"
},
"super_tables": [
{
"name": "stb",
"child_table_exists": "no",
"childtable_count": 10,
"insert_rows": 100000,
"childtable_prefix": "d",
"insert_mode": "taosc",
"timestamp_step": 10000,
"start_timestamp":"now-12d",
"columns": [
{ "type": "bool", "name": "bc"},
{ "type": "float", "name": "fc" },
{ "type": "double", "name": "dc"},
{ "type": "tinyint", "name": "ti"},
{ "type": "smallint", "name": "si" },
{ "type": "int", "name": "ic" },
{ "type": "bigint", "name": "bi" },
{ "type": "utinyint", "name": "uti"},
{ "type": "usmallint", "name": "usi"},
{ "type": "uint", "name": "ui" },
{ "type": "ubigint", "name": "ubi"},
{ "type": "binary", "name": "bin", "len": 16},
{ "type": "nchar", "name": "nch", "len": 32}
],
"tags": [
{"type": "tinyint", "name": "groupid","max": 10,"min": 1},
{"name": "location","type": "binary", "len": 16, "values":
["San Francisco", "Los Angles", "San Diego", "San Jose", "Palo Alto", "Campbell", "Mountain View","Sunnyvale", "Santa Clara", "Cupertino"]
}
]
}
]
}
]
}

96
tests/army/community/cluster/snapshot.py Normal file
View File

@ -0,0 +1,96 @@
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import sys
import time
import random
import taos
import frame
import frame.etool
from frame.log import *
from frame.cases import *
from frame.sql import *
from frame.caseBase import *
from frame import *
class TDTestCase(TBase):
def insertData(self):
tdLog.info(f"insert data.")
# taosBenchmark run
jfile = etool.curFile(__file__, "snapshot.json")
etool.runBenchmark(json=jfile)
tdSql.execute(f"use {self.db}")
# set insert data information
self.childtable_count = 10
self.insert_rows = 100000
self.timestamp_step = 10000
def doAction(self):
tdLog.info(f"do action.")
self.flushDb()
# split vgroups
self.splitVGroups()
self.trimDb()
self.checkAggCorrect()
# balance vgroups
self.balanceVGroupLeader()
# replica to 1
self.alterReplica(1)
self.checkAggCorrect()
self.compactDb()
self.alterReplica(3)
vgids = self.getVGroup(self.db)
selid = random.choice(vgids)
self.balanceVGroupLeaderOn(selid)
# run
def run(self):
tdLog.debug(f"start to excute {__file__}")
# insert data
self.insertData()
# check insert data correct
self.checkInsertCorrect()
# save
self.snapshotAgg()
# do action
self.doAction()
# check save agg result correct
self.checkAggCorrect()
# check insert correct again
self.checkInsertCorrect()
tdLog.success(f"{__file__} successfully executed")
tdCases.addLinux(__file__, TDTestCase())
tdCases.addWindows(__file__, TDTestCase())

58
tests/army/enterprise/s3/s3_basic.json Normal file
View File

@ -0,0 +1,58 @@
{
"filetype": "insert",
"cfgdir": "/etc/taos",
"host": "127.0.0.1",
"port": 6030,
"user": "root",
"password": "taosdata",
"connection_pool_size": 8,
"num_of_records_per_req": 2000,
"thread_count": 2,
"create_table_thread_count": 1,
"confirm_parameter_prompt": "no",
"databases": [
{
"dbinfo": {
"name": "db",
"drop": "yes",
"vgroups": 2,
"replica": 1,
"duration":"1d",
"keep": "3d,6d,30d"
},
"super_tables": [
{
"name": "stb",
"child_table_exists": "no",
"childtable_count": 4,
"insert_rows": 1000000,
"childtable_prefix": "d",
"insert_mode": "taosc",
"timestamp_step": 1000,
"start_timestamp":"now-13d",
"columns": [
{ "type": "bool", "name": "bc"},
{ "type": "float", "name": "fc" },
{ "type": "double", "name": "dc"},
{ "type": "tinyint", "name": "ti", "values":["1"]},
{ "type": "smallint", "name": "si" },
{ "type": "int", "name": "ic" },
{ "type": "bigint", "name": "bi" },
{ "type": "utinyint", "name": "uti"},
{ "type": "usmallint", "name": "usi"},
{ "type": "uint", "name": "ui" },
{ "type": "ubigint", "name": "ubi"},
{ "type": "binary", "name": "bin", "len": 32},
{ "type": "nchar", "name": "nch", "len": 64}
],
"tags": [
{"type": "tinyint", "name": "groupid","max": 10,"min": 1},
{"name": "location","type": "binary", "len": 16, "values":
["San Francisco", "Los Angles", "San Diego", "San Jose", "Palo Alto", "Campbell", "Mountain View","Sunnyvale", "Santa Clara", "Cupertino"]
}
]
}
]
}
]
}

114
tests/army/enterprise/s3/s3_basic.py Normal file
View File

@ -0,0 +1,114 @@
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import sys
import time
import taos
import frame
import frame.etool
import frame.eos
from frame.log import *
from frame.cases import *
from frame.sql import *
from frame.caseBase import *
from frame.srvCtl import *
from frame import *
from frame.eos import *
#
# 192.168.1.52 MINIO S3 API KEY: MQCEIoaPGUs1mhXgpUAu:XTgpN2dEMInnYgqN4gj3G5zgb39ROtsisKKy0GFa
#
'''
s3EndPoint http://192.168.1.52:9000
s3AccessKey MQCEIoaPGUs1mhXgpUAu:XTgpN2dEMInnYgqN4gj3G5zgb39ROtsisKKy0GFa
s3BucketName ci-bucket
s3UploadDelaySec 60
'''
class TDTestCase(TBase):
updatecfgDict = {
's3EndPoint': 'http://192.168.1.52:9000',
's3AccessKey': 'MQCEIoaPGUs1mhXgpUAu:XTgpN2dEMInnYgqN4gj3G5zgb39ROtsisKKy0GFa',
's3BucketName': 'ci-bucket',
's3BlockSize': '10240',
's3BlockCacheSize': '320',
's3PageCacheSize': '10240',
's3UploadDelaySec':'60'
}
def insertData(self):
tdLog.info(f"insert data.")
# taosBenchmark run
json = etool.curFile(__file__, "s3_basic.json")
etool.runBenchmark(json=json)
tdSql.execute(f"use {self.db}")
# set insert data information
self.childtable_count = 4
self.insert_rows = 1000000
self.timestamp_step = 1000
def doAction(self):
tdLog.info(f"do action.")
self.flushDb()
self.compactDb()
# sleep 70s
tdLog.info(f"wait 65s ...")
time.sleep(65)
self.trimDb(True)
rootPath = sc.clusterRootPath()
cmd = f"ls {rootPath}/dnode1/data20/vnode/vnode*/tsdb/*.data"
tdLog.info(cmd)
loop = 0
while len(eos.runRetList(cmd)) > 0 and loop < 40:
time.sleep(5)
self.trimDb(True)
loop += 1
# run
def run(self):
tdLog.debug(f"start to excute {__file__}")
# insert data
self.insertData()
# check insert data correct
self.checkInsertCorrect()
# save
self.snapshotAgg()
# do action
self.doAction()
# check save agg result correct
self.checkAggCorrect()
# check insert correct again
self.checkInsertCorrect()
# drop database and free s3 file
self.dropDb()
tdLog.success(f"{__file__} successfully executed")
tdCases.addLinux(__file__, TDTestCase())
tdCases.addWindows(__file__, TDTestCase())

112
tests/army/frame/caseBase.py
View File

@ -15,6 +15,7 @@ import sys
import os
import time
import datetime
import random
from frame.log import *
from frame.sql import *
@ -42,17 +43,13 @@ class TBase:
self.db = "db"
self.stb = "stb"
# variant in taosBenchmark json
self.childtable_count = 2
self.insert_rows = 1000000
self.timestamp_step = 1000
# sql
self.sqlSum = f"select sum(ic) from {self.stb}"
self.sqlMax = f"select max(ic) from {self.stb}"
self.sqlMin = f"select min(ic) from {self.stb}"
self.sqlAvg = f"select avg(ic) from {self.stb}"
self.sqlFirst = f"select first(ts) from {self.stb}"
self.sqlLast = f"select last(ts) from {self.stb}"
# stop
def stop(self):
@ -63,14 +60,62 @@ class TBase:
# db action
#
def trimDb(self):
tdSql.execute(f"trim database {self.db}")
def trimDb(self, show = False):
tdSql.execute(f"trim database {self.db}", show = show)
def compactDb(self):
tdSql.execute(f"compact database {self.db}")
def compactDb(self, show = False):
tdSql.execute(f"compact database {self.db}", show = show)
def flushDb(self, show = False):
tdSql.execute(f"flush database {self.db}", show = show)
def dropDb(self, show = False):
tdSql.execute(f"drop database {self.db}", show = show)
def splitVGroups(self):
vgids = self.getVGroup(self.db)
selid = random.choice(vgids)
sql = f"split vgroup {selid}"
tdSql.execute(sql, show=True)
if self.waitTransactionZero() is False:
tdLog.exit(f"{sql} transaction not finished")
return False
return True
def alterReplica(self, replica):
sql = f"alter database {self.db} replica {replica}"
tdSql.execute(sql, show=True)
if self.waitTransactionZero() is False:
tdLog.exit(f"{sql} transaction not finished")
return False
return True
def balanceVGroup(self):
sql = f"balance vgroup"
tdSql.execute(sql, show=True)
if self.waitTransactionZero() is False:
tdLog.exit(f"{sql} transaction not finished")
return False
return True
def balanceVGroupLeader(self):
sql = f"balance vgroup leader"
tdSql.execute(sql, show=True)
if self.waitTransactionZero() is False:
tdLog.exit(f"{sql} transaction not finished")
return False
return True
def balanceVGroupLeaderOn(self, vgId):
sql = f"balance vgroup leader on {vgId}"
tdSql.execute(sql, show=True)
if self.waitTransactionZero() is False:
tdLog.exit(f"{sql} transaction not finished")
return False
return True
def flushDb(self):
tdSql.execute(f"flush database {self.db}")
#
# check db correct
@ -91,12 +136,13 @@ class TBase:
tdSql.checkAgg(sql, 0)
# save agg result
def snapshotAgg(self):
def snapshotAgg(self):
self.sum = tdSql.getFirstValue(self.sqlSum)
self.avg = tdSql.getFirstValue(self.sqlAvg)
self.min = tdSql.getFirstValue(self.sqlMin)
self.max = tdSql.getFirstValue(self.sqlMax)
self.first = tdSql.getFirstValue(self.sqlFirst)
self.last = tdSql.getFirstValue(self.sqlLast)
# check agg
def checkAggCorrect(self):
@ -104,3 +150,41 @@ class TBase:
tdSql.checkAgg(self.sqlAvg, self.avg)
tdSql.checkAgg(self.sqlMin, self.min)
tdSql.checkAgg(self.sqlMax, self.max)
tdSql.checkAgg(self.sqlFirst, self.first)
tdSql.checkAgg(self.sqlLast, self.last)
#
# get db information
#
# get vgroups
def getVGroup(self, db_name):
vgidList = []
sql = f"select vgroup_id from information_schema.ins_vgroups where db_name='{db_name}'"
res = tdSql.getResult(sql)
rows = len(res)
for i in range(rows):
vgidList.append(res[i][0])
return vgidList
#
# util
#
# wait transactions count to zero , return False is translation not finished
def waitTransactionZero(self, seconds = 300, interval = 1):
# wait end
for i in range(seconds):
sql ="show transactions;"
rows = tdSql.query(sql)
if rows == 0:
tdLog.info("transaction count became zero.")
return True
#tdLog.info(f"i={i} wait ...")
time.sleep(interval)
return False

21
tests/army/frame/eos.py
View File

@ -20,11 +20,16 @@ import os
import time
import datetime
import platform
import subprocess
# if windows platform return True
def isWin():
return platform.system().lower() == 'windows'
#
# execute programe
#
# wait util execute file finished
def exe(file):
return os.system(file)
@ -34,3 +39,19 @@ def exeNoWait(file):
print("exe no wait")
# run return output and error
def run(command):
process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
process.wait(3)
output = process.stdout.read().decode(encoding="gbk")
error = process.stderr.read().decode(encoding="gbk")
return output, error
# return list after run
def runRetList(command):
lines = []
output,error = run(command)
return output.splitlines()

3
tests/army/frame/epath.py
View File

@ -51,3 +51,6 @@ def binPath():
def binFile(filename):
return binPath() + filename

17
tests/army/frame/srvCtl.py
View File

@ -16,6 +16,8 @@ import os
import time
import datetime
from frame.server.dnodes import *
class srvCtl:
def __init__(self):
# record server information
@ -24,4 +26,19 @@ class srvCtl:
self.mLevel = 0
self.mLevelDisk = 0
#
# about path
#
# get cluster root path like /root/TDinternal/sim/
def clusterRootPath(self):
return tdDnodes.getDnodesRootDir()
# return dnode data files list
def dnodeDataFiles(self, idx):
files = []
return files
sc = srvCtl()

11
tests/parallel_test/cases.task
View File

@ -5,10 +5,17 @@
#unit-test
,,y,unit-test,bash test.sh
#army-test
#
# army-test
#
,,y,army,./pytest.sh python3 ./test.py -f enterprise/multi-level/mlevel_basic.py -N 3 -L 3 -D 2
,,y,army,./pytest.sh python3 ./test.py -f enterprise/s3/s3_basic.py -L 3 -D 1
,,y,army,./pytest.sh python3 ./test.py -f community/cluster/snapshot.py -N 3 -L 3 -D 2
#system test
#
# system test
#
,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/stream_basic.py
,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/scalar_function.py
,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/at_once_interval.py

2
tests/system-test/0-others/splitVGroupRep1.py
View File

@ -164,7 +164,7 @@ class TDTestCase:
self.c2Sum = None
# create database db
sql = f"create database @db_name vgroups {self.vgroups1} replica 1"
sql = f"create database @db_name vgroups {self.vgroups1} replica 1 wal_retention_period 1 wal_retention_size 1"
self.exeDouble(sql)
# create super talbe st

2
tests/system-test/8-stream/max_delay_session.py
View File

@ -27,6 +27,8 @@ class TDTestCase:
self.tb_stream_des_table = f'{self.tb_name}{self.tdCom.des_table_suffix}'
self.tdCom.date_time = self.tdCom.dataDict["start_ts"]
time.sleep(1)
if watermark is not None:
watermark_value = f'{self.tdCom.dataDict["watermark"]}s'
else:

1
tools/shell/inc/shellInt.h
View File

@ -66,6 +66,7 @@ typedef struct {
char file[PATH_MAX];
char password[TSDB_USET_PASSWORD_LEN];
bool is_gen_auth;
bool is_bi_mode;
bool is_raw_time;
bool is_version;
bool is_dump_config;

6
tools/shell/src/shellArguments.c
View File

@ -44,6 +44,7 @@
#define SHELL_NET_ROLE "Net role when network connectivity test, options: client|server."
#define SHELL_PKT_LEN "Packet length used for net test, default is 1024 bytes."
#define SHELL_PKT_NUM "Packet numbers used for net test, default is 100."
#define SHELL_BI_MODE "Set BI mode"
#define SHELL_VERSION "Print program version."
#ifdef WEBSOCKET
@ -59,6 +60,7 @@ void shellPrintHelp() {
printf("Usage: taos [OPTION...] \r\n\r\n");
printf("%s%s%s%s\r\n", indent, "-a,", indent, SHELL_AUTH);
printf("%s%s%s%s\r\n", indent, "-A,", indent, SHELL_GEN_AUTH);
printf("%s%s%s%s\r\n", indent, "-B,", indent, SHELL_BI_MODE);
printf("%s%s%s%s\r\n", indent, "-c,", indent, SHELL_CFG_DIR);
printf("%s%s%s%s\r\n", indent, "-C,", indent, SHELL_DMP_CFG);
printf("%s%s%s%s\r\n", indent, "-d,", indent, SHELL_DB);
@ -127,6 +129,7 @@ static struct argp_option shellOptions[] = {
{"timeout", 'T', "SECONDS", 0, SHELL_TIMEOUT},
#endif
{"pktnum", 'N', "PKTNUM", 0, SHELL_PKT_NUM},
{"bimode", 'B', 0, 0, SHELL_BI_MODE},
{0},
};
@ -173,6 +176,9 @@ static int32_t shellParseSingleOpt(int32_t key, char *arg) {
case 'A':
pArgs->is_gen_auth = true;
break;
case 'B':
pArgs->is_bi_mode = true;
break;
case 'c':
#ifdef WEBSOCKET
pArgs->cloud = false;

8
tools/shell/src/shellAuto.c
View File

@ -84,8 +84,8 @@ SWords shellCommands[] = {
{"alter topic", 0, 0, NULL},
{"alter user <user_name> <user_actions> <anyword> ;", 0, 0, NULL},
#ifdef TD_ENTERPRISE
{"balance vgroup;", 0, 0, NULL},
{"balance vgroup leader <vgroup_id>", 0, 0, NULL},
{"balance vgroup ;", 0, 0, NULL},
{"balance vgroup leader on <vgroup_id>", 0, 0, NULL},
#endif
// 20
@ -531,8 +531,8 @@ void showHelp() {
printf(
"\n\n\
----- special commands on enterpise version ----- \n\
balance vgroup; \n\
balance vgroup leader <vgroup_id> \n\
balance vgroup ;\n\
balance vgroup leader on <vgroup_id> \n\
compact database <db_name>; \n\
redistribute vgroup <vgroup_id> dnode <dnode_id> ;\n\
split vgroup <vgroup_id>;");

14
tools/shell/src/shellCommand.c
View File

@ -62,7 +62,6 @@ int32_t shellCountPrefixOnes(uint8_t c) {
}
void shellGetPrevCharSize(const char *str, int32_t pos, int32_t *size, int32_t *width) {
ASSERT(pos > 0);
if (pos <= 0) return;
TdWchar wc;
@ -82,7 +81,6 @@ void shellGetPrevCharSize(const char *str, int32_t pos, int32_t *size, int32_t *
}
void shellGetNextCharSize(const char *str, int32_t pos, int32_t *size, int32_t *width) {
ASSERT(pos >= 0);
if(pos < 0) return;
TdWchar wc;
@ -91,7 +89,6 @@ void shellGetNextCharSize(const char *str, int32_t pos, int32_t *size, int32_t *
}
void shellInsertChar(SShellCmd *cmd, char *c, int32_t size) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return;
TdWchar wc;
@ -138,7 +135,6 @@ void shellInsertStr(SShellCmd *cmd, char *str, int32_t size) {
}
void shellBackspaceChar(SShellCmd *cmd) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return;
if (cmd->cursorOffset > 0) {
@ -159,7 +155,6 @@ void shellBackspaceChar(SShellCmd *cmd) {
}
void shellClearLineBefore(SShellCmd *cmd) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return;
shellClearScreen(cmd->endOffset + PSIZE, cmd->screenOffset + PSIZE);
@ -174,7 +169,6 @@ void shellClearLineBefore(SShellCmd *cmd) {
}
void shellClearLineAfter(SShellCmd *cmd) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return;
shellClearScreen(cmd->endOffset + PSIZE, cmd->screenOffset + PSIZE);
@ -184,7 +178,6 @@ void shellClearLineAfter(SShellCmd *cmd) {
}
void shellDeleteChar(SShellCmd *cmd) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return;
if (cmd->cursorOffset < cmd->commandSize) {
@ -203,7 +196,6 @@ void shellDeleteChar(SShellCmd *cmd) {
}
void shellMoveCursorLeft(SShellCmd *cmd) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return;
if (cmd->cursorOffset > 0) {
@ -218,7 +210,6 @@ void shellMoveCursorLeft(SShellCmd *cmd) {
}
void shellMoveCursorRight(SShellCmd *cmd) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return;
if (cmd->cursorOffset < cmd->commandSize) {
@ -233,7 +224,6 @@ void shellMoveCursorRight(SShellCmd *cmd) {
}
void shellPositionCursorHome(SShellCmd *cmd) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return;
if (cmd->cursorOffset > 0) {
@ -254,7 +244,6 @@ void positionCursorMiddle(SShellCmd *cmd) {
}
void shellPositionCursorEnd(SShellCmd *cmd) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return;
if (cmd->cursorOffset < cmd->commandSize) {
@ -290,7 +279,6 @@ void shellPositionCursor(int32_t step, int32_t direction) {
}
void shellUpdateBuffer(SShellCmd *cmd) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return;
if (shellRegexMatch(cmd->buffer, "(\\s+$)|(^$)", REG_EXTENDED)) strcat(cmd->command, " ");
@ -306,7 +294,6 @@ void shellUpdateBuffer(SShellCmd *cmd) {
}
bool shellIsReadyGo(SShellCmd *cmd) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return false;
char *total = (char *)taosMemoryCalloc(1, SHELL_MAX_COMMAND_SIZE);
@ -334,7 +321,6 @@ void shellGetMbSizeInfo(const char *str, int32_t *size, int32_t *width) {
}
void shellResetCommand(SShellCmd *cmd, const char s[]) {
ASSERT(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
if(cmd->cursorOffset > cmd->commandSize || cmd->endOffset < cmd->screenOffset) return;
shellClearScreen(cmd->endOffset + PSIZE, cmd->screenOffset + PSIZE);

10
tools/shell/src/shellEngine.c
View File

@ -1291,6 +1291,16 @@ int32_t shellExecute() {
shellSetConn(shell.conn, runOnce);
shellReadHistory();
if(shell.args.is_bi_mode) {
// need set bi mode
printf("Set BI mode is true.\n");
#ifdef WEBSOCKET
//ws_taos_set_conn_mode(shell.ws_conn, TAOS_CONN_MODE_BI, 1);
#else
taos_set_conn_mode(shell.conn, TAOS_CONN_MODE_BI, 1);
#endif
}
if (runOnce) {
if (pArgs->commands != NULL) {
printf("%s%s\r\n", shell.info.promptHeader, pArgs->commands);