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

Merge branch '3.0' into enh/refactorBackend

This commit is contained in:
yihaoDeng 2023-11-30 10:31:27 +08:00
parent 08cc976975 0d04dc296e
commit 5c32680cd8
42 changed files with 3751 additions and 2687 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
Jenkinsfile2
View File

@ -355,7 +355,7 @@ pipeline {
}
parallel {
stage('check docs') {
agent{label " slave1_47 || slave1_48 || slave1_49 || slave1_52 || worker03 || slave215 || slave217 || slave219 || Mac_catalina "}
agent{label " slave1_47 || slave1_48 || slave1_49 || slave1_50 || slave1_52 || slave1_59 || slave1_63 || worker03 || slave215 || slave217 || slave219 || Mac_catalina "}
steps {
check_docs()
}
@ -401,7 +401,7 @@ pipeline {
}
}
stage('linux test') {
agent{label " slave1_47 || slave1_48 || slave1_49 || slave1_52 || worker03 || slave215 || slave217 || slave219 "}
agent{label " slave1_47 || slave1_48 || slave1_49 || slave1_50 || slave1_52 || slave1_59 || slave1_63 || worker03 || slave215 || slave217 || slave219 "}
options { skipDefaultCheckout() }
when {
changeRequest()

17
cmake/cmake.define
View File

@ -1,5 +1,5 @@
cmake_minimum_required(VERSION 3.0)
set(CMAKE_VERBOSE_MAKEFILE FALSE)
set(CMAKE_VERBOSE_MAKEFILE TRUE)
set(TD_BUILD_TAOSA_INTERNAL FALSE)
#set output directory
@ -159,6 +159,7 @@ ELSE ()
CHECK_C_COMPILER_FLAG("-mavx2" COMPILER_SUPPORT_AVX2)
CHECK_C_COMPILER_FLAG("-mavx512f" COMPILER_SUPPORT_AVX512F)
CHECK_C_COMPILER_FLAG("-mavx512vbmi" COMPILER_SUPPORT_AVX512BMI)
CHECK_C_COMPILER_FLAG("-mavx512vl" COMPILER_SUPPORT_AVX512VL)
IF (COMPILER_SUPPORT_SSE42)
SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -msse4.2")
@ -166,11 +167,11 @@ ELSE ()
ENDIF()
IF ("${SIMD_SUPPORT}" MATCHES "true")
IF (COMPILER_SUPPORT_FMA)
IF (COMPILER_SUPPORT_FMA)
SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -mfma")
SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -mfma")
ENDIF()
IF (COMPILER_SUPPORT_AVX)
ENDIF()
IF (COMPILER_SUPPORT_AVX)
SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -mavx")
SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -mavx")
ENDIF()
@ -183,7 +184,13 @@ ELSE ()
IF (COMPILER_SUPPORT_AVX512F AND COMPILER_SUPPORT_AVX512BMI)
SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -mavx512f -mavx512vbmi")
SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -mavx512f -mavx512vbmi")
MESSAGE(STATUS "avx512 supported by gcc")
MESSAGE(STATUS "avx512f/avx512bmi supported by compiler")
ENDIF()
IF (COMPILER_SUPPORT_AVX512VL)
SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -mavx512vl")
SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -mavx512vl")
MESSAGE(STATUS "avx512vl supported by compiler")
ENDIF()
ENDIF()

3
include/common/tmsg.h
View File

@ -414,7 +414,8 @@ typedef enum ENodeType {
QUERY_NODE_PHYSICAL_PLAN_STREAM_EVENT,
QUERY_NODE_PHYSICAL_PLAN_HASH_JOIN,
QUERY_NODE_PHYSICAL_PLAN_GROUP_CACHE,
QUERY_NODE_PHYSICAL_PLAN_DYN_QUERY_CTRL
QUERY_NODE_PHYSICAL_PLAN_DYN_QUERY_CTRL,
QUERY_NODE_END
} ENodeType;
typedef struct {

1
include/libs/nodes/nodes.h
View File

@ -110,6 +110,7 @@ int64_t nodesMakeAllocatorWeakRef(int64_t allocatorId);
int64_t nodesReleaseAllocatorWeakRef(int64_t allocatorId);
void nodesDestroyAllocator(int64_t allocatorId);
int32_t getNodeSize(ENodeType type);
SNode* nodesMakeNode(ENodeType type);
void nodesDestroyNode(SNode* pNode);
void nodesFree(void* p);

1
include/libs/nodes/querynodes.h
View File

@ -449,6 +449,7 @@ typedef struct SVnodeModifyOpStmt {
SHashObj* pSubTableHashObj; // SHashObj<table_name, STableMeta*>
SHashObj* pTableNameHashObj; // set of table names for refreshing meta, sync mode
SHashObj* pDbFNameHashObj; // set of db names for refreshing meta, sync mode
SHashObj* pTableCxtHashObj; // temp SHashObj<tuid, STableDataCxt*> for single request
SArray* pVgDataBlocks; // SArray<SVgroupDataCxt*>
SVCreateTbReq* pCreateTblReq;
TdFilePtr fp;

16
include/libs/stream/tstream.h
View File

@ -674,13 +674,15 @@ typedef struct STaskStatusEntry {
int32_t statusLastDuration; // to record the last duration of current status
int64_t stage;
int32_t nodeId;
int64_t verStart; // start version in WAL, only valid for source task
int64_t verEnd; // end version in WAL, only valid for source task
int64_t processedVer; // only valid for source task
int32_t relatedHTask; // has related fill-history task
int64_t activeCheckpointId; // current active checkpoint id
bool checkpointFailed; // denote if the checkpoint is failed or not
double inputQUsed; // in MiB
int64_t verStart; // start version in WAL, only valid for source task
int64_t verEnd; // end version in WAL, only valid for source task
int64_t processedVer; // only valid for source task
int32_t relatedHTask; // has related fill-history task
int64_t activeCheckpointId; // current active checkpoint id
bool checkpointFailed; // denote if the checkpoint is failed or not
bool inputQChanging; // inputQ is changing or not
int64_t inputQUnchangeCounter;
double inputQUsed; // in MiB
double inputRate;
double sinkQuota; // existed quota size for sink task
double sinkDataSize; // sink to dst data size

2
include/util/tcompression.h
View File

@ -139,6 +139,8 @@ int32_t getWordLength(char type);
int32_t tsDecompressIntImpl_Hw(const char *const input, const int32_t nelements, char *const output, const char type);
int32_t tsDecompressFloatImplAvx512(const char *const input, const int32_t nelements, char *const output);
int32_t tsDecompressFloatImplAvx2(const char *const input, const int32_t nelements, char *const output);
int32_t tsDecompressTimestampAvx512(const char* const input, const int32_t nelements, char *const output, bool bigEndian);
int32_t tsDecompressTimestampAvx2(const char* const input, const int32_t nelements, char *const output, bool bigEndian);
/*************************************************************************
* STREAM COMPRESSION

1
packaging/tools/set_core.sh
View File

@ -38,3 +38,4 @@ source /etc/profile
${csudo}mkdir -p ${corePath} ||:
${csudo}sysctl -w kernel.core_pattern=${corePath}/core-%e-%p ||:
${csudo}echo "${corePath}/core-%e-%p" | ${csudo}tee /proc/sys/kernel/core_pattern ||:
${csudo}echo "kernel.core_pattern = ${corePath}/core_%e-%p" >> /etc/sysctl.conf ||:

102
source/client/test/clientTests.cpp
View File

@ -126,9 +126,9 @@ void queryCallback(void* param, void* res, int32_t code) {
taos_fetch_raw_block_a(res, fetchCallback, param);
}
void createNewTable(TAOS* pConn, int32_t index) {
void createNewTable(TAOS* pConn, int32_t index, int32_t numOfRows, int64_t startTs, const char* pVarchar) {
char str[1024] = {0};
sprintf(str, "create table tu%d using st2 tags(%d)", index, index);
sprintf(str, "create table if not exists tu%d using st2 tags(%d)", index, index);
TAOS_RES* pRes = taos_query(pConn, str);
if (taos_errno(pRes) != 0) {
@ -136,22 +136,43 @@ void createNewTable(TAOS* pConn, int32_t index) {
}
taos_free_result(pRes);
for (int32_t i = 0; i < 10000; i += 20) {
char sql[1024] = {0};
sprintf(sql,
"insert into tu%d values(now+%da, %d)(now+%da, %d)(now+%da, %d)(now+%da, %d)"
"(now+%da, %d)(now+%da, %d)(now+%da, %d)(now+%da, %d)(now+%da, %d)(now+%da, %d)"
"(now+%da, %d)(now+%da, %d)(now+%da, %d)(now+%da, %d)"
"(now+%da, %d)(now+%da, %d)(now+%da, %d)(now+%da, %d)(now+%da, %d)(now+%da, %d)",
index, i, i, i + 1, i + 1, i + 2, i + 2, i + 3, i + 3, i + 4, i + 4, i + 5, i + 5, i + 6, i + 6, i + 7,
i + 7, i + 8, i + 8, i + 9, i + 9, i + 10, i + 10, i + 11, i + 11, i + 12, i + 12, i + 13, i + 13, i + 14,
i + 14, i + 15, i + 15, i + 16, i + 16, i + 17, i + 17, i + 18, i + 18, i + 19, i + 19);
TAOS_RES* p = taos_query(pConn, sql);
if (taos_errno(p) != 0) {
printf("failed to insert data, reason:%s\n", taos_errstr(p));
}
if (startTs == 0) {
for (int32_t i = 0; i < numOfRows; i += 20) {
char sql[1024] = {0};
sprintf(sql,
"insert into tu%d values(now+%da, %d)(now+%da, %d)(now+%da, %d)(now+%da, %d)"
"(now+%da, %d)(now+%da, %d)(now+%da, %d)(now+%da, %d)(now+%da, %d)(now+%da, %d)"
"(now+%da, %d)(now+%da, %d)(now+%da, %d)(now+%da, %d)"
"(now+%da, %d)(now+%da, %d)(now+%da, %d)(now+%da, %d)(now+%da, %d)(now+%da, %d)",
index, i, i, i + 1, i + 1, i + 2, i + 2, i + 3, i + 3, i + 4, i + 4, i + 5, i + 5, i + 6, i + 6, i + 7,
i + 7, i + 8, i + 8, i + 9, i + 9, i + 10, i + 10, i + 11, i + 11, i + 12, i + 12, i + 13, i + 13, i + 14,
i + 14, i + 15, i + 15, i + 16, i + 16, i + 17, i + 17, i + 18, i + 18, i + 19, i + 19);
TAOS_RES* p = taos_query(pConn, sql);
if (taos_errno(p) != 0) {
printf("failed to insert data, reason:%s\n", taos_errstr(p));
}
taos_free_result(p);
taos_free_result(p);
}
} else {
for (int32_t i = 0; i < numOfRows; i += 20) {
char sql[1024*50] = {0};
sprintf(sql,
"insert into tu%d values(%ld, %d, '%s')(%ld, %d, '%s')(%ld, %d, '%s')(%ld, %d, '%s')(%ld, %d, '%s')(%ld, "
"%d, '%s')(%ld, %d, '%s')(%ld, %d, '%s')(%ld, %d, '%s')(%ld, %d, '%s')(%ld, %d, '%s')(%ld, %d, '%s')(%ld, %d, '%s')(%ld, %d, '%s')(%ld, %d, '%s')(%ld, %d, "
"'%s')(%ld, %d, '%s')(%ld, %d, '%s')(%ld, %d, '%s')(%ld, %d, '%s')",
index, startTs, i, pVarchar, startTs + 1, i + 1, pVarchar, startTs + 2, i + 2, pVarchar, startTs + 3, i + 3, pVarchar, startTs + 4, i + 4,
pVarchar, startTs + 5, i + 5, pVarchar, startTs + 6, i + 6, pVarchar, startTs + 7, i + 7, pVarchar, startTs + 8, i + 8, pVarchar, startTs + 9, i + 9,
pVarchar, startTs + 10, i + 10, pVarchar, startTs + 11, i + 11, pVarchar, startTs + 12, i + 12, pVarchar, startTs + 13, i + 13, pVarchar, startTs + 14,
i + 14, pVarchar, startTs + 15, i + 15, pVarchar, startTs + 16, i + 16, pVarchar, startTs + 17, i + 17, pVarchar, startTs + 18, i + 18,
pVarchar, startTs + 19, i + 19, pVarchar);
TAOS_RES* p = taos_query(pConn, sql);
if (taos_errno(p) != 0) {
printf("failed to insert data, reason:%s\n", taos_errstr(p));
}
taos_free_result(p);
}
}
}
@ -808,14 +829,7 @@ TEST(clientCase, projection_query_tables) {
TAOS_RES* pRes = taos_query(pConn, "use abc1");
taos_free_result(pRes);
pRes = taos_query(pConn, "create stable st1 (ts timestamp, k int) tags(a int)");
if (taos_errno(pRes) != 0) {
printf("failed to create table tu, reason:%s\n", taos_errstr(pRes));
}
taos_free_result(pRes);
pRes = taos_query(pConn, "create stable st2 (ts timestamp, k int) tags(a int)");
pRes = taos_query(pConn, "create stable st2 (ts timestamp, k int, f varchar(4096)) tags(a int)");
if (taos_errno(pRes) != 0) {
printf("failed to create table tu, reason:%s\n", taos_errstr(pRes));
}
@ -828,28 +842,32 @@ TEST(clientCase, projection_query_tables) {
taos_free_result(pRes);
int64_t start = 1685959190000;
const char* pstr =
"abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefgh"
"ijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnop"
"qrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwx"
"yzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdef"
"ghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz!@#$%^&&*&^^%$#@!qQWERTYUIOPASDFGHJKL:"
"QWERTYUIOP{}";
int32_t code = -1;
for(int32_t i = 0; i < 1000000; ++i) {
char t[512] = {0};
for(int32_t i = 0; i < 10000; ++i) {
char str[1024] = {0};
sprintf(str, "create table if not exists tu%d using st2 tags(%d)", i, i);
sprintf(t, "insert into t1 values(now, %d)", i);
while(1) {
void* p = taos_query(pConn, t);
code = taos_errno(p);
taos_free_result(p);
if (code != 0) {
printf("insert data error, retry\n");
} else {
break;
}
TAOS_RES* px = taos_query(pConn, str);
if (taos_errno(px) != 0) {
printf("failed to create table tu, reason:%s\n", taos_errstr(pRes));
}
taos_free_result(px);
}
for(int32_t j = 0; j < 5000; ++j) {
start += 20;
for (int32_t i = 0; i < 10000; ++i) {
createNewTable(pConn, i, 20, start, pstr);
}
}
for (int32_t i = 0; i < 1; ++i) {
printf("create table :%d\n", i);
createNewTable(pConn, i);
}
//
// pRes = taos_query(pConn, "select * from tu");
// if (taos_errno(pRes) != 0) {

22
source/dnode/mnode/impl/inc/mndStream.h
View File

@ -31,7 +31,7 @@ typedef struct SStreamTransInfo {
// time to generated the checkpoint, if now() - checkpointTs >= tsCheckpointInterval, this checkpoint will be discard
// to avoid too many checkpoints for a taskk in the waiting list
typedef struct SCheckpointCandEntry {
char* pName;
char * pName;
int64_t streamId;
int64_t checkpointTs;
int64_t checkpointId;
@ -43,13 +43,13 @@ typedef struct SStreamTransMgmt {
} SStreamTransMgmt;
typedef struct SStreamExecInfo {
SArray *pNodeList;
int64_t ts; // snapshot ts
SStreamTransMgmt transMgmt;
int64_t activeCheckpoint; // active check point id
SHashObj * pTaskMap;
SArray * pTaskList;
TdThreadMutex lock;
SArray * pNodeList;
int64_t ts; // snapshot ts
SStreamTransMgmt transMgmt;
int64_t activeCheckpoint; // active check point id
SHashObj * pTaskMap;
SArray * pTaskList;
TdThreadMutex lock;
} SStreamExecInfo;
extern SStreamExecInfo execInfo;
@ -61,9 +61,9 @@ void mndReleaseStream(SMnode *pMnode, SStreamObj *pStream);
int32_t mndDropStreamByDb(SMnode *pMnode, STrans *pTrans, SDbObj *pDb);
int32_t mndPersistStream(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream);
int32_t mndStreamRegisterTrans(STrans* pTrans, const char* pName, const char* pSrcDb, const char* pDstDb);
bool mndStreamTransConflictOtherTrans(SMnode *pMnode, const char *pSrcDb, const char *pDstDb);
int32_t mndAddtoCheckpointWaitingList(SStreamObj* pStream, int64_t checkpointId);
int32_t mndStreamRegisterTrans(STrans *pTrans, const char *pName, const char *pSrcDb, const char *pDstDb);
int32_t mndAddtoCheckpointWaitingList(SStreamObj *pStream, int64_t checkpointId);
bool streamTransConflictOtherTrans(SMnode *pMnode, const char *pSrcDb, const char *pDstDb, bool lock);
// for sma
// TODO refactor

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

@ -797,12 +797,14 @@ static int32_t checkForNumOfStreams(SMnode *pMnode, SStreamObj *pStreamObj) { /
}
static int32_t mndProcessCreateStreamReq(SRpcMsg *pReq) {
SMnode * pMnode = pReq->info.node;
int32_t code = -1;
SStreamObj * pStream = NULL;
SCMCreateStreamReq createStreamReq = {0};
SStreamObj streamObj = {0};
SMnode * pMnode = pReq->info.node;
int32_t code = -1;
SStreamObj *pStream = NULL;
SStreamObj streamObj = {0};
char * sql = NULL;
int32_t sqlLen = 0;
SCMCreateStreamReq createStreamReq = {0};
if (tDeserializeSCMCreateStreamReq(pReq->pCont, pReq->contLen, &createStreamReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
goto _OVER;
@ -833,8 +835,6 @@ static int32_t mndProcessCreateStreamReq(SRpcMsg *pReq) {
goto _OVER;
}
char * sql = NULL;
int32_t sqlLen = 0;
if (createStreamReq.sql != NULL) {
sqlLen = strlen(createStreamReq.sql);
sql = taosMemoryMalloc(sqlLen + 1);
@ -1043,7 +1043,7 @@ static int32_t mndProcessStreamCheckpointTrans(SMnode *pMnode, SStreamObj *pStre
return -1;
}
bool conflict = mndStreamTransConflictOtherTrans(pMnode, pStream->sourceDb, pStream->targetDb);
bool conflict = streamTransConflictOtherTrans(pMnode, pStream->sourceDb, pStream->targetDb, true);
if (conflict) {
mndAddtoCheckpointWaitingList(pStream, checkpointId);
mWarn("checkpoint conflict with other trans in %s, ignore the checkpoint for stream:%s %" PRIx64, pStream->sourceDb,
@ -1395,7 +1395,7 @@ static int32_t mndProcessDropStreamReq(SRpcMsg *pReq) {
}
// check if it is conflict with other trans in both sourceDb and targetDb.
bool conflict = mndStreamTransConflictOtherTrans(pMnode, pStream->sourceDb, pStream->targetDb);
bool conflict = streamTransConflictOtherTrans(pMnode, pStream->sourceDb, pStream->targetDb, true);
if (conflict) {
sdbRelease(pMnode->pSdb, pStream);
tFreeMDropStreamReq(&dropReq);
@ -1878,7 +1878,7 @@ static int32_t mndProcessPauseStreamReq(SRpcMsg *pReq) {
}
// check if it is conflict with other trans in both sourceDb and targetDb.
bool conflict = mndStreamTransConflictOtherTrans(pMnode, pStream->sourceDb, pStream->targetDb);
bool conflict = streamTransConflictOtherTrans(pMnode, pStream->sourceDb, pStream->targetDb, true);
if (conflict) {
sdbRelease(pMnode->pSdb, pStream);
return -1;
@ -2013,7 +2013,7 @@ static int32_t mndProcessResumeStreamReq(SRpcMsg *pReq) {
}
// check if it is conflict with other trans in both sourceDb and targetDb.
bool conflict = mndStreamTransConflictOtherTrans(pMnode, pStream->sourceDb, pStream->targetDb);
bool conflict = streamTransConflictOtherTrans(pMnode, pStream->sourceDb, pStream->targetDb, true);
if (conflict) {
sdbRelease(pMnode->pSdb, pStream);
return -1;
@ -2800,7 +2800,7 @@ static int32_t mndResetStatusFromCheckpoint(SMnode *pMnode, int32_t transId) {
break;
}
bool conflict = mndStreamTransConflictOtherTrans(pMnode, pStream->sourceDb, pStream->targetDb);
bool conflict = streamTransConflictOtherTrans(pMnode, pStream->sourceDb, pStream->targetDb, false);
if (conflict) {
mError("stream:%s other trans exists in DB:%s & %s failed to start reset-status trans", pStream->name,
pStream->sourceDb, pStream->targetDb);
@ -2835,7 +2835,7 @@ static SStreamTask *mndGetStreamTask(STaskId *pId, SStreamObj *pStream) {
static bool needDropRelatedFillhistoryTask(STaskStatusEntry *pTaskEntry, SStreamExecInfo *pExecNode) {
if (pTaskEntry->status == TASK_STATUS__STREAM_SCAN_HISTORY && pTaskEntry->statusLastDuration >= 10) {
if (fabs(pTaskEntry->inputQUsed) <= DBL_EPSILON) {
if (!pTaskEntry->inputQChanging && pTaskEntry->inputQUnchangeCounter > 10) {
int32_t numOfReady = 0;
int32_t numOfTotal = 0;
for (int32_t k = 0; k < taosArrayGetSize(pExecNode->pTaskList); ++k) {
@ -2978,6 +2978,7 @@ int32_t mndProcessStreamHb(SRpcMsg *pReq) {
bool snodeChanged = false;
for (int32_t i = 0; i < req.numOfTasks; ++i) {
STaskStatusEntry *p = taosArrayGet(req.pTaskStatus, i);
STaskStatusEntry *pTaskEntry = taosHashGet(execInfo.pTaskMap, &p->id, sizeof(p->id));
if (pTaskEntry == NULL) {
mError("s-task:0x%" PRIx64 " not found in mnode task list", p->id.taskId);
@ -2988,6 +2989,18 @@ int32_t mndProcessStreamHb(SRpcMsg *pReq) {
updateStageInfo(pTaskEntry, p->stage);
if (pTaskEntry->nodeId == SNODE_HANDLE) snodeChanged = true;
} else {
// task is idle for more than 50 sec.
if (fabs(pTaskEntry->inputQUsed - p->inputQUsed) <= DBL_EPSILON) {
if (!pTaskEntry->inputQChanging) {
pTaskEntry->inputQUnchangeCounter++;
} else {
pTaskEntry->inputQChanging = false;
}
} else {
pTaskEntry->inputQChanging = true;
pTaskEntry->inputQUnchangeCounter = 0;
}
streamTaskStatusCopy(pTaskEntry, p);
if (p->activeCheckpointId != 0) {
if (activeCheckpointId != 0) {

37
source/dnode/mnode/impl/src/mndStreamTrans.c
View File

@ -35,17 +35,15 @@ int32_t mndStreamRegisterTrans(STrans* pTrans, const char* pName, const char* pS
}
int32_t clearFinishedTrans(SMnode* pMnode) {
SArray* pList = taosArrayInit(4, sizeof(SKeyInfo));
size_t keyLen = 0;
SArray* pList = taosArrayInit(4, sizeof(SKeyInfo));
void* pIter = NULL;
taosThreadMutexLock(&execInfo.lock);
void* pIter = NULL;
while ((pIter = taosHashIterate(execInfo.transMgmt.pDBTrans, pIter)) != NULL) {
SStreamTransInfo* pEntry = (SStreamTransInfo*)pIter;
STrans* pTrans = mndAcquireTrans(pMnode, pEntry->transId);
// let's clear the finished trans
STrans* pTrans = mndAcquireTrans(pMnode, pEntry->transId);
if (pTrans == NULL) {
void* pKey = taosHashGetKey(pEntry, &keyLen);
// key is the name of src/dst db name
@ -66,38 +64,49 @@ int32_t clearFinishedTrans(SMnode* pMnode) {
}
mDebug("clear %d finished stream-trans, remained:%d", (int32_t)num, taosHashGetSize(execInfo.transMgmt.pDBTrans));
taosThreadMutexUnlock(&execInfo.lock);
terrno = TSDB_CODE_SUCCESS;
taosArrayDestroy(pList);
return 0;
}
bool mndStreamTransConflictOtherTrans(SMnode* pMnode, const char* pSrcDb, const char* pDstDb) {
clearFinishedTrans(pMnode);
bool streamTransConflictOtherTrans(SMnode* pMnode, const char* pSrcDb, const char* pDstDb, bool lock) {
if (lock) {
taosThreadMutexLock(&execInfo.lock);
}
taosThreadMutexLock(&execInfo.lock);
int32_t num = taosHashGetSize(execInfo.transMgmt.pDBTrans);
if (num <= 0) {
taosThreadMutexUnlock(&execInfo.lock);
if (lock) {
taosThreadMutexUnlock(&execInfo.lock);
}
return false;
}
SStreamTransInfo* pEntry = taosHashGet(execInfo.transMgmt.pDBTrans, pSrcDb, strlen(pSrcDb));
clearFinishedTrans(pMnode);
SStreamTransInfo *pEntry = taosHashGet(execInfo.transMgmt.pDBTrans, pSrcDb, strlen(pSrcDb));
if (pEntry != NULL) {
taosThreadMutexUnlock(&execInfo.lock);
if (lock) {
taosThreadMutexUnlock(&execInfo.lock);
}
mWarn("conflict with other transId:%d in Db:%s, trans:%s", pEntry->transId, pSrcDb, pEntry->name);
return true;
}
pEntry = taosHashGet(execInfo.transMgmt.pDBTrans, pDstDb, strlen(pDstDb));
if (pEntry != NULL) {
taosThreadMutexUnlock(&execInfo.lock);
if (lock) {
taosThreadMutexUnlock(&execInfo.lock);
}
mWarn("conflict with other transId:%d in Db:%s, trans:%s", pEntry->transId, pSrcDb, pEntry->name);
return true;
}
taosThreadMutexUnlock(&execInfo.lock);
if (lock) {
taosThreadMutexUnlock(&execInfo.lock);
}
return false;
}

31
source/dnode/mnode/impl/src/mndUser.c
View File

@ -708,9 +708,8 @@ SSdbRaw *mndUserActionEncode(SUserObj *pUser) {
int32_t numOfAlterViews = taosHashGetSize(pUser->alterViews);
int32_t numOfTopics = taosHashGetSize(pUser->topics);
int32_t numOfUseDbs = taosHashGetSize(pUser->useDbs);
int32_t size = sizeof(SUserObj) + USER_RESERVE_SIZE +
(numOfReadDbs + numOfWriteDbs + numOfUseDbs) * TSDB_DB_FNAME_LEN + numOfTopics * TSDB_TOPIC_FNAME_LEN +
ipWhiteReserve;
int32_t size = sizeof(SUserObj) + USER_RESERVE_SIZE + (numOfReadDbs + numOfWriteDbs) * TSDB_DB_FNAME_LEN +
numOfTopics * TSDB_TOPIC_FNAME_LEN + ipWhiteReserve;
char *stb = taosHashIterate(pUser->readTbs, NULL);
while (stb != NULL) {
@ -720,7 +719,7 @@ SSdbRaw *mndUserActionEncode(SUserObj *pUser) {
size += keyLen;
size_t valueLen = 0;
valueLen = strlen(stb);
valueLen = strlen(stb) + 1;
size += sizeof(int32_t);
size += valueLen;
stb = taosHashIterate(pUser->readTbs, stb);
@ -734,7 +733,7 @@ SSdbRaw *mndUserActionEncode(SUserObj *pUser) {
size += keyLen;
size_t valueLen = 0;
valueLen = strlen(stb);
valueLen = strlen(stb) + 1;
size += sizeof(int32_t);
size += valueLen;
stb = taosHashIterate(pUser->writeTbs, stb);
@ -748,7 +747,7 @@ SSdbRaw *mndUserActionEncode(SUserObj *pUser) {
size += keyLen;
size_t valueLen = 0;
valueLen = strlen(stb);
valueLen = strlen(stb) + 1;
size += sizeof(int32_t);
size += valueLen;
stb = taosHashIterate(pUser->alterTbs, stb);
@ -762,7 +761,7 @@ SSdbRaw *mndUserActionEncode(SUserObj *pUser) {
size += keyLen;
size_t valueLen = 0;
valueLen = strlen(stb);
valueLen = strlen(stb) + 1;
size += sizeof(int32_t);
size += valueLen;
stb = taosHashIterate(pUser->readViews, stb);
@ -776,7 +775,7 @@ SSdbRaw *mndUserActionEncode(SUserObj *pUser) {
size += keyLen;
size_t valueLen = 0;
valueLen = strlen(stb);
valueLen = strlen(stb) + 1;
size += sizeof(int32_t);
size += valueLen;
stb = taosHashIterate(pUser->writeViews, stb);
@ -790,11 +789,21 @@ SSdbRaw *mndUserActionEncode(SUserObj *pUser) {
size += keyLen;
size_t valueLen = 0;
valueLen = strlen(stb);
valueLen = strlen(stb) + 1;
size += sizeof(int32_t);
size += valueLen;
stb = taosHashIterate(pUser->alterViews, stb);
}
}
int32_t *useDb = taosHashIterate(pUser->useDbs, NULL);
while (useDb != NULL) {
size_t keyLen = 0;
void *key = taosHashGetKey(useDb, &keyLen);
size += sizeof(int32_t);
size += keyLen;
size += sizeof(int32_t);
useDb = taosHashIterate(pUser->useDbs, useDb);
}
SSdbRaw *pRaw = sdbAllocRaw(SDB_USER, USER_VER_NUMBER, size);
if (pRaw == NULL) goto _OVER;
@ -925,7 +934,7 @@ SSdbRaw *mndUserActionEncode(SUserObj *pUser) {
stb = taosHashIterate(pUser->alterViews, stb);
}
int32_t *useDb = taosHashIterate(pUser->useDbs, NULL);
useDb = taosHashIterate(pUser->useDbs, NULL);
while (useDb != NULL) {
size_t keyLen = 0;
void *key = taosHashGetKey(useDb, &keyLen);

4
source/dnode/vnode/src/inc/vnodeInt.h
View File

@ -93,7 +93,11 @@ typedef struct SQueryNode SQueryNode;
#define VNODE_RSMA2_DIR "rsma2"
#define VNODE_TQ_STREAM "stream"
#if SUSPEND_RESUME_TEST // only for test purpose
#define VNODE_BUFPOOL_SEGMENTS 1
#else
#define VNODE_BUFPOOL_SEGMENTS 3
#endif
#define VND_INFO_FNAME "vnode.json"
#define VND_INFO_FNAME_TMP "vnode_tmp.json"

3
source/dnode/vnode/src/tq/tq.c
View File

@ -1423,6 +1423,9 @@ int32_t tqProcessTaskDropHTask(STQ* pTq, SRpcMsg* pMsg) {
SStreamTaskId id = {.streamId = pTask->hTaskInfo.id.streamId, .taskId = pTask->hTaskInfo.id.taskId};
streamBuildAndSendDropTaskMsg(pTask->pMsgCb, pMeta->vgId, &id);
// clear the scheduler status
streamTaskSetSchedStatusInactive(pTask);
tqDebug("s-task:%s set scheduler status:%d after drop fill-history task", pTask->id.idStr, pTask->status.schedStatus);
streamMetaReleaseTask(pMeta, pTask);
return TSDB_CODE_SUCCESS;
}

167
source/dnode/vnode/src/tsdb/tsdbRead2.c
View File

@ -48,7 +48,7 @@ static int32_t doMergeMemIMemRows(TSDBROW* pRow, TSDBROW* piRow, STableBlockScan
static int32_t mergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pBlockScanInfo, int64_t key,
STsdbReader* pReader);
static int32_t initDelSkylineIterator(STableBlockScanInfo* pBlockScanInfo, int32_t order, SCostSummary* pCost);
static int32_t initDelSkylineIterator(STableBlockScanInfo* pBlockScanInfo, int32_t order, SReadCostSummary* pCost);
static STsdb* getTsdbByRetentions(SVnode* pVnode, SQueryTableDataCond* pCond, SRetention* retentions, const char* idstr,
int8_t* pLevel);
static SVersionRange getQueryVerRange(SVnode* pVnode, SQueryTableDataCond* pCond, int8_t level);
@ -58,6 +58,7 @@ static TSDBKEY getCurrentKeyInBuf(STableBlockScanInfo* pScanInfo, STsdbRea
static bool hasDataInFileBlock(const SBlockData* pBlockData, const SFileBlockDumpInfo* pDumpInfo);
static void initBlockDumpInfo(STsdbReader* pReader, SDataBlockIter* pBlockIter);
static int32_t getInitialDelIndex(const SArray* pDelSkyline, int32_t order);
static void updateComposedBlockInfo(STsdbReader* pReader, double el, STableBlockScanInfo* pBlockScanInfo);
static bool outOfTimeWindow(int64_t ts, STimeWindow* pWindow) { return (ts > pWindow->ekey) || (ts < pWindow->skey); }
@ -168,7 +169,7 @@ static int32_t filesetIteratorNext(SFilesetIter* pIter, STsdbReader* pReader, bo
return TSDB_CODE_SUCCESS;
}
SCostSummary* pCost = &pReader->cost;
SReadCostSummary* pCost = &pReader->cost;
pIter->pLastBlockReader->uid = 0;
tMergeTreeClose(&pIter->pLastBlockReader->mergeTree);
@ -291,11 +292,7 @@ static SSDataBlock* createResBlock(SQueryTableDataCond* pCond, int32_t capacity)
}
static int32_t tsdbInitReaderLock(STsdbReader* pReader) {
int32_t code = -1;
qTrace("tsdb/read: %p, pre-init read mutex: %p, code: %d", pReader, &pReader->readerMutex, code);
code = taosThreadMutexInit(&pReader->readerMutex, NULL);
int32_t code = taosThreadMutexInit(&pReader->readerMutex, NULL);
qTrace("tsdb/read: %p, post-init read mutex: %p, code: %d", pReader, &pReader->readerMutex, code);
return code;
@ -324,22 +321,14 @@ static int32_t tsdbAcquireReader(STsdbReader* pReader) {
}
static int32_t tsdbTryAcquireReader(STsdbReader* pReader) {
int32_t code = -1;
qTrace("tsdb/read: %p, pre-trytake read mutex: %p, code: %d", pReader, &pReader->readerMutex, code);
code = taosThreadMutexTryLock(&pReader->readerMutex);
int32_t code = taosThreadMutexTryLock(&pReader->readerMutex);
qTrace("tsdb/read: %p, post-trytake read mutex: %p, code: %d", pReader, &pReader->readerMutex, code);
return code;
}
static int32_t tsdbReleaseReader(STsdbReader* pReader) {
int32_t code = -1;
qTrace("tsdb/read: %p, pre-untake read mutex: %p, code: %d", pReader, &pReader->readerMutex, code);
code = taosThreadMutexUnlock(&pReader->readerMutex);
int32_t code = taosThreadMutexUnlock(&pReader->readerMutex);
qTrace("tsdb/read: %p, post-untake read mutex: %p, code: %d", pReader, &pReader->readerMutex, code);
return code;
@ -432,6 +421,7 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, void
}
tsdbInitReaderLock(pReader);
tsem_init(&pReader->resumeAfterSuspend, 0, 0);
*ppReader = pReader;
return code;
@ -1015,8 +1005,8 @@ static int32_t copyBlockDataToSDataBlock(STsdbReader* pReader) {
// check if current block are all handled
if (pDumpInfo->rowIndex >= 0 && pDumpInfo->rowIndex < pRecord->numRow) {
int64_t ts = pBlockData->aTSKEY[pDumpInfo->rowIndex];
if (outOfTimeWindow(ts,
&pReader->info.window)) { // the remain data has out of query time window, ignore current block
if (outOfTimeWindow(ts, &pReader->info.window)) {
// the remain data has out of query time window, ignore current block
setBlockAllDumped(pDumpInfo, ts, pReader->info.order);
}
} else {
@ -1123,16 +1113,12 @@ static bool getNeighborBlockOfSameTable(SDataBlockIter* pBlockIter, SFileDataBlo
}
int32_t step = asc ? 1 : -1;
// *nextIndex = pBlockInfo->tbBlockIdx + step;
// *pBlockIndex = *(SBlockIndex*)taosArrayGet(pTableBlockScanInfo->pBlockList, *nextIndex);
STableDataBlockIdx* pTableDataBlockIdx =
taosArrayGet(pTableBlockScanInfo->pBlockIdxList, pBlockInfo->tbBlockIdx + step);
SFileDataBlockInfo* p = taosArrayGet(pBlockIter->blockList, pTableDataBlockIdx->globalIndex);
memcpy(pRecord, &p->record, sizeof(SBrinRecord));
*nextIndex = pBlockInfo->tbBlockIdx + step;
// tMapDataGetItemByIdx(&pTableBlockScanInfo->mapData, pIndex->ordinalIndex, pBlock, tGetDataBlk);
return true;
}
@ -1376,23 +1362,19 @@ static int32_t buildDataBlockFromBuf(STsdbReader* pReader, STableBlockScanInfo*
return TSDB_CODE_SUCCESS;
}
SSDataBlock* pBlock = pReader->resBlockInfo.pResBlock;
int64_t st = taosGetTimestampUs();
SSDataBlock* pBlock = pReader->resBlockInfo.pResBlock;
int32_t code = buildDataBlockFromBufImpl(pBlockScanInfo, endKey, pReader->resBlockInfo.capacity, pReader);
blockDataUpdateTsWindow(pBlock, pReader->suppInfo.slotId[0]);
pBlock->info.id.uid = pBlockScanInfo->uid;
double el = (taosGetTimestampUs() - st) / 1000.0;
updateComposedBlockInfo(pReader, el, pBlockScanInfo);
setComposedBlockFlag(pReader, true);
double elapsedTime = (taosGetTimestampUs() - st) / 1000.0;
tsdbDebug("%p build data block from cache completed, elapsed time:%.2f ms, numOfRows:%" PRId64 ", brange:%" PRId64
" - %" PRId64 ", uid:%" PRIu64 ", %s",
pReader, elapsedTime, pBlock->info.rows, pBlock->info.window.skey, pBlock->info.window.ekey,
pReader, el, pBlock->info.rows, pBlock->info.window.skey, pBlock->info.window.ekey,
pBlockScanInfo->uid, pReader->idStr);
pReader->cost.buildmemBlock += elapsedTime;
pReader->cost.buildmemBlock += el;
return code;
}
@ -2293,13 +2275,12 @@ static int32_t loadNeighborIfOverlap(SFileDataBlockInfo* pBlockInfo, STableBlock
return code;
}
static void updateComposedBlockInfo(STsdbReader* pReader, double el, STableBlockScanInfo* pBlockScanInfo) {
void updateComposedBlockInfo(STsdbReader* pReader, double el, STableBlockScanInfo* pBlockScanInfo) {
SSDataBlock* pResBlock = pReader->resBlockInfo.pResBlock;
pResBlock->info.id.uid = (pBlockScanInfo != NULL) ? pBlockScanInfo->uid : 0;
pResBlock->info.dataLoad = 1;
blockDataUpdateTsWindow(pResBlock, pReader->suppInfo.slotId[0]);
setComposedBlockFlag(pReader, true);
pReader->cost.composedBlocks += 1;
@ -2356,7 +2337,6 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
pBlockScanInfo = *pReader->status.pTableIter;
if (pReader->pIgnoreTables &&
taosHashGet(*pReader->pIgnoreTables, &pBlockScanInfo->uid, sizeof(pBlockScanInfo->uid))) {
// setBlockAllDumped(pDumpInfo, pBlock->maxKey.ts, pReader->info.order);
return code;
}
}
@ -2436,7 +2416,7 @@ int32_t getInitialDelIndex(const SArray* pDelSkyline, int32_t order) {
return ASCENDING_TRAVERSE(order) ? 0 : taosArrayGetSize(pDelSkyline) - 1;
}
int32_t initDelSkylineIterator(STableBlockScanInfo* pBlockScanInfo, int32_t order, SCostSummary* pCost) {
int32_t initDelSkylineIterator(STableBlockScanInfo* pBlockScanInfo, int32_t order, SReadCostSummary* pCost) {
int32_t code = 0;
int32_t newDelDataInFile = taosArrayGetSize(pBlockScanInfo->pFileDelData);
if (newDelDataInFile == 0 &&
@ -2935,6 +2915,8 @@ static int32_t buildBlockFromBufferSequentially(STsdbReader* pReader) {
SReaderStatus* pStatus = &pReader->status;
STableUidList* pUidList = &pStatus->uidList;
tsdbDebug("seq load data blocks from cache, %s", pReader->idStr);
while (1) {
if (pReader->code != TSDB_CODE_SUCCESS) {
tsdbWarn("tsdb reader is stopped ASAP, code:%s, %s", strerror(pReader->code), pReader->idStr);
@ -3043,6 +3025,8 @@ static ERetrieveType doReadDataFromLastFiles(STsdbReader* pReader) {
SSDataBlock* pResBlock = pReader->resBlockInfo.pResBlock;
SDataBlockIter* pBlockIter = &pReader->status.blockIter;
tsdbDebug("seq load data blocks from stt files %s", pReader->idStr);
while (1) {
terrno = 0;
@ -3774,7 +3758,6 @@ int32_t buildDataBlockFromBufImpl(STableBlockScanInfo* pBlockScanInfo, int64_t e
int32_t code = TSDB_CODE_SUCCESS;
do {
// SRow* pTSRow = NULL;
TSDBROW row = {.type = -1};
bool freeTSRow = false;
tsdbGetNextRowInMem(pBlockScanInfo, pReader, &row, endKey, &freeTSRow);
@ -3783,6 +3766,7 @@ int32_t buildDataBlockFromBufImpl(STableBlockScanInfo* pBlockScanInfo, int64_t e
}
if (row.type == TSDBROW_ROW_FMT) {
int64_t ts = row.pTSRow->ts;;
code = doAppendRowFromTSRow(pBlock, pReader, row.pTSRow, pBlockScanInfo);
if (freeTSRow) {
@ -3792,13 +3776,17 @@ int32_t buildDataBlockFromBufImpl(STableBlockScanInfo* pBlockScanInfo, int64_t e
if (code) {
return code;
}
pBlockScanInfo->lastProcKey = ts;
} else {
code = doAppendRowFromFileBlock(pBlock, pReader, row.pBlockData, row.iRow);
if (code) {
break;
}
pBlockScanInfo->lastProcKey = row.pBlockData->aTSKEY[row.iRow];
}
// no data in buffer, return immediately
if (!(pBlockScanInfo->iter.hasVal || pBlockScanInfo->iiter.hasVal)) {
break;
@ -4107,7 +4095,7 @@ void tsdbReaderClose2(STsdbReader* pReader) {
tsdbDataFileReaderClose(&pReader->pFileReader);
}
SCostSummary* pCost = &pReader->cost;
SReadCostSummary* pCost = &pReader->cost;
SFilesetIter* pFilesetIter = &pReader->status.fileIter;
if (pFilesetIter->pLastBlockReader != NULL) {
SLastBlockReader* pLReader = pFilesetIter->pLastBlockReader;
@ -4122,6 +4110,7 @@ void tsdbReaderClose2(STsdbReader* pReader) {
tsdbUntakeReadSnap2(pReader, pReader->pReadSnap, true);
pReader->pReadSnap = NULL;
tsem_destroy(&pReader->resumeAfterSuspend);
tsdbReleaseReader(pReader);
tsdbUninitReaderLock(pReader);
@ -4154,6 +4143,8 @@ int32_t tsdbReaderSuspend2(STsdbReader* pReader) {
SReaderStatus* pStatus = &pReader->status;
STableBlockScanInfo* pBlockScanInfo = NULL;
pReader->status.suspendInvoked = true; // record the suspend status
if (pStatus->loadFromFile) {
SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(&pReader->status.blockIter);
if (pBlockInfo != NULL) {
@ -4167,84 +4158,34 @@ int32_t tsdbReaderSuspend2(STsdbReader* pReader) {
tsdbDataFileReaderClose(&pReader->pFileReader);
SCostSummary* pCost = &pReader->cost;
SReadCostSummary* pCost = &pReader->cost;
pReader->status.pLDataIterArray = destroySttBlockReader(pReader->status.pLDataIterArray, &pCost->sttCost);
pReader->status.pLDataIterArray = taosArrayInit(4, POINTER_BYTES);
// resetDataBlockScanInfo excluding lastKey
STableBlockScanInfo** p = NULL;
int32_t iter = 0;
while ((p = tSimpleHashIterate(pStatus->pTableMap, p, &iter)) != NULL) {
STableBlockScanInfo* pInfo = *(STableBlockScanInfo**)p;
pInfo->iterInit = false;
pInfo->iter.hasVal = false;
pInfo->iiter.hasVal = false;
if (pInfo->iter.iter != NULL) {
pInfo->iter.iter = tsdbTbDataIterDestroy(pInfo->iter.iter);
}
if (pInfo->iiter.iter != NULL) {
pInfo->iiter.iter = tsdbTbDataIterDestroy(pInfo->iiter.iter);
}
pInfo->delSkyline = taosArrayDestroy(pInfo->delSkyline);
pInfo->pFileDelData = taosArrayDestroy(pInfo->pFileDelData);
}
} else {
// resetDataBlockScanInfo excluding lastKey
STableBlockScanInfo** p = NULL;
int32_t iter = 0;
while ((p = tSimpleHashIterate(pStatus->pTableMap, p, &iter)) != NULL) {
STableBlockScanInfo* pInfo = *(STableBlockScanInfo**)p;
pInfo->iterInit = false;
pInfo->iter.hasVal = false;
pInfo->iiter.hasVal = false;
if (pInfo->iter.iter != NULL) {
pInfo->iter.iter = tsdbTbDataIterDestroy(pInfo->iter.iter);
}
if (pInfo->iiter.iter != NULL) {
pInfo->iiter.iter = tsdbTbDataIterDestroy(pInfo->iiter.iter);
}
pInfo->delSkyline = taosArrayDestroy(pInfo->delSkyline);
}
pBlockScanInfo = pStatus->pTableIter == NULL ? NULL : *pStatus->pTableIter;
if (pBlockScanInfo) {
// save lastKey to restore memory iterator
STimeWindow w = pReader->resBlockInfo.pResBlock->info.window;
pBlockScanInfo->lastProcKey = ASCENDING_TRAVERSE(pReader->info.order) ? w.ekey : w.skey;
// reset current current table's data block scan info,
pBlockScanInfo->iterInit = false;
pBlockScanInfo->iter.hasVal = false;
pBlockScanInfo->iiter.hasVal = false;
if (pBlockScanInfo->iter.iter != NULL) {
pBlockScanInfo->iter.iter = tsdbTbDataIterDestroy(pBlockScanInfo->iter.iter);
}
if (pBlockScanInfo->iiter.iter != NULL) {
pBlockScanInfo->iiter.iter = tsdbTbDataIterDestroy(pBlockScanInfo->iiter.iter);
}
pBlockScanInfo->pBlockList = taosArrayDestroy(pBlockScanInfo->pBlockList);
pBlockScanInfo->pBlockIdxList = taosArrayDestroy(pBlockScanInfo->pBlockIdxList);
// TODO: keep skyline for reuse
pBlockScanInfo->delSkyline = taosArrayDestroy(pBlockScanInfo->delSkyline);
}
}
// resetDataBlockScanInfo excluding lastKey
STableBlockScanInfo** p = NULL;
int32_t step = ASCENDING_TRAVERSE(pReader->info.order)? 1:-1;
int32_t iter = 0;
while ((p = tSimpleHashIterate(pStatus->pTableMap, p, &iter)) != NULL) {
STableBlockScanInfo* pInfo = *(STableBlockScanInfo**)p;
clearBlockScanInfo(pInfo);
pInfo->sttKeyInfo.nextProcKey = pInfo->lastProcKey + step;
}
pStatus->uidList.currentIndex = 0;
initReaderStatus(pStatus);
tsdbUntakeReadSnap2(pReader, pReader->pReadSnap, false);
pReader->pReadSnap = NULL;
pReader->flag = READER_STATUS_SUSPEND;
#if SUSPEND_RESUME_TEST
tsem_post(&pReader->resumeAfterSuspend);
#endif
tsdbDebug("reader: %p suspended uid %" PRIu64 " in this query %s", pReader, pBlockScanInfo ? pBlockScanInfo->uid : 0,
pReader->idStr);
return code;
@ -4399,6 +4340,16 @@ int32_t tsdbNextDataBlock2(STsdbReader* pReader, bool* hasNext) {
SReaderStatus* pStatus = &pReader->status;
// NOTE: the following codes is used to perform test for suspend/resume for tsdbReader when it blocks the commit
// the data should be ingested in round-robin and all the child tables should be createted before ingesting data
// the version range of query will be used to identify the correctness of suspend/resume functions.
// this function will blocked before loading the SECOND block from vnode-buffer, and restart itself from sst-files
#if SUSPEND_RESUME_TEST
if (!pReader->status.suspendInvoked && !pReader->status.loadFromFile) {
tsem_wait(&pReader->resumeAfterSuspend);
}
#endif
code = tsdbAcquireReader(pReader);
qTrace("tsdb/read: %p, take read mutex, code: %d", pReader, code);

1
source/dnode/vnode/src/tsdb/tsdbReadUtil.c
View File

@ -210,6 +210,7 @@ void clearBlockScanInfo(STableBlockScanInfo* p) {
p->iterInit = false;
p->iter.hasVal = false;
p->iiter.hasVal = false;
p->sttKeyInfo.status = STT_FILE_READER_UNINIT;
if (p->iter.iter != NULL) {
p->iter.iter = tsdbTbDataIterDestroy(p->iter.iter);

14
source/dnode/vnode/src/tsdb/tsdbReadUtil.h
View File

@ -96,7 +96,7 @@ typedef struct SResultBlockInfo {
int64_t capacity;
} SResultBlockInfo;
typedef struct SCostSummary {
typedef struct SReadCostSummary {
int64_t numOfBlocks;
double blockLoadTime;
double buildmemBlock;
@ -110,7 +110,7 @@ typedef struct SCostSummary {
double createScanInfoList;
double createSkylineIterTime;
double initLastBlockReader;
} SCostSummary;
} SReadCostSummary;
typedef struct STableUidList {
uint64_t* tableUidList; // access table uid list in uid ascending order list
@ -122,12 +122,6 @@ typedef struct {
int32_t numOfSttFiles;
} SBlockNumber;
typedef struct SBlockIndex {
int32_t ordinalIndex;
int64_t inFileOffset;
STimeWindow window; // todo replace it with overlap flag.
} SBlockIndex;
typedef struct SBlockOrderWrapper {
int64_t uid;
int64_t offset;
@ -192,6 +186,7 @@ typedef struct SFileBlockDumpInfo {
} SFileBlockDumpInfo;
typedef struct SReaderStatus {
bool suspendInvoked;
bool loadFromFile; // check file stage
bool composedDataBlock; // the returned data block is a composed block or not
SSHashObj* pTableMap; // SHash<STableBlockScanInfo>
@ -220,7 +215,8 @@ struct STsdbReader {
int32_t type; // query type: 1. retrieve all data blocks, 2. retrieve direct prev|next rows
SBlockLoadSuppInfo suppInfo;
STsdbReadSnap* pReadSnap;
SCostSummary cost;
tsem_t resumeAfterSuspend;
SReadCostSummary cost;
SHashObj** pIgnoreTables;
SSHashObj* pSchemaMap; // keep the retrieved schema info, to avoid the overhead by repeatly load schema
SDataFileReader* pFileReader; // the file reader

11
source/libs/executor/src/streameventwindowoperator.c
View File

@ -610,6 +610,13 @@ void streamEventReloadState(SOperatorInfo* pOperator) {
compactEventWindow(pOperator, &curInfo, pInfo->pSeUpdated, pInfo->pSeDeleted, false);
qDebug("===stream=== reload state. save result %" PRId64 ", %" PRIu64, curInfo.winInfo.sessionWin.win.skey,
curInfo.winInfo.sessionWin.groupId);
if (IS_VALID_SESSION_WIN(curInfo.winInfo)) {
saveSessionOutputBuf(pAggSup, &curInfo.winInfo);
}
if (!curInfo.pWinFlag->endFlag) {
continue;
}
if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE) {
saveResult(curInfo.winInfo, pInfo->pSeUpdated);
@ -621,10 +628,6 @@ void streamEventReloadState(SOperatorInfo* pOperator) {
getSessionHashKey(&curInfo.winInfo.sessionWin, &key);
tSimpleHashPut(pAggSup->pResultRows, &key, sizeof(SSessionKey), &curInfo.winInfo, sizeof(SResultWindowInfo));
}
if (IS_VALID_SESSION_WIN(curInfo.winInfo)) {
saveSessionOutputBuf(pAggSup, &curInfo.winInfo);
}
}
taosMemoryFree(pBuf);

3274
source/libs/nodes/src/nodesCodeFuncs.c
View File

File diff suppressed because it is too large Load Diff

1239
source/libs/nodes/src/nodesUtilFuncs.c
View File

File diff suppressed because it is too large Load Diff

2
source/libs/parser/inc/parInsertUtil.h
View File

@ -50,7 +50,7 @@ void insCheckTableDataOrder(STableDataCxt *pTableCxt, TSKEY tsKey);
int32_t insGetTableDataCxt(SHashObj *pHash, void *id, int32_t idLen, STableMeta *pTableMeta,
SVCreateTbReq **pCreateTbReq, STableDataCxt **pTableCxt, bool colMode, bool ignoreColVals);
int32_t initTableColSubmitData(STableDataCxt *pTableCxt);
int32_t insMergeTableDataCxt(SHashObj *pTableHash, SArray **pVgDataBlocks);
int32_t insMergeTableDataCxt(SHashObj *pTableHash, SArray **pVgDataBlocks, bool isRebuild);
int32_t insBuildVgDataBlocks(SHashObj *pVgroupsHashObj, SArray *pVgDataBlocks, SArray **pDataBlocks);
void insDestroyTableDataCxtHashMap(SHashObj *pTableCxtHash);
void insDestroyVgroupDataCxt(SVgroupDataCxt *pVgCxt);

2
source/libs/parser/src/parInsertSml.c
View File

@ -425,7 +425,7 @@ SQuery* smlInitHandle() {
int32_t smlBuildOutput(SQuery* handle, SHashObj* pVgHash) {
SVnodeModifyOpStmt* pStmt = (SVnodeModifyOpStmt*)(handle)->pRoot;
// merge according to vgId
int32_t code = insMergeTableDataCxt(pStmt->pTableBlockHashObj, &pStmt->pVgDataBlocks);
int32_t code = insMergeTableDataCxt(pStmt->pTableBlockHashObj, &pStmt->pVgDataBlocks, true);
if (code != TSDB_CODE_SUCCESS) {
uError("insMergeTableDataCxt failed");
return code;

84
source/libs/parser/src/parInsertSql.c
View File

@ -55,6 +55,7 @@ typedef struct SInsertParseContext {
bool usingDuplicateTable;
bool forceUpdate;
bool needTableTagVal;
bool needRequest; // whether or not request server
} SInsertParseContext;
typedef int32_t (*_row_append_fn_t)(SMsgBuf* pMsgBuf, const void* value, int32_t len, void* param);
@ -652,6 +653,10 @@ static int32_t parseTagValue(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStm
}
static int32_t buildCreateTbReq(SVnodeModifyOpStmt* pStmt, STag* pTag, SArray* pTagName) {
if (pStmt->pCreateTblReq) {
tdDestroySVCreateTbReq(pStmt->pCreateTblReq);
taosMemoryFreeClear(pStmt->pCreateTblReq);
}
pStmt->pCreateTblReq = taosMemoryCalloc(1, sizeof(SVCreateTbReq));
if (NULL == pStmt->pCreateTblReq) {
return TSDB_CODE_OUT_OF_MEMORY;
@ -1797,9 +1802,10 @@ static void clearStbRowsDataContext(SStbRowsDataContext* pStbRowsCxt) {
taosMemoryFreeClear(pStbRowsCxt->pCreateCtbReq);
}
static int32_t parseOneStbRow(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt, const char** ppSql,
SStbRowsDataContext* pStbRowsCxt, bool* pGotRow, SToken* pToken) {
bool bFirstTable = false;
static int32_t parseOneStbRow(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt, const char** ppSql,
SStbRowsDataContext* pStbRowsCxt, bool* pGotRow, SToken* pToken,
STableDataCxt** ppTableDataCxt) {
bool bFirstTable = false;
int32_t code = getStbRowValues(pCxt, pStmt, ppSql, pStbRowsCxt, pGotRow, pToken, &bFirstTable);
if (code != TSDB_CODE_SUCCESS || !*pGotRow) {
return code;
@ -1809,15 +1815,14 @@ static int32_t parseOneStbRow(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pSt
code = processCtbAutoCreationAndCtbMeta(pCxt, pStmt, pStbRowsCxt);
}
STableDataCxt* pTableDataCxt = NULL;
code = insGetTableDataCxt(pStmt->pTableBlockHashObj, &pStbRowsCxt->pCtbMeta->uid, sizeof(pStbRowsCxt->pCtbMeta->uid),
pStbRowsCxt->pCtbMeta, &pStbRowsCxt->pCreateCtbReq, &pTableDataCxt, false, true);
initTableColSubmitData(pTableDataCxt);
pStbRowsCxt->pCtbMeta, &pStbRowsCxt->pCreateCtbReq, ppTableDataCxt, false, true);
initTableColSubmitData(*ppTableDataCxt);
if (code == TSDB_CODE_SUCCESS) {
SRow** pRow = taosArrayReserve(pTableDataCxt->pData->aRowP, 1);
code = tRowBuild(pStbRowsCxt->aColVals, pTableDataCxt->pSchema, pRow);
SRow** pRow = taosArrayReserve((*ppTableDataCxt)->pData->aRowP, 1);
code = tRowBuild(pStbRowsCxt->aColVals, (*ppTableDataCxt)->pSchema, pRow);
if (TSDB_CODE_SUCCESS == code) {
insCheckTableDataOrder(pTableDataCxt, TD_ROW_KEY(*pRow));
insCheckTableDataOrder(*ppTableDataCxt, TD_ROW_KEY(*pRow));
}
}
@ -1915,7 +1920,8 @@ static int32_t parseValues(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt,
if (!pStmt->stbSyntax) {
code = parseOneRow(pCxt, &pStmt->pSql, rowsDataCxt.pTableDataCxt, &gotRow, pToken);
} else {
code = parseOneStbRow(pCxt, pStmt, &pStmt->pSql, rowsDataCxt.pStbRowsCxt, &gotRow, pToken);
STableDataCxt* pTableDataCxt = NULL;
code = parseOneStbRow(pCxt, pStmt, &pStmt->pSql, rowsDataCxt.pStbRowsCxt, &gotRow, pToken, &pTableDataCxt);
}
}
@ -1979,7 +1985,14 @@ static int32_t parseCsvFile(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt
if (!pStmt->stbSyntax) {
code = parseOneRow(pCxt, (const char**)&pRow, rowsDataCxt.pTableDataCxt, &gotRow, &token);
} else {
code = parseOneStbRow(pCxt, pStmt, (const char**)&pRow, rowsDataCxt.pStbRowsCxt, &gotRow, &token);
STableDataCxt* pTableDataCxt = NULL;
code = parseOneStbRow(pCxt, pStmt, (const char**)&pRow, rowsDataCxt.pStbRowsCxt, &gotRow, &token, &pTableDataCxt);
if (code == TSDB_CODE_SUCCESS) {
SStbRowsDataContext* pStbRowsCxt = rowsDataCxt.pStbRowsCxt;
void* pData = pTableDataCxt;
taosHashPut(pStmt->pTableCxtHashObj, &pStbRowsCxt->pCtbMeta->uid, sizeof(pStbRowsCxt->pCtbMeta->uid), &pData,
POINTER_BYTES);
}
}
if (code && firstLine) {
firstLine = false;
@ -1992,7 +2005,7 @@ static int32_t parseCsvFile(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt
(*pNumOfRows)++;
}
if (TSDB_CODE_SUCCESS == code && (*pNumOfRows) > tsMaxInsertBatchRows) {
if (TSDB_CODE_SUCCESS == code && (*pNumOfRows) >= tsMaxInsertBatchRows) {
pStmt->fileProcessing = true;
break;
}
@ -2003,7 +2016,7 @@ static int32_t parseCsvFile(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt
parserDebug("0x%" PRIx64 " %d rows have been parsed", pCxt->pComCxt->requestId, *pNumOfRows);
if (TSDB_CODE_SUCCESS == code && 0 == (*pNumOfRows) &&
if (TSDB_CODE_SUCCESS == code && 0 == (*pNumOfRows) && 0 == pStmt->totalRowsNum &&
(!TSDB_QUERY_HAS_TYPE(pStmt->insertType, TSDB_QUERY_TYPE_STMT_INSERT)) && !pStmt->fileProcessing) {
code = buildSyntaxErrMsg(&pCxt->msg, "no any data points", NULL);
}
@ -2011,6 +2024,12 @@ static int32_t parseCsvFile(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt
}
static int32_t parseDataFromFileImpl(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt, SRowsDataContext rowsDataCxt) {
// init only for file
if (NULL == pStmt->pTableCxtHashObj) {
pStmt->pTableCxtHashObj =
taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
}
int32_t numOfRows = 0;
int32_t code = parseCsvFile(pCxt, pStmt, rowsDataCxt, &numOfRows);
if (TSDB_CODE_SUCCESS == code) {
@ -2022,7 +2041,18 @@ static int32_t parseDataFromFileImpl(SInsertParseContext* pCxt, SVnodeModifyOpSt
} else {
parserDebug("0x%" PRIx64 " insert from csv. File is too large, do it in batches.", pCxt->pComCxt->requestId);
}
if (pStmt->insertType != TSDB_QUERY_TYPE_FILE_INSERT) {
return buildSyntaxErrMsg(&pCxt->msg, "keyword VALUES or FILE is exclusive", NULL);
}
}
// just record pTableCxt whose data come from file
if (!pStmt->stbSyntax && numOfRows > 0) {
void* pData = rowsDataCxt.pTableDataCxt;
taosHashPut(pStmt->pTableCxtHashObj, &pStmt->pTableMeta->uid, sizeof(pStmt->pTableMeta->uid), &pData,
POINTER_BYTES);
}
return code;
}
@ -2061,6 +2091,9 @@ static int32_t parseDataClause(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pS
NEXT_TOKEN(pStmt->pSql, token);
switch (token.type) {
case TK_VALUES:
if (TSDB_QUERY_HAS_TYPE(pStmt->insertType, TSDB_QUERY_TYPE_FILE_INSERT)) {
return buildSyntaxErrMsg(&pCxt->msg, "keyword VALUES or FILE is exclusive", token.z);
}
return parseValuesClause(pCxt, pStmt, rowsDataCxt, &token);
case TK_FILE:
return parseFileClause(pCxt, pStmt, rowsDataCxt, &token);
@ -2275,8 +2308,25 @@ static int32_t parseInsertBodyBottom(SInsertParseContext* pCxt, SVnodeModifyOpSt
return setStmtInfo(pCxt, pStmt);
}
// release old array alloced by merge
pStmt->freeArrayFunc(pStmt->pVgDataBlocks);
pStmt->pVgDataBlocks = NULL;
bool fileOnly = (pStmt->insertType == TSDB_QUERY_TYPE_FILE_INSERT);
if (fileOnly) {
// none data, skip merge & buildvgdata
if (0 == taosHashGetSize(pStmt->pTableCxtHashObj)) {
pCxt->needRequest = false;
return TSDB_CODE_SUCCESS;
}
}
// merge according to vgId
int32_t code = insMergeTableDataCxt(pStmt->pTableBlockHashObj, &pStmt->pVgDataBlocks);
int32_t code = insMergeTableDataCxt(fileOnly ? pStmt->pTableCxtHashObj : pStmt->pTableBlockHashObj,
&pStmt->pVgDataBlocks, pStmt->fileProcessing);
// clear tmp hashobj only
taosHashClear(pStmt->pTableCxtHashObj);
if (TSDB_CODE_SUCCESS == code) {
code = insBuildVgDataBlocks(pStmt->pVgroupsHashObj, pStmt->pVgDataBlocks, &pStmt->pDataBlocks);
}
@ -2718,6 +2768,7 @@ int32_t parseInsertSql(SParseContext* pCxt, SQuery** pQuery, SCatalogReq* pCatal
.msg = {.buf = pCxt->pMsg, .len = pCxt->msgLen},
.missCache = false,
.usingDuplicateTable = false,
.needRequest = true,
.forceUpdate = (NULL != pCatalogReq ? pCatalogReq->forceUpdate : false)};
int32_t code = initInsertQuery(&context, pCatalogReq, pMetaData, pQuery);
@ -2732,5 +2783,10 @@ int32_t parseInsertSql(SParseContext* pCxt, SQuery** pQuery, SCatalogReq* pCatal
code = setRefreshMeta(*pQuery);
}
insDestroyBoundColInfo(&context.tags);
// if no data to insert, set emptyMode to avoid request server
if (!context.needRequest) {
(*pQuery)->execMode = QUERY_EXEC_MODE_EMPTY_RESULT;
}
return code;
}

2
source/libs/parser/src/parInsertStmt.c
View File

@ -58,7 +58,7 @@ int32_t qBuildStmtOutput(SQuery* pQuery, SHashObj* pVgHash, SHashObj* pBlockHash
// merge according to vgId
if (taosHashGetSize(pBlockHash) > 0) {
code = insMergeTableDataCxt(pBlockHash, &pVgDataBlocks);
code = insMergeTableDataCxt(pBlockHash, &pVgDataBlocks, true);
}
if (TSDB_CODE_SUCCESS == code) {
code = insBuildVgDataBlocks(pVgHash, pVgDataBlocks, &pStmt->pDataBlocks);

29
source/libs/parser/src/parInsertUtil.c
View File

@ -289,6 +289,14 @@ static int32_t rebuildTableData(SSubmitTbData* pSrc, SSubmitTbData** pDst) {
pTmp->uid = pSrc->uid;
pTmp->sver = pSrc->sver;
pTmp->pCreateTbReq = NULL;
if (pTmp->flags & SUBMIT_REQ_AUTO_CREATE_TABLE) {
if (pSrc->pCreateTbReq) {
cloneSVreateTbReq(pSrc->pCreateTbReq, &pTmp->pCreateTbReq);
} else {
pTmp->flags &= ~SUBMIT_REQ_AUTO_CREATE_TABLE;
}
}
if (pTmp->flags & SUBMIT_REQ_COLUMN_DATA_FORMAT) {
pTmp->aCol = taosArrayInit(128, sizeof(SColData));
if (NULL == pTmp->aCol) {
@ -416,15 +424,21 @@ void insDestroyTableDataCxtHashMap(SHashObj* pTableCxtHash) {
taosHashCleanup(pTableCxtHash);
}
static int32_t fillVgroupDataCxt(STableDataCxt* pTableCxt, SVgroupDataCxt* pVgCxt) {
static int32_t fillVgroupDataCxt(STableDataCxt* pTableCxt, SVgroupDataCxt* pVgCxt, bool isRebuild) {
if (NULL == pVgCxt->pData->aSubmitTbData) {
pVgCxt->pData->aSubmitTbData = taosArrayInit(128, sizeof(SSubmitTbData));
if (NULL == pVgCxt->pData->aSubmitTbData) {
return TSDB_CODE_OUT_OF_MEMORY;
}
}
// push data to submit, rebuild empty data for next submit
taosArrayPush(pVgCxt->pData->aSubmitTbData, pTableCxt->pData);
rebuildTableData(pTableCxt->pData, &pTableCxt->pData);
if (isRebuild) {
rebuildTableData(pTableCxt->pData, &pTableCxt->pData);
} else {
taosMemoryFreeClear(pTableCxt->pData);
}
qDebug("add tableDataCxt uid:%" PRId64 " to vgId:%d", pTableCxt->pMeta->uid, pVgCxt->vgId);
@ -467,7 +481,7 @@ int insColDataComp(const void* lp, const void* rp) {
return 0;
}
int32_t insMergeTableDataCxt(SHashObj* pTableHash, SArray** pVgDataBlocks) {
int32_t insMergeTableDataCxt(SHashObj* pTableHash, SArray** pVgDataBlocks, bool isRebuild) {
SHashObj* pVgroupHash = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, false);
SArray* pVgroupList = taosArrayInit(8, POINTER_BYTES);
if (NULL == pVgroupHash || NULL == pVgroupList) {
@ -502,6 +516,13 @@ int32_t insMergeTableDataCxt(SHashObj* pTableHash, SArray** pVgDataBlocks) {
tColDataSortMerge(pTableCxt->pData->aCol);
} else {
// skip the table has no data to insert
// eg: import a csv without valid data
// if (0 == taosArrayGetSize(pTableCxt->pData->aRowP)) {
// qWarn("no row in tableDataCxt uid:%" PRId64 " ", pTableCxt->pMeta->uid);
// p = taosHashIterate(pTableHash, p);
// continue;
// }
if (!pTableCxt->ordered) {
code = tRowSort(pTableCxt->pData->aRowP);
}
@ -520,7 +541,7 @@ int32_t insMergeTableDataCxt(SHashObj* pTableHash, SArray** pVgDataBlocks) {
pVgCxt = *(SVgroupDataCxt**)pp;
}
if (TSDB_CODE_SUCCESS == code) {
code = fillVgroupDataCxt(pTableCxt, pVgCxt);
code = fillVgroupDataCxt(pTableCxt, pVgCxt, isRebuild);
}
}
if (TSDB_CODE_SUCCESS == code) {

24
source/libs/stream/src/streamExec.c
View File

@ -252,14 +252,15 @@ static void streamScanHistoryDataImpl(SStreamTask* pTask, SArray* pRes, int32_t*
SScanhistoryDataInfo streamScanHistoryData(SStreamTask* pTask, int64_t st) {
ASSERT(pTask->info.taskLevel == TASK_LEVEL__SOURCE);
void* exec = pTask->exec.pExecutor;
bool finished = false;
void* exec = pTask->exec.pExecutor;
bool finished = false;
const char* id = pTask->id.idStr;
qSetStreamOpOpen(exec);
while (1) {
if (streamTaskShouldPause(pTask)) {
stDebug("s-task:%s paused from the scan-history task", pTask->id.idStr);
stDebug("s-task:%s paused from the scan-history task", id);
// quit from step1, not continue to handle the step2
return (SScanhistoryDataInfo){TASK_SCANHISTORY_QUIT, 0};
}
@ -267,8 +268,7 @@ SScanhistoryDataInfo streamScanHistoryData(SStreamTask* pTask, int64_t st) {
SArray* pRes = taosArrayInit(0, sizeof(SSDataBlock));
if (pRes == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
stError("s-task:%s scan-history prepare result block failed, code:%s, retry later", pTask->id.idStr,
tstrerror(terrno));
stError("s-task:%s scan-history prepare result block failed, code:%s, retry later", id, tstrerror(terrno));
continue;
}
@ -281,12 +281,12 @@ SScanhistoryDataInfo streamScanHistoryData(SStreamTask* pTask, int64_t st) {
}
// dispatch the generated results
int32_t code = handleResultBlocks(pTask, pRes, size);
/*int32_t code = */handleResultBlocks(pTask, pRes, size);
int64_t el = taosGetTimestampMs() - st;
// downstream task input queue is full, try in 5sec
if (pTask->inputq.status == TASK_INPUT_STATUS__BLOCKED) {
if (pTask->inputq.status == TASK_INPUT_STATUS__BLOCKED && (pTask->info.fillHistory == 1)) {
return (SScanhistoryDataInfo){TASK_SCANHISTORY_REXEC, 5000};
}
@ -294,9 +294,9 @@ SScanhistoryDataInfo streamScanHistoryData(SStreamTask* pTask, int64_t st) {
return (SScanhistoryDataInfo){TASK_SCANHISTORY_CONT, 0};
}
if (el >= STREAM_SCAN_HISTORY_TIMESLICE) {
stDebug("s-task:%s fill-history:%d time slice exhausted, elapsed time:%.2fs, retry in 100ms",
pTask->id.idStr, pTask->info.fillHistory, el / 1000.0);
if (el >= STREAM_SCAN_HISTORY_TIMESLICE && (pTask->info.fillHistory == 1)) {
stDebug("s-task:%s fill-history:%d time slice exhausted, elapsed time:%.2fs, retry in 100ms", id,
pTask->info.fillHistory, el / 1000.0);
return (SScanhistoryDataInfo){TASK_SCANHISTORY_REXEC, 100};
}
}
@ -543,7 +543,7 @@ int32_t streamProcessTranstateBlock(SStreamTask* pTask, SStreamDataBlock* pBlock
* todo: the batch of blocks should be tuned dynamic, according to the total elapsed time of each batch of blocks, the
* appropriate batch of blocks should be handled in 5 to 10 sec.
*/
int32_t streamExecForAll(SStreamTask* pTask) {
int32_t doStreamExecTask(SStreamTask* pTask) {
const char* id = pTask->id.idStr;
// merge multiple input data if possible in the input queue.
@ -654,7 +654,7 @@ int32_t streamExecTask(SStreamTask* pTask) {
int8_t schedStatus = streamTaskSetSchedStatusActive(pTask);
if (schedStatus == TASK_SCHED_STATUS__WAITING) {
while (1) {
int32_t code = streamExecForAll(pTask);
int32_t code = doStreamExecTask(pTask);
if (code < 0) { // todo this status should be removed
atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__FAILED);
return -1;

13
source/libs/stream/src/streamStart.c
View File

@ -471,6 +471,14 @@ int32_t streamProcessCheckRsp(SStreamTask* pTask, const SStreamTaskCheckRsp* pRs
streamMetaUpdateTaskDownstreamStatus(pTask, pTask->execInfo.init, taosGetTimestampMs(), false);
// automatically set the related fill-history task to be failed.
if (HAS_RELATED_FILLHISTORY_TASK(pTask)) {
STaskId* pId = &pTask->hTaskInfo.id;
SStreamTask* pHTask = streamMetaAcquireTask(pTask->pMeta, pId->streamId, pId->taskId);
streamMetaUpdateTaskDownstreamStatus(pHTask, pHTask->execInfo.init, taosGetTimestampMs(), false);
streamMetaReleaseTask(pTask->pMeta, pHTask);
}
} else { // TASK_DOWNSTREAM_NOT_READY, let's retry in 100ms
STaskRecheckInfo* pInfo = createRecheckInfo(pTask, pRsp);
@ -1072,8 +1080,9 @@ int32_t streamMetaUpdateTaskDownstreamStatus(SStreamTask* pTask, int64_t startTs
taosHashPut(pDst, &id, sizeof(id), &initTs, sizeof(STaskInitTs));
int32_t numOfTotal = streamMetaGetNumOfTasks(pMeta);
int32_t numOfRecv = taosHashGetSize(pStartInfo->pReadyTaskSet) + taosHashGetSize(pStartInfo->pFailedTaskSet);
if (taosHashGetSize(pStartInfo->pReadyTaskSet) + taosHashGetSize(pStartInfo->pFailedTaskSet) == numOfTotal) {
if (numOfRecv == numOfTotal) {
pStartInfo->readyTs = taosGetTimestampMs();
pStartInfo->elapsedTime = (pStartInfo->startTs != 0) ? pStartInfo->readyTs - pStartInfo->startTs : 0;
@ -1087,6 +1096,8 @@ int32_t streamMetaUpdateTaskDownstreamStatus(SStreamTask* pTask, int64_t startTs
displayStatusInfo(pMeta, pStartInfo->pFailedTaskSet, false);
streamMetaResetStartInfo(pStartInfo);
} else {
stDebug("vgId:%d recv check down results:%d, total:%d", pMeta->vgId, numOfRecv, numOfTotal);
}
streamMetaWUnLock(pMeta);

4
source/libs/transport/src/transCli.c
View File

@ -957,8 +957,8 @@ static void cliSendCb(uv_write_t* req, int status) {
SCliMsg* pMsg = !transQueueEmpty(&pConn->cliMsgs) ? transQueueGet(&pConn->cliMsgs, 0) : NULL;
if (pMsg != NULL) {
int64_t cost = taosGetTimestampUs() - pMsg->st;
if (cost > 1000 * 20) {
tWarn("%s conn %p send cost:%dus, send exception", CONN_GET_INST_LABEL(pConn), pConn, (int)cost);
if (cost > 1000 * 50) {
tTrace("%s conn %p send cost:%dus ", CONN_GET_INST_LABEL(pConn), pConn, (int)cost);
}
}

105
source/util/src/tcompression.c
View File

@ -538,66 +538,71 @@ int32_t tsDecompressTimestampImp(const char *const input, const int32_t nelement
memcpy(output, input + 1, nelements * longBytes);
return nelements * longBytes;
} else if (input[0] == 1) { // Decompress
int64_t *ostream = (int64_t *)output;
if (tsSIMDEnable && tsAVX512Enable) {
tsDecompressTimestampAvx512(input, nelements, output, false);
} else if (tsSIMDEnable && tsAVX2Enable) {
tsDecompressTimestampAvx2(input, nelements, output, false);
} else {
int64_t *ostream = (int64_t *)output;
int32_t ipos = 1, opos = 0;
int8_t nbytes = 0;
int64_t prev_value = 0;
int64_t prev_delta = 0;
int64_t delta_of_delta = 0;
int32_t ipos = 1, opos = 0;
int8_t nbytes = 0;
int64_t prev_value = 0;
int64_t prev_delta = 0;
int64_t delta_of_delta = 0;
while (1) {
uint8_t flags = input[ipos++];
// Decode dd1
uint64_t dd1 = 0;
nbytes = flags & INT8MASK(4);
if (nbytes == 0) {
delta_of_delta = 0;
} else {
if (is_bigendian()) {
memcpy(((char *)(&dd1)) + longBytes - nbytes, input + ipos, nbytes);
while (1) {
uint8_t flags = input[ipos++];
// Decode dd1
uint64_t dd1 = 0;
nbytes = flags & INT8MASK(4);
if (nbytes == 0) {
delta_of_delta = 0;
} else {
memcpy(&dd1, input + ipos, nbytes);
if (is_bigendian()) {
memcpy(((char *)(&dd1)) + longBytes - nbytes, input + ipos, nbytes);
} else {
memcpy(&dd1, input + ipos, nbytes);
}
delta_of_delta = ZIGZAG_DECODE(int64_t, dd1);
}
delta_of_delta = ZIGZAG_DECODE(int64_t, dd1);
}
ipos += nbytes;
if (opos == 0) {
prev_value = delta_of_delta;
prev_delta = 0;
ostream[opos++] = delta_of_delta;
} else {
ipos += nbytes;
if (opos == 0) {
prev_value = delta_of_delta;
prev_delta = 0;
ostream[opos++] = delta_of_delta;
} else {
prev_delta = delta_of_delta + prev_delta;
prev_value = prev_value + prev_delta;
ostream[opos++] = prev_value;
}
if (opos == nelements) return nelements * longBytes;
// Decode dd2
uint64_t dd2 = 0;
nbytes = (flags >> 4) & INT8MASK(4);
if (nbytes == 0) {
delta_of_delta = 0;
} else {
if (is_bigendian()) {
memcpy(((char *)(&dd2)) + longBytes - nbytes, input + ipos, nbytes);
} else {
memcpy(&dd2, input + ipos, nbytes);
}
// zigzag_decoding
delta_of_delta = ZIGZAG_DECODE(int64_t, dd2);
}
ipos += nbytes;
prev_delta = delta_of_delta + prev_delta;
prev_value = prev_value + prev_delta;
ostream[opos++] = prev_value;
if (opos == nelements) return nelements * longBytes;
}
if (opos == nelements) return nelements * longBytes;
// Decode dd2
uint64_t dd2 = 0;
nbytes = (flags >> 4) & INT8MASK(4);
if (nbytes == 0) {
delta_of_delta = 0;
} else {
if (is_bigendian()) {
memcpy(((char *)(&dd2)) + longBytes - nbytes, input + ipos, nbytes);
} else {
memcpy(&dd2, input + ipos, nbytes);
}
// zigzag_decoding
delta_of_delta = ZIGZAG_DECODE(int64_t, dd2);
}
ipos += nbytes;
prev_delta = delta_of_delta + prev_delta;
prev_value = prev_value + prev_delta;
ostream[opos++] = prev_value;
if (opos == nelements) return nelements * longBytes;
}
} else {
ASSERT(0);
return -1;
}
return nelements * longBytes;
}
/* --------------------------------------------Double Compression ---------------------------------------------- */

300
source/util/src/tdecompress.c
View File

@ -53,11 +53,8 @@ int32_t tsDecompressIntImpl_Hw(const char *const input, const int32_t nelements,
int64_t prevValue = 0;
#if __AVX2__
while (1) {
if (_pos == nelements) break;
uint64_t w = 0;
memcpy(&w, ip, LONG_BYTES);
while (_pos < nelements) {
uint64_t w = *(uint64_t*) ip;
char selector = (char)(w & INT64MASK(4)); // selector = 4
char bit = bit_per_integer[(int32_t)selector]; // bit = 3
@ -114,7 +111,7 @@ int32_t tsDecompressIntImpl_Hw(const char *const input, const int32_t nelements,
__m256i signmask = _mm256_and_si256(_mm256_set1_epi64x(1), zigzagVal);
signmask = _mm256_sub_epi64(_mm256_setzero_si256(), signmask);
// get the four zigzag values here
// get four zigzag values here
__m256i delta = _mm256_xor_si256(_mm256_srli_epi64(zigzagVal, 1), signmask);
// calculate the cumulative sum (prefix sum) for each number
@ -250,73 +247,264 @@ int32_t tsDecompressFloatImplAvx2(const char *const input, const int32_t nelemen
return 0;
}
int32_t tsDecompressTimestampAvx2(const char* const input, const int32_t nelements, char *const output, bool bigEndian) {
int32_t tsDecompressTimestampAvx2(const char *const input, const int32_t nelements, char *const output,
bool bigEndian) {
#if 0
int64_t *ostream = (int64_t *)output;
int32_t ipos = 1, opos = 0;
int8_t nbytes = 0;
int64_t prevValue = 0;
int64_t prevDelta = 0;
int64_t deltaOfDelta = 0;
int32_t longBytes = LONG_BYTES;
__m128i prevVal = _mm_setzero_si128();
__m128i prevDelta = _mm_setzero_si128();
#if __AVX2__
int32_t batch = nelements >> 1;
int32_t remainder = nelements & 0x01;
__mmask16 mask2[16] = {0, 0x0001, 0x0003, 0x0007, 0x000f, 0x001f, 0x003f, 0x007f, 0x00ff};
int32_t batch = nelements >> 2;
int32_t remainder = nelements & 0x1;
while (1) {
int32_t i = 0;
if (batch > 1) {
// first loop
uint8_t flags = input[ipos++];
// Decode dd1
uint64_t dd1 = 0;
nbytes = flags & INT8MASK(4);
int8_t nbytes1 = flags & INT8MASK(4); // range of nbytes starts from 0 to 7
int8_t nbytes2 = (flags >> 4) & INT8MASK(4);
__m128i data1;
if (nbytes1 == 0) {
data1 = _mm_setzero_si128();
} else {
memcpy(&data1, (const void*) (input + ipos), nbytes1);
}
__m128i data2;
if (nbytes2 == 0) {
data2 = _mm_setzero_si128();
} else {
memcpy(&data2, (const void*) (input + ipos + nbytes1), nbytes2);
}
data2 = _mm_broadcastq_epi64(data2);
__m128i zzVal = _mm_blend_epi32(data2, data1, 0x03);
// ZIGZAG_DECODE(T, v) (((v) >> 1) ^ -((T)((v)&1)))
__m128i signmask = _mm_and_si128(_mm_set1_epi64x(1), zzVal);
signmask = _mm_sub_epi64(_mm_setzero_si128(), signmask);
// get two zigzag values here
__m128i deltaOfDelta = _mm_xor_si128(_mm_srli_epi64(zzVal, 1), signmask);
__m128i deltaCurrent = _mm_add_epi64(deltaOfDelta, prevDelta);
deltaCurrent = _mm_add_epi64(_mm_slli_si128(deltaCurrent, 8), deltaCurrent);
__m128i val = _mm_add_epi64(deltaCurrent, prevVal);
_mm_storeu_si128((__m128i *)&ostream[opos], val);
// keep the previous value
prevVal = _mm_shuffle_epi32 (val, 0xEE);
// keep the previous delta of delta, for the first item
prevDelta = _mm_shuffle_epi32(deltaOfDelta, 0xEE);
opos += 2;
ipos += nbytes1 + nbytes2;
i += 1;
}
// the remain
for(; i < batch; ++i) {
uint8_t flags = input[ipos++];
int8_t nbytes1 = flags & INT8MASK(4); // range of nbytes starts from 0 to 7
int8_t nbytes2 = (flags >> 4) & INT8MASK(4);
// __m128i data1 = _mm_maskz_loadu_epi8(mask2[nbytes1], (const void*)(input + ipos));
// __m128i data2 = _mm_maskz_loadu_epi8(mask2[nbytes2], (const void*)(input + ipos + nbytes1));
__m128i data1;
if (nbytes1 == 0) {
data1 = _mm_setzero_si128();
} else {
int64_t dd = 0;
memcpy(&dd, (const void*) (input + ipos), nbytes1);
data1 = _mm_loadu_si64(&dd);
}
__m128i data2;
if (nbytes2 == 0) {
data2 = _mm_setzero_si128();
} else {
int64_t dd = 0;
memcpy(&dd, (const void*) (input + ipos + nbytes1), nbytes2);
data2 = _mm_loadu_si64(&dd);
}
data2 = _mm_broadcastq_epi64(data2);
__m128i zzVal = _mm_blend_epi32(data2, data1, 0x03);
// ZIGZAG_DECODE(T, v) (((v) >> 1) ^ -((T)((v)&1)))
__m128i signmask = _mm_and_si128(_mm_set1_epi64x(1), zzVal);
signmask = _mm_sub_epi64(_mm_setzero_si128(), signmask);
// get two zigzag values here
__m128i deltaOfDelta = _mm_xor_si128(_mm_srli_epi64(zzVal, 1), signmask);
__m128i deltaCurrent = _mm_add_epi64(deltaOfDelta, prevDelta);
deltaCurrent = _mm_add_epi64(_mm_slli_si128(deltaCurrent, 8), deltaCurrent);
__m128i val = _mm_add_epi64(deltaCurrent, prevVal);
_mm_storeu_si128((__m128i *)&ostream[opos], val);
// keep the previous value
prevVal = _mm_shuffle_epi32 (val, 0xEE);
// keep the previous delta of delta
__m128i delta = _mm_add_epi64(_mm_slli_si128(deltaOfDelta, 8), deltaOfDelta);
prevDelta = _mm_shuffle_epi32(_mm_add_epi64(delta, prevDelta), 0xEE);
opos += 2;
ipos += nbytes1 + nbytes2;
}
if (remainder > 0) {
uint64_t dd = 0;
uint8_t flags = input[ipos++];
int32_t nbytes = flags & INT8MASK(4);
int64_t deltaOfDelta = 0;
if (nbytes == 0) {
deltaOfDelta = 0;
} else {
if (bigEndian) {
memcpy(((char *)(&dd1)) + longBytes - nbytes, input + ipos, nbytes);
} else {
memcpy(&dd1, input + ipos, nbytes);
}
deltaOfDelta = ZIGZAG_DECODE(int64_t, dd1);
// if (is_bigendian()) {
// memcpy(((char *)(&dd1)) + longBytes - nbytes, input + ipos, nbytes);
// } else {
memcpy(&dd, input + ipos, nbytes);
// }
deltaOfDelta = ZIGZAG_DECODE(int64_t, dd);
}
ipos += nbytes;
prevDelta += deltaOfDelta;
prevValue += prevDelta;
ostream[opos++] = prevValue;
if (opos == nelements) {
return nelements * longBytes;
}
// Decode dd2
uint64_t dd2 = 0;
nbytes = (flags >> 4) & INT8MASK(4);
if (nbytes == 0) {
deltaOfDelta = 0;
if (opos == 0) {
ostream[opos++] = deltaOfDelta;
} else {
if (bigEndian) {
memcpy(((char *)(&dd2)) + longBytes - nbytes, input + ipos, nbytes);
} else {
memcpy(&dd2, input + ipos, nbytes);
}
// zigzag_decoding
deltaOfDelta = ZIGZAG_DECODE(int64_t, dd2);
}
ipos += nbytes;
prevDelta += deltaOfDelta;
prevValue += prevDelta;
ostream[opos++] = prevValue;
if (opos == nelements) {
return nelements * longBytes;
int64_t prevDeltaX = deltaOfDelta + prevDelta[1];
ostream[opos++] = prevVal[1] + prevDeltaX;
}
}
#endif
#endif
return 0;
}
int32_t tsDecompressTimestampAvx512(const char *const input, const int32_t nelements, char *const output,
bool UNUSED_PARAM(bigEndian)) {
int64_t *ostream = (int64_t *)output;
int32_t ipos = 1, opos = 0;
#if __AVX512VL__
__m128i prevVal = _mm_setzero_si128();
__m128i prevDelta = _mm_setzero_si128();
int32_t numOfBatch = nelements >> 1;
int32_t remainder = nelements & 0x01;
__mmask16 mask2[16] = {0, 0x0001, 0x0003, 0x0007, 0x000f, 0x001f, 0x003f, 0x007f, 0x00ff};
int32_t i = 0;
if (numOfBatch > 1) {
// first loop
uint8_t flags = input[ipos++];
int8_t nbytes1 = flags & INT8MASK(4); // range of nbytes starts from 0 to 7
int8_t nbytes2 = (flags >> 4) & INT8MASK(4);
__m128i data1 = _mm_maskz_loadu_epi8(mask2[nbytes1], (const void*)(input + ipos));
__m128i data2 = _mm_maskz_loadu_epi8(mask2[nbytes2], (const void*)(input + ipos + nbytes1));
data2 = _mm_broadcastq_epi64(data2);
__m128i zzVal = _mm_blend_epi32(data2, data1, 0x03);
// ZIGZAG_DECODE(T, v) (((v) >> 1) ^ -((T)((v)&1)))
__m128i signmask = _mm_and_si128(_mm_set1_epi64x(1), zzVal);
signmask = _mm_sub_epi64(_mm_setzero_si128(), signmask);
// get two zigzag values here
__m128i deltaOfDelta = _mm_xor_si128(_mm_srli_epi64(zzVal, 1), signmask);
__m128i deltaCurrent = _mm_add_epi64(deltaOfDelta, prevDelta);
deltaCurrent = _mm_add_epi64(_mm_slli_si128(deltaCurrent, 8), deltaCurrent);
__m128i val = _mm_add_epi64(deltaCurrent, prevVal);
_mm_storeu_si128((__m128i *)&ostream[opos], val);
// keep the previous value
prevVal = _mm_shuffle_epi32 (val, 0xEE);
// keep the previous delta of delta, for the first item
prevDelta = _mm_shuffle_epi32(deltaOfDelta, 0xEE);
opos += 2;
ipos += nbytes1 + nbytes2;
i += 1;
}
// the remain
for(; i < numOfBatch; ++i) {
uint8_t flags = input[ipos++];
int8_t nbytes1 = flags & INT8MASK(4); // range of nbytes starts from 0 to 7
int8_t nbytes2 = (flags >> 4) & INT8MASK(4);
__m128i data1 = _mm_maskz_loadu_epi8(mask2[nbytes1], (const void*)(input + ipos));
__m128i data2 = _mm_maskz_loadu_epi8(mask2[nbytes2], (const void*)(input + ipos + nbytes1));
data2 = _mm_broadcastq_epi64(data2);
__m128i zzVal = _mm_blend_epi32(data2, data1, 0x03);
// ZIGZAG_DECODE(T, v) (((v) >> 1) ^ -((T)((v)&1)))
__m128i signmask = _mm_and_si128(_mm_set1_epi64x(1), zzVal);
signmask = _mm_sub_epi64(_mm_setzero_si128(), signmask);
// get two zigzag values here
__m128i deltaOfDelta = _mm_xor_si128(_mm_srli_epi64(zzVal, 1), signmask);
__m128i deltaCurrent = _mm_add_epi64(deltaOfDelta, prevDelta);
deltaCurrent = _mm_add_epi64(_mm_slli_si128(deltaCurrent, 8), deltaCurrent);
__m128i val = _mm_add_epi64(deltaCurrent, prevVal);
_mm_storeu_si128((__m128i *)&ostream[opos], val);
// keep the previous value
prevVal = _mm_shuffle_epi32 (val, 0xEE);
// keep the previous delta of delta
__m128i delta = _mm_add_epi64(_mm_slli_si128(deltaOfDelta, 8), deltaOfDelta);
prevDelta = _mm_shuffle_epi32(_mm_add_epi64(delta, prevDelta), 0xEE);
opos += 2;
ipos += nbytes1 + nbytes2;
}
if (remainder > 0) {
uint64_t dd = 0;
uint8_t flags = input[ipos++];
int32_t nbytes = flags & INT8MASK(4);
int64_t deltaOfDelta = 0;
if (nbytes == 0) {
deltaOfDelta = 0;
} else {
memcpy(&dd, input + ipos, nbytes);
deltaOfDelta = ZIGZAG_DECODE(int64_t, dd);
}
ipos += nbytes;
if (opos == 0) {
ostream[opos++] = deltaOfDelta;
} else {
int64_t prevDeltaX = deltaOfDelta + prevDelta[1];
ostream[opos++] = prevVal[1] + prevDeltaX;
}
}
#endif
return 0;
}

94
source/util/test/decompressTest.cpp Normal file
View File

@ -0,0 +1,94 @@
#include <gtest/gtest.h>
#include <stdlib.h>
#include <tcompression.h>
#include <random>
namespace {} // namespace
TEST(utilTest, decompress_test) {
int64_t tsList[10] = {1700000000, 1700000100, 1700000200, 1700000300, 1700000400,
1700000500, 1700000600, 1700000700, 1700000800, 1700000900};
char* pOutput[10 * sizeof(int64_t)] = {0};
int32_t len = tsCompressTimestamp(tsList, sizeof(tsList), sizeof(tsList) / sizeof(tsList[0]), pOutput, 10, ONE_STAGE_COMP, NULL, 0);
char* decompOutput[10 * 8] = {0};
tsDecompressTimestamp(pOutput, len, 10, decompOutput, sizeof(int64_t)*10, ONE_STAGE_COMP, NULL, 0);
for(int32_t i = 0; i < 10; ++i) {
std::cout<< ((int64_t*)decompOutput)[i] << std::endl;
}
memset(decompOutput, 0, 10*8);
tsDecompressTimestampAvx512(reinterpret_cast<const char* const>(pOutput), 10,
reinterpret_cast<char* const>(decompOutput), false);
for(int32_t i = 0; i < 10; ++i) {
std::cout<<((int64_t*)decompOutput)[i] << std::endl;
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
int64_t tsList1[7] = {1700000000, 1700000000, 1700000000, 1700000000, 1700000000, 1700000000, 1700000900};
int32_t len1 = tsCompressTimestamp(tsList1, sizeof(tsList1), sizeof(tsList1) / sizeof(tsList1[0]), pOutput, 7, ONE_STAGE_COMP, NULL, 0);
memset(decompOutput, 0, 10*8);
tsDecompressTimestampAvx512(reinterpret_cast<const char* const>(pOutput), 7,
reinterpret_cast<char* const>(decompOutput), false);
for(int32_t i = 0; i < 7; ++i) {
std::cout<<((int64_t*)decompOutput)[i] << std::endl;
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
int64_t tsList2[1] = {1700000000};
int32_t len2 = tsCompressTimestamp(tsList2, sizeof(tsList2), sizeof(tsList2) / sizeof(tsList2[0]), pOutput, 1, ONE_STAGE_COMP, NULL, 0);
memset(decompOutput, 0, 10*8);
tsDecompressTimestampAvx512(reinterpret_cast<const char* const>(pOutput), 1,
reinterpret_cast<char* const>(decompOutput), false);
for(int32_t i = 0; i < 1; ++i) {
std::cout<<((int64_t*)decompOutput)[i] << std::endl;
}
}
TEST(utilTest, decompress_perf_test) {
int32_t num = 10000;
int64_t* pList = static_cast<int64_t*>(taosMemoryCalloc(num, sizeof(int64_t)));
int64_t iniVal = 1700000000;
uint32_t v = 100;
for(int32_t i = 0; i < num; ++i) {
iniVal += taosRandR(&v)%10;
pList[i] = iniVal;
}
char* px = static_cast<char*>(taosMemoryMalloc(num * sizeof(int64_t)));
int32_t len = tsCompressTimestamp(pList, num * sizeof(int64_t), num, px, num, ONE_STAGE_COMP, NULL, 0);
char* pOutput = static_cast<char*>(taosMemoryMalloc(num * sizeof(int64_t)));
int64_t st = taosGetTimestampUs();
for(int32_t k = 0; k < 10000; ++k) {
tsDecompressTimestamp(px, len, num, pOutput, sizeof(int64_t) * num, ONE_STAGE_COMP, NULL, 0);
}
int64_t el1 = taosGetTimestampUs() - st;
std::cout << "soft decompress elapsed time:" << el1 << " us" << std::endl;
memset(pOutput, 0, num * sizeof(int64_t));
st = taosGetTimestampUs();
for(int32_t k = 0; k < 10000; ++k) {
tsDecompressTimestampAvx512(px, num, pOutput, false);
}
int64_t el2 = taosGetTimestampUs() - st;
std::cout << "SIMD decompress elapsed time:" << el2 << " us" << std::endl;
taosMemoryFree(pList);
taosMemoryFree(pOutput);
taosMemoryFree(px);
}

3
tests/develop-test/win-test-file
View File

@ -1,4 +1,7 @@
python3 ./test.py -f 2-query/table_count_scan.py
python3 ./test.py -f 2-query/pseudo_column.py
python3 ./test.py -f 2-query/ts-range.py
python3 ./test.py -f 2-query/tag_scan.py
python3 ./test.py -f 2-query/show_create_db.py
python3 ./test.py -f 5-taos-tools/taosbenchmark/auto_create_table_json.py
python3 ./test.py -f 5-taos-tools/taosbenchmark/custom_col_tag.py

2
tests/parallel_test/cases.task
View File

@ -198,6 +198,7 @@
#,,y,system-test,./pytest.sh python3 test.py -f 7-tmq/tmqVnodeTransform-db.py -N 6 -n 3
,,y,system-test,./pytest.sh python3 test.py -f 7-tmq/tmqVnodeSplit-stb-select.py -N 2 -n 1
,,y,system-test,./pytest.sh python3 test.py -f 7-tmq/tmqVnodeSplit-stb-select-duplicatedata.py -N 3 -n 3
,,y,system-test,./pytest.sh python3 test.py -f 7-tmq/tmqVnodeSplit-stb-select-duplicatedata-false.py -N 3 -n 3
,,y,system-test,./pytest.sh python3 test.py -f 7-tmq/tmqVnodeSplit-stb-select.py -N 3 -n 3
,,y,system-test,./pytest.sh python3 test.py -f 7-tmq/tmqVnodeSplit-stb.py -N 3 -n 3
,,y,system-test,./pytest.sh python3 test.py -f 7-tmq/tmqVnodeSplit-column.py -N 3 -n 3
@ -299,6 +300,7 @@ e
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/precisionUS.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/precisionNS.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/test_ts4219.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/ts-4272.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/test_ts4295.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/test_td27388.py
,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/show.py

60
tests/script/win-test-file
View File

@ -19,19 +19,18 @@
./test.sh -f tsim/dnode/balance3.sim
./test.sh -f tsim/vnode/replica3_many.sim
./test.sh -f tsim/stable/metrics_idx.sim
./test.sh -f tsim/db/alter_replica_13.sim
./test.sh -f tsim/sync/3Replica1VgElect.sim
./test.sh -f tsim/sync/3Replica5VgElect.sim
./test.sh -f tsim/valgrind/checkError6.sim
./test.sh -f tsim/user/basic.sim
./test.sh -f tsim/user/password.sim
./test.sh -f tsim/user/whitelist.sim
./test.sh -f tsim/user/privilege_db.sim
./test.sh -f tsim/user/privilege_sysinfo.sim
./test.sh -f tsim/user/privilege_topic.sim
./test.sh -f tsim/user/privilege_table.sim
./test.sh -f tsim/user/privilege_create_db.sim
./test.sh -f tsim/db/alter_option.sim
./test.sh -f tsim/db/alter_replica_31.sim
./test.sh -f tsim/db/basic1.sim
./test.sh -f tsim/db/basic2.sim
./test.sh -f tsim/db/basic3.sim
@ -97,6 +96,7 @@
./test.sh -f tsim/insert/delete0.sim
./test.sh -f tsim/insert/update1_sort_merge.sim
./test.sh -f tsim/insert/update2.sim
./test.sh -f tsim/insert/insert_stb.sim
./test.sh -f tsim/parser/alter__for_community_version.sim
./test.sh -f tsim/parser/alter_column.sim
./test.sh -f tsim/parser/alter_stable.sim
@ -186,6 +186,7 @@
./test.sh -f tsim/query/session.sim
./test.sh -f tsim/query/join_interval.sim
./test.sh -f tsim/query/join_pk.sim
./test.sh -f tsim/query/count_spread.sim
./test.sh -f tsim/query/unionall_as_table.sim
./test.sh -f tsim/query/multi_order_by.sim
./test.sh -f tsim/query/sys_tbname.sim
@ -197,10 +198,14 @@
./test.sh -f tsim/query/emptyTsRange_scl.sim
./test.sh -f tsim/query/partitionby.sim
./test.sh -f tsim/query/tableCount.sim
./test.sh -f tsim/query/show_db_table_kind.sim
./test.sh -f tsim/query/bi_star_table.sim
./test.sh -f tsim/query/bi_tag_scan.sim
./test.sh -f tsim/query/tag_scan.sim
./test.sh -f tsim/query/nullColSma.sim
./test.sh -f tsim/query/bug3398.sim
./test.sh -f tsim/query/explain_tsorder.sim
./test.sh -f tsim/query/apercentile.sim
./test.sh -f tsim/qnode/basic1.sim
./test.sh -f tsim/snode/basic1.sim
./test.sh -f tsim/mnode/basic1.sim
@ -237,52 +242,6 @@
./test.sh -f tsim/table/table.sim
./test.sh -f tsim/table/tinyint.sim
./test.sh -f tsim/table/vgroup.sim
./test.sh -f tsim/stream/basic0.sim -g
./test.sh -f tsim/stream/basic1.sim
./test.sh -f tsim/stream/basic2.sim
./test.sh -f tsim/stream/basic3.sim
./test.sh -f tsim/stream/basic4.sim
./test.sh -f tsim/stream/checkpointInterval0.sim
./test.sh -f tsim/stream/checkStreamSTable1.sim
./test.sh -f tsim/stream/checkStreamSTable.sim
./test.sh -f tsim/stream/deleteInterval.sim
./test.sh -f tsim/stream/deleteSession.sim
./test.sh -f tsim/stream/deleteState.sim
./test.sh -f tsim/stream/distributeInterval0.sim
./test.sh -f tsim/stream/distributeIntervalRetrive0.sim
./test.sh -f tsim/stream/distributeSession0.sim
./test.sh -f tsim/stream/drop_stream.sim
./test.sh -f tsim/stream/fillHistoryBasic1.sim
./test.sh -f tsim/stream/fillHistoryBasic2.sim
./test.sh -f tsim/stream/fillHistoryBasic3.sim
./test.sh -f tsim/stream/fillIntervalDelete0.sim
./test.sh -f tsim/stream/fillIntervalDelete1.sim
./test.sh -f tsim/stream/fillIntervalLinear.sim
./test.sh -f tsim/stream/fillIntervalPartitionBy.sim
./test.sh -f tsim/stream/fillIntervalPrevNext1.sim
./test.sh -f tsim/stream/fillIntervalPrevNext.sim
./test.sh -f tsim/stream/fillIntervalRange.sim
./test.sh -f tsim/stream/fillIntervalValue.sim
./test.sh -f tsim/stream/ignoreCheckUpdate.sim
./test.sh -f tsim/stream/ignoreExpiredData.sim
./test.sh -f tsim/stream/partitionby1.sim
./test.sh -f tsim/stream/partitionbyColumnInterval.sim
./test.sh -f tsim/stream/partitionbyColumnSession.sim
./test.sh -f tsim/stream/partitionbyColumnState.sim
./test.sh -f tsim/stream/partitionby.sim
./test.sh -f tsim/stream/pauseAndResume.sim
./test.sh -f tsim/stream/schedSnode.sim
./test.sh -f tsim/stream/session0.sim
./test.sh -f tsim/stream/session1.sim
./test.sh -f tsim/stream/sliding.sim
./test.sh -f tsim/stream/state0.sim
./test.sh -f tsim/stream/state1.sim
./test.sh -f tsim/stream/triggerInterval0.sim
./test.sh -f tsim/stream/triggerSession0.sim
./test.sh -f tsim/stream/udTableAndTag0.sim
./test.sh -f tsim/stream/udTableAndTag1.sim
./test.sh -f tsim/stream/udTableAndTag2.sim
./test.sh -f tsim/stream/windowClose.sim
./test.sh -f tsim/trans/lossdata1.sim
./test.sh -f tsim/tmq/basic1.sim
./test.sh -f tsim/tmq/basic2.sim
@ -431,3 +390,8 @@
./test.sh -f tsim/tag/drop_tag.sim
./test.sh -f tsim/tag/tbNameIn.sim
./test.sh -f tmp/monitor.sim
./test.sh -f tsim/tagindex/add_index.sim
./test.sh -f tsim/tagindex/sma_and_tag_index.sim
./test.sh -f tsim/view/view.sim
./test.sh -f tsim/query/cache_last.sim
./test.sh -f tsim/query/const.sim

19
tests/system-test/0-others/user_privilege.py
View File

@ -27,6 +27,7 @@ class TDTestCase:
tdSql.init(conn.cursor())
self.setsql = TDSetSql()
self.stbname = 'stb'
self.user_name = 'test'
self.binary_length = 20 # the length of binary for column_dict
self.nchar_length = 20 # the length of nchar for column_dict
self.dbnames = ['db1', 'db2']
@ -54,12 +55,12 @@ class TDTestCase:
]
self.tbnum = 4
self.stbnum_grant = 200
def create_user(self):
user_name = 'test'
tdSql.execute(f'create user {user_name} pass "test"')
tdSql.execute(f'grant read on {self.dbnames[0]}.{self.stbname} with t2 = "Beijing" to {user_name}')
tdSql.execute(f'grant write on {self.dbnames[1]}.{self.stbname} with t1 = 2 to {user_name}')
tdSql.execute(f'create user {self.user_name} pass "test"')
tdSql.execute(f'grant read on {self.dbnames[0]}.{self.stbname} with t2 = "Beijing" to {self.user_name}')
tdSql.execute(f'grant write on {self.dbnames[1]}.{self.stbname} with t1 = 2 to {self.user_name}')
def prepare_data(self):
for db in self.dbnames:
@ -70,6 +71,8 @@ class TDTestCase:
tdSql.execute(f'create table {self.stbname}_{i} using {self.stbname} tags({self.tag_list[i]})')
for j in self.values_list:
tdSql.execute(f'insert into {self.stbname}_{i} values({j})')
for i in range(self.stbnum_grant):
tdSql.execute(f'create table {self.stbname}_grant_{i} (ts timestamp, c0 int) tags(t0 int)')
def user_read_privilege_check(self, dbname):
testconn = taos.connect(user='test', password='test')
@ -128,12 +131,20 @@ class TDTestCase:
tdLog.exit(f"{caller.filename}({caller.lineno}) failed: sql:{sql}, expect error not occured")
pass
def user_privilege_grant_check(self):
for db in self.dbnames:
tdSql.execute(f"use {db}")
for i in range(self.stbnum_grant):
tdSql.execute(f'grant read on {db}.{self.stbname}_grant_{i} to {self.user_name}')
tdSql.execute(f'grant write on {db}.{self.stbname}_grant_{i} to {self.user_name}')
def run(self):
self.prepare_data()
self.create_user()
self.user_read_privilege_check(self.dbnames[0])
self.user_write_privilege_check(self.dbnames[1])
self.user_privilege_error_check()
self.user_privilege_grant_check()
def stop(self):
tdSql.close()

205
tests/system-test/1-insert/ts-4272.py Normal file
View File

@ -0,0 +1,205 @@
import csv
from datetime import datetime
import taos
from util.log import *
from util.sql import *
from util.cases import *
from util.dnodes import *
from util.common import *
class TDTestCase:
def init(self, conn, logSql, replicaVar=1):
self.replicaVar = int(replicaVar)
self.testcasePath = os.path.split(__file__)[0]
self.testcasefilename = os.path.split(__file__)[-1]
self.ts = 1700638570000 # 2023-11-22T07:36:10.000Z
self.db = 'db1'
self.tb1 = 'd001'
self.tb2 = 'd002'
self.stable0 = "meters"
self.stable1 = "stb_1"
self.stable2 = "stb_null"
self.tag1 = f'using {self.stable0}(groupId) tags(1)'
self.tag2 = f'using {self.stable0}(groupId) tags(2)'
self.file1 = f"{self.testcasePath}/b.csv"
self.file2 = f"{self.testcasePath}/c.csv"
#os.system("rm -rf %s/b.csv" %self.testcasePath)
tdLog.debug(f"start to excute {__file__}")
tdSql.init(conn.cursor(), logSql)
def check_count(self, rows, records):
tdSql.execute(f"use {self.db};")
tdSql.query(f"select tbname,count(*) from {self.stable0} group by tbname order by tbname;")
tdSql.checkRows(rows)
for i in range(rows):
tdSql.checkData(i, 1, records[i])
def reset_tb(self):
# create database and tables
# os.system("taos -s 'drop database if exists d1;'")
# os.system("taos -s 'create database d1;use d1;create stable meters (ts timestamp, current float, voltage int, phase float) tags (location binary(64), groupId int);'")
# os.system(f"taos -s 'use d1;create table d2001 using meters(groupId) tags(5);'")
# res = os.system(f"taos -s 'use d1;create table d2002 using meters(groupId) tags(6);'")
# if (0 != res):
# tdLog.exit(f"create tb error")
tdSql.execute(f"drop database if exists {self.db};")
tdSql.execute(f"create database {self.db};")
tdSql.execute(f"use {self.db};")
tdSql.execute(f"create stable {self.stable0} (ts timestamp, current float, voltage int, phase float) tags (location binary(64), groupId int);")
tdSql.execute(f"create table {self.tb1} {self.tag1};")
tdSql.execute(f"create table {self.tb2} {self.tag2};")
tdSql.execute(f"create stable {self.stable1} (ts timestamp , q_int int , q_bigint bigint , q_smallint smallint , q_tinyint tinyint , q_float float , q_double double , q_bool bool , q_binary binary(100) , q_nchar nchar(100) , q_ts timestamp , q_int_null int , q_bigint_null bigint , q_smallint_null smallint , q_tinyint_null tinyint, q_float_null float , q_double_null double , q_bool_null bool , q_binary_null binary(20) , q_nchar_null nchar(20) , q_ts_null timestamp) tags(loc nchar(100) , t_int int , t_bigint bigint , t_smallint smallint , t_tinyint tinyint, t_bool bool , t_binary binary(100) , t_nchar nchar(100) ,t_float float , t_double double , t_ts timestamp);")
tdSql.execute(f"create stable {self.stable2} (ts timestamp , q_int int , q_bigint bigint , q_smallint smallint , q_tinyint tinyint , q_float float , q_double double , q_bool bool , q_binary binary(100) , q_nchar nchar(100) , q_ts timestamp , q_int_null int , q_bigint_null bigint , q_smallint_null smallint , q_tinyint_null tinyint, q_float_null float , q_double_null double , q_bool_null bool , q_binary_null binary(20) , q_nchar_null nchar(20) , q_ts_null timestamp) tags(loc nchar(100) , t_int int , t_bigint bigint , t_smallint smallint , t_tinyint tinyint, t_bool bool , t_binary binary(100) , t_nchar nchar(100) ,t_float float , t_double double , t_ts timestamp);")
def test(self, sql):
sql = f"use {self.db};" + sql
res = os.system(f'taos -s "{sql}"')
# if (0 != res):
# tdLog.exit(f"taos sql error")
def check(self):
# same table, auto create + create
sql = f"insert into {self.tb1} {self.tag1} file '{self.file1}' {self.tb1} {self.tag1} file '{self.file2}';"
self.test(sql)
# same table, create + insert
sql = f"insert into {self.tb1} {self.tag1} file '{self.file1}' {self.tb1} file '{self.file2}';"
self.test(sql)
# same table, insert + create
sql = f"insert into {self.tb1} file '{self.file1}' {self.tb1} {self.tag1} file '{self.file2}';"
self.test(sql)
# same table, insert + insert
sql = f"insert into {self.tb1} file '{self.file1}' {self.tb1} file '{self.file2}';"
self.test(sql)
# diff table auto create + create
sql = f"insert into {self.tb1} {self.tag1} file '{self.file1}' {self.tb2} {self.tag2} file '{self.file2}';"
self.test(sql)
# diff table, create + insert
sql = f"insert into {self.tb1} {self.tag1} file '{self.file1}' {self.tb2} file '{self.file2}';"
self.test(sql)
# diff table, insert + create
sql = f"insert into {self.tb1} file '{self.file1}' {self.tb2} {self.tag2} file '{self.file2}';"
self.test(sql)
# diff table, insert + insert
sql = f"insert into {self.tb1} file '{self.file1}' {self.tb2} file '{self.file2}';"
self.test(sql)
# bigNum = 1010000
# self.check_count(5, [2100, 2100, bigNum, bigNum, bigNum])
result = os.popen("taos -s 'select count(*) from %s.%s'" %(self.db, self.tb1))
res = result.read()
if (f"OK" in res):
tdLog.info(f"check count success")
def make_csv(self, filepath, once, qtime, startts):
f = open(filepath, 'w')
with f:
writer = csv.writer(f)
for j in range(qtime):
ts = startts + j*once
rows = []
for i in range(once):
rows.append([ts + i, 0.3 + (i%10)/100.0, 210 + i%10, 10.0 + (i%20)/20.0])
writer.writerows(rows)
f.close()
print(datetime.now(), filepath, " ready!")
def test_mix(self):
#forbid use both value and file in one insert
result = os.popen(f"insert into {self.tb1} file '{self.file2}' {self.tb2} values('2021-07-13 14:06:34.630', 10.2, 219, 0.32);")
res = result.read()
if (f"error" in res):
tdLog.info(f"forbid success")
def test_bigcsv(self):
# prepare csv
print("start csv data prepare")
once = 10000
qtime1 = 101
qtime2 = 100
rowNum1 = qtime1 * once
rowNum2 = qtime2 * once
self.make_csv(self.file1, once, qtime1, self.ts - 86400000)
self.make_csv(self.file2, once, qtime2, self.ts)
print("end csv data prepare")
# auto create + insert
sql = f"insert into {self.tb1} {self.tag1} file '{self.file1}';"
self.test(sql)
# only insert
sql = f"insert into {self.tb2} file '{self.file2}';"
self.test(sql)
print("end insert to table")
#tdSql.execute(f"use d1;")
tdSql.query(f"select tbname,count(*) from {self.stable0} group by tbname order by tbname;")
tdSql.checkRows(2)
tdSql.checkData(0, 1, rowNum1)
tdSql.checkData(1, 1, rowNum2)
print("check insert file to table success")
def make_stable_csv(self, filepath, once, qtime, startts, table_name):
f = open(filepath, 'w')
with f:
writer = csv.writer(f)
for j in range(qtime):
offset = j*once
ts = startts + offset
rows = []
for i in range(once):
rows.append([table_name, ts + i, offset + i, 'NULL'])
writer.writerows(rows)
f.close()
print(datetime.now(), filepath, " ready!")
def test_stable_csv(self):
# prepare csv
print("start stable_csv data prepare")
once = 10000
qtime1 = 101
qtime2 = 100
# rowNum1 = qtime1 * once
# rowNum2 = qtime2 * once
child_1 = f"{self.stable1}_1"
child_2 = f"{self.stable2}_1"
self.make_stable_csv(self.file1, once, qtime1, self.ts - 86400000, child_1)
self.make_stable_csv(self.file2, once, qtime2, self.ts, child_2)
print("end stable_csv data prepare")
# insert create child table of stable
sql = f"insert into {self.db}.{self.stable1}(tbname,ts,q_int,q_binary) file '{self.file1}' {self.db}.{self.stable2}(tbname,ts,q_int,q_binary) file '{self.file2}';"
self.test(sql)
print("end insert to stable")
#tdSql.execute(f"insert into {self.db}.{child_1}(ts, q_int) values(now, 1);")
tdSql.query(f"select tbname,count(*) from {self.stable1} group by tbname order by tbname;")
tdSql.checkRows(0)
print("check stable success")
def run(self):
tdSql.prepare()
self.reset_tb()
self.test_stable_csv()
self.test_bigcsv()
self.test_mix()
self.check()
tdSql.close()
def stop(self):
tdLog.success(f"{__file__} successfully executed")
tdCases.addLinux(__file__, TDTestCase())
tdCases.addWindows(__file__, TDTestCase())

6
tests/system-test/7-tmq/tmqCommon.py
View File

@ -75,7 +75,7 @@ class TMQCom:
if tdSql.getRows() == expectRows:
break
else:
time.sleep(5)
time.sleep(0.5)
for i in range(expectRows):
tdLog.info ("consume id: %d, consume msgs: %d, consume rows: %d"%(tdSql.getData(i , 1), tdSql.getData(i , 2), tdSql.getData(i , 3)))
@ -156,7 +156,7 @@ class TMQCom:
tdLog.info("row: %d"%(actRows))
if (actRows >= rows):
loopFlag = 0
time.sleep(0.02)
time.sleep(0.5)
return
def getStartCommitNotifyFromTmqsim(self,cdbName='cdb',rows=1):
@ -167,7 +167,7 @@ class TMQCom:
tdLog.info("row: %d"%(actRows))
if (actRows >= rows):
loopFlag = 0
time.sleep(0.02)
time.sleep(0.5)
return
def create_database(self,tsql, dbName,dropFlag=1,vgroups=4,replica=1):

218
tests/system-test/7-tmq/tmqVnodeSplit-stb-select-duplicatedata-false.py Normal file
View File

@ -0,0 +1,218 @@
import taos
import sys
import time
import socket
import os
import threading
import math
from util.log import *
from util.sql import *
from util.cases import *
from util.dnodes import *
from util.common import *
from util.cluster import *
sys.path.append("./7-tmq")
from tmqCommon import *
sys.path.append("./6-cluster")
from clusterCommonCreate import *
from clusterCommonCheck import clusterComCheck
class TDTestCase:
def __init__(self):
self.vgroups = 1
self.ctbNum = 10
self.rowsPerTbl = 10000
def init(self, conn, logSql, replicaVar=1):
self.replicaVar = int(replicaVar)
tdLog.debug(f"start to excute {__file__}")
tdSql.init(conn.cursor(), False)
def getDataPath(self):
selfPath = tdCom.getBuildPath()
return selfPath + '/../sim/dnode%d/data/vnode/vnode%d/wal/*';
def prepareTestEnv(self):
tdLog.printNoPrefix("======== prepare test env include database, stable, ctables, and insert data: ")
paraDict = {'dbName': 'dbt',
'dropFlag': 1,
'event': '',
'vgroups': 1,
'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': 10,
'rowsPerTbl': 10000,
'batchNum': 10,
'startTs': 1640966400000, # 2022-01-01 00:00:00.000
'pollDelay': 60,
'showMsg': 1,
'showRow': 1,
'snapshot': 0}
paraDict['vgroups'] = self.vgroups
paraDict['ctbNum'] = self.ctbNum
paraDict['rowsPerTbl'] = self.rowsPerTbl
tdCom.drop_all_db()
tmqCom.initConsumerTable()
tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], wal_retention_period=36000,vgroups=paraDict["vgroups"],replica=self.replicaVar)
tdLog.info("create stb")
tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"])
return
def restartAndRemoveWal(self, deleteWal):
tdDnodes = cluster.dnodes
tdSql.query("select * from information_schema.ins_vnodes")
for result in tdSql.queryResult:
if result[2] == 'dbt':
tdLog.debug("dnode is %d"%(result[0]))
dnodeId = result[0]
vnodeId = result[1]
tdDnodes[dnodeId - 1].stoptaosd()
time.sleep(1)
dataPath = self.getDataPath()
dataPath = dataPath%(dnodeId,vnodeId)
tdLog.debug("dataPath:%s"%dataPath)
if deleteWal:
if os.system('rm -rf ' + dataPath) != 0:
tdLog.exit("rm error")
tdDnodes[dnodeId - 1].starttaosd()
time.sleep(1)
break
tdLog.debug("restart dnode ok")
def splitVgroups(self):
tdSql.query("select * from information_schema.ins_vnodes")
vnodeId = 0
for result in tdSql.queryResult:
if result[2] == 'dbt':
vnodeId = result[1]
tdLog.debug("vnode is %d"%(vnodeId))
break
splitSql = "split vgroup %d" %(vnodeId)
tdLog.debug("splitSql:%s"%(splitSql))
tdSql.query(splitSql)
tdLog.debug("splitSql ok")
def tmqCase1(self, deleteWal=False):
tdLog.printNoPrefix("======== test case 1: ")
paraDict = {'dbName': 'dbt',
'dropFlag': 1,
'event': '',
'vgroups': 1,
'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': 'ctb1',
'ctbStartIdx': 0,
'ctbNum': 10,
'rowsPerTbl': 10000,
'batchNum': 10,
'startTs': 1640966400000, # 2022-01-01 00:00:00.000
'pollDelay': 120,
'showMsg': 1,
'showRow': 1,
'snapshot': 0}
paraDict['vgroups'] = self.vgroups
paraDict['ctbNum'] = self.ctbNum
paraDict['rowsPerTbl'] = self.rowsPerTbl
topicNameList = ['topic1']
# expectRowsList = []
tmqCom.initConsumerTable()
tdLog.info("create topics from stb with filter")
queryString = "select * from %s.%s"%(paraDict['dbName'], paraDict['stbName'])
# sqlString = "create topic %s as stable %s" %(topicNameList[0], paraDict['stbName'])
sqlString = "create topic %s as %s" %(topicNameList[0], queryString)
tdLog.info("create topic sql: %s"%sqlString)
tdSql.execute(sqlString)
# tdSql.query(queryString)
# expectRowsList.append(tdSql.getRows())
# init consume info, and start tmq_sim, then check consume result
tdLog.info("insert consume info to consume processor")
consumerId = 0
expectrowcnt = paraDict["rowsPerTbl"] * paraDict["ctbNum"] * 2
topicList = topicNameList[0]
ifcheckdata = 1
ifManualCommit = 1
keyList = 'group.id:cgrp1, enable.auto.commit:true, auto.commit.interval.ms:200, 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("wait the consume result")
tdLog.info("create ctb1")
tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict['ctbPrefix'],
ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx'])
tdLog.info("create ctb2")
paraDict['ctbPrefix'] = "ctb2"
tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict['ctbPrefix'],
ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx'])
tdLog.info("insert ctb1 data")
pInsertThread = tmqCom.asyncInsertDataByInterlace(paraDict)
tmqCom.getStartConsumeNotifyFromTmqsim()
tmqCom.getStartCommitNotifyFromTmqsim()
#restart dnode & remove wal
self.restartAndRemoveWal(deleteWal)
# split vgroup
self.splitVgroups()
tdLog.info("insert ctb2 data")
pInsertThread1 = tmqCom.asyncInsertDataByInterlace(paraDict)
pInsertThread.join()
pInsertThread1.join()
expectRows = 1
resultList = tmqCom.selectConsumeResult(expectRows)
if expectrowcnt / 2 >= resultList[0]:
tdLog.info("expect consume rows: %d, act consume rows: %d"%(expectrowcnt / 2, resultList[0]))
tdLog.exit("%d tmq consume rows error!"%consumerId)
# tmqCom.checkFileContent(consumerId, queryString)
time.sleep(2)
for i in range(len(topicNameList)):
tdSql.query("drop topic %s"%topicNameList[i])
if deleteWal == True:
clusterComCheck.check_vgroups_status(vgroup_numbers=2,db_replica=self.replicaVar,db_name="dbt",count_number=240)
tdLog.printNoPrefix("======== test case 1 end ...... ")
def run(self):
self.prepareTestEnv()
self.tmqCase1(False)
def stop(self):
tdSql.close()
tdLog.success(f"{__file__} successfully executed")
event = threading.Event()
tdCases.addLinux(__file__, TDTestCase())
tdCases.addWindows(__file__, TDTestCase())

2
tests/system-test/7-tmq/tmqVnodeSplit-stb-select-duplicatedata.py
View File

@ -207,8 +207,6 @@ class TDTestCase:
def run(self):
self.prepareTestEnv()
self.tmqCase1(True)
self.prepareTestEnv()
self.tmqCase1(False)
def stop(self):
tdSql.close()

196
tests/system-test/7-tmq/tmq_per.py Normal file
View File

@ -0,0 +1,196 @@
import taos
import sys
import time
import socket
import os
import threading
import math
from util.log import *
from util.sql import *
from util.cases import *
from util.dnodes import *
from util.common import *
from util.cluster import *
sys.path.append("./7-tmq")
from tmqCommon import *
from util.cluster import *
sys.path.append("./6-cluster")
from clusterCommonCreate import *
from clusterCommonCheck import clusterComCheck
class TDTestCase:
def __init__(self):
self.vgroups = 1
self.ctbNum = 10000
self.rowsPerTbl = 10000
def init(self, conn, logSql, replicaVar=1):
self.replicaVar = int(replicaVar)
tdLog.debug(f"start to excute {__file__}")
tdSql.init(conn.cursor(), False)
def getDataPath(self):
selfPath = tdCom.getBuildPath()
return selfPath + '/../sim/dnode%d/data/vnode/vnode%d/wal/*';
def prepareTestEnv(self):
tdLog.printNoPrefix("======== prepare test env include database, stable, ctables, and insert data: ")
paraDict = {'dbName': 'dbt',
'dropFlag': 1,
'event': '',
'vgroups': 1,
'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': 10,
'rowsPerTbl': 10000,
'batchNum': 10,
'startTs': 1640966400000, # 2022-01-01 00:00:00.000
'pollDelay': 60,
'showMsg': 1,
'showRow': 1,
'snapshot': 0}
paraDict['vgroups'] = self.vgroups
paraDict['ctbNum'] = self.ctbNum
paraDict['rowsPerTbl'] = self.rowsPerTbl
tdCom.drop_all_db()
tmqCom.initConsumerTable()
tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], wal_retention_period=36000,vgroups=paraDict["vgroups"],replica=self.replicaVar)
tdLog.info("create stb")
tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"])
return
def restartAndRemoveWal(self, deleteWal):
tdDnodes = cluster.dnodes
tdSql.query("select * from information_schema.ins_vnodes")
for result in tdSql.queryResult:
if result[2] == 'dbt':
tdLog.debug("dnode is %d"%(result[0]))
dnodeId = result[0]
vnodeId = result[1]
tdDnodes[dnodeId - 1].stoptaosd()
time.sleep(1)
dataPath = self.getDataPath()
dataPath = dataPath%(dnodeId,vnodeId)
tdLog.debug("dataPath:%s"%dataPath)
if deleteWal:
if os.system('rm -rf ' + dataPath) != 0:
tdLog.exit("rm error")
tdDnodes[dnodeId - 1].starttaosd()
time.sleep(1)
break
tdLog.debug("restart dnode ok")
def splitVgroups(self):
tdSql.query("select * from information_schema.ins_vnodes")
vnodeId = 0
for result in tdSql.queryResult:
if result[2] == 'dbt':
vnodeId = result[1]
tdLog.debug("vnode is %d"%(vnodeId))
break
splitSql = "split vgroup %d" %(vnodeId)
tdLog.debug("splitSql:%s"%(splitSql))
tdSql.query(splitSql)
tdLog.debug("splitSql ok")
def tmqCase1(self, deleteWal=False):
tdLog.printNoPrefix("======== test case 1: ")
paraDict = {'dbName': 'dbt',
'dropFlag': 1,
'event': '',
'vgroups': 1,
'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': 'ctb1',
'ctbStartIdx': 0,
'ctbNum': 10000,
'rowsPerTbl': 10000,
'batchNum': 10000,
'startTs': 1640966400000, # 2022-01-01 00:00:00.000
'pollDelay': 5,
'showMsg': 1,
'showRow': 1,
'snapshot': 0}
paraDict['vgroups'] = self.vgroups
paraDict['ctbNum'] = self.ctbNum
print
paraDict['rowsPerTbl'] = self.rowsPerTbl
topicNameList = ['topic1']
# expectRowsList = []
tmqCom.initConsumerTable()
tdLog.info("create topics from stb ")
queryString = "stable %s.%s"%(paraDict['dbName'], paraDict['stbName'])
# sqlString = "create topic %s as stable %s" %(topicNameList[0], paraDict['stbName'])
sqlString = "create topic %s as %s" %(topicNameList[0], queryString)
tdLog.info("create topic sql: %s"%sqlString)
tdSql.execute(sqlString)
# tdSql.query(queryString)
# expectRowsList.append(tdSql.getRows())
# init consume info, and start tmq_sim, then check consume result
tdLog.info("insert consume info to consume processor")
consumerId = 0
expectrowcnt = paraDict["rowsPerTbl"] * paraDict["ctbNum"]
topicList = topicNameList[0]
ifcheckdata = 1
ifManualCommit = 1
keyList = 'group.id:cgrp1, enable.auto.commit:true, auto.commit.interval.ms:200, auto.offset.reset:earliest'
tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
tdLog.info("create ctb1")
tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict['ctbPrefix'],
ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx'])
tdLog.info("insert ctb1 data")
pInsertThread = tmqCom.asyncInsertDataByInterlace(paraDict)
pInsertThread.join()
tdLog.info("start consume processor")
tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot'])
tdLog.info("wait the consume result")
tmqCom.getStartConsumeNotifyFromTmqsim()
tmqCom.getStartCommitNotifyFromTmqsim()
expectRows = 1
tdLog.info("expectRows:%d"%expectRows)
resultList = tmqCom.selectConsumeResult(expectRows)
# for i in range(len(topicNameList)):
# tdSql.query("drop topic %s"%topicNameList[i])
if deleteWal == True:
clusterComCheck.check_vgroups_status(vgroup_numbers=1,db_replica=self.replicaVar,db_name="dbt",count_number=240)
tdLog.printNoPrefix("======== test case 1 end ...... ")
def run(self):
self.prepareTestEnv()
self.tmqCase1(True)
def stop(self):
tdSql.close()
tdLog.success(f"{__file__} successfully executed")
event = threading.Event()
tdCases.addLinux(__file__, TDTestCase())
tdCases.addWindows(__file__, TDTestCase())

79
tests/system-test/win-test-file
View File

@ -1,3 +1,12 @@
python3 ./test.py -f 2-query/tbname_vgroup.py
python3 ./test.py -f 2-query/stbJoin.py
python3 ./test.py -f 2-query/stbJoin.py -Q 2
python3 ./test.py -f 2-query/stbJoin.py -Q 3
python3 ./test.py -f 2-query/stbJoin.py -Q 4
python3 ./test.py -f 2-query/hint.py
python3 ./test.py -f 2-query/hint.py -Q 2
python3 ./test.py -f 2-query/hint.py -Q 3
python3 ./test.py -f 2-query/hint.py -Q 4
python3 ./test.py -f 2-query/nestedQuery.py
python3 ./test.py -f 2-query/nestedQuery_str.py
python3 ./test.py -f 2-query/nestedQuery_math.py
@ -18,7 +27,30 @@ python3 ./test.py -f 2-query/nestedQuery_math.py -Q 4
python3 ./test.py -f 2-query/nestedQuery_time.py -Q 4
python3 ./test.py -f 2-query/nestedQuery_26.py -Q 4
python3 ./test.py -f 2-query/interval_limit_opt.py -Q 4
python3 ./test.py -f 2-query/interval_unit.py
python3 ./test.py -f 2-query/interval_unit.py -Q 2
python3 ./test.py -f 2-query/interval_unit.py -Q 3
python3 ./test.py -f 2-query/interval_unit.py -Q 4
python3 ./test.py -f 2-query/partition_by_col.py -Q 4
python3 ./test.py -f 2-query/partition_by_col.py -Q 3
python3 ./test.py -f 2-query/partition_by_col.py -Q 2
python3 ./test.py -f 2-query/partition_by_col.py
python3 ./test.py -f 2-query/partition_by_col_agg.py
python3 ./test.py -f 2-query/partition_by_col_agg.py -Q 2
python3 ./test.py -f 2-query/partition_by_col_agg.py -Q 3
python3 ./test.py -f 2-query/partition_by_col_agg.py -Q 4
python3 ./test.py -f 2-query/interval_limit_opt_2.py -Q 4
python3 ./test.py -f 2-query/interval_limit_opt_2.py -Q 3
python3 ./test.py -f 2-query/interval_limit_opt_2.py -Q 2
python3 ./test.py -f 2-query/interval_limit_opt_2.py
python3 ./test.py -f 2-query/func_to_char_timestamp.py
python3 ./test.py -f 2-query/func_to_char_timestamp.py -Q 2
python3 ./test.py -f 2-query/func_to_char_timestamp.py -Q 3
python3 ./test.py -f 2-query/func_to_char_timestamp.py -Q 4
python3 ./test.py -f 2-query/last_cache_scan.py
python3 ./test.py -f 2-query/last_cache_scan.py -Q 2
python3 ./test.py -f 2-query/last_cache_scan.py -Q 3
python3 ./test.py -f 2-query/last_cache_scan.py -Q 4
python3 ./test.py -f 7-tmq/tmqShow.py
python3 ./test.py -f 7-tmq/tmqDropStb.py
python3 ./test.py -f 7-tmq/subscribeStb0.py
@ -29,11 +61,13 @@ python3 ./test.py -f 7-tmq/subscribeDb0.py -N 3 -n 3
python3 ./test.py -f 7-tmq/ins_topics_test.py
python3 ./test.py -f 7-tmq/tmqMaxTopic.py
python3 ./test.py -f 7-tmq/tmqParamsTest.py
python3 ./test.py -f 7-tmq/tmqParamsTest.py -R
python3 ./test.py -f 7-tmq/tmqClientConsLog.py
python3 ./test.py -f 7-tmq/tmqMaxGroupIds.py
python3 ./test.py -f 7-tmq/tmqConsumeDiscontinuousData.py
python3 ./test.py -f 7-tmq/tmqOffset.py
python3 ./test.py -f 7-tmq/tmqDropConsumer.py
python3 ./test.py -f 1-insert/insert_stb.py
python3 ./test.py -f 1-insert/delete_stable.py
python3 ./test.py -f 2-query/out_of_order.py -Q 3
python3 ./test.py -f 2-query/out_of_order.py
@ -61,10 +95,14 @@ python3 ./test.py -f 2-query/slimit.py -R
python3 ./test.py -f 2-query/slimit.py -Q 2
python3 ./test.py -f 2-query/slimit.py -Q 3
python3 ./test.py -f 2-query/slimit.py -Q 4
python3 ./test.py -f 3-enterprise/restore/restoreDnode.py -N 5 -M 3
python3 ./test.py -f 3-enterprise/restore/restoreVnode.py -N 5 -M 3
python3 ./test.py -f 3-enterprise/restore/restoreMnode.py -N 5 -M 3
python3 ./test.py -f 3-enterprise/restore/restoreQnode.py -N 5 -M 3
python3 ./test.py -f 2-query/ts-4233.py
python3 ./test.py -f 2-query/ts-4233.py -Q 2
python3 ./test.py -f 2-query/ts-4233.py -Q 3
python3 ./test.py -f 2-query/ts-4233.py -Q 4
python3 ./test.py -f 3-enterprise/restore/restoreDnode.py -N 5 -M 3 -i False
python3 ./test.py -f 3-enterprise/restore/restoreVnode.py -N 5 -M 3 -i False
python3 ./test.py -f 3-enterprise/restore/restoreMnode.py -N 5 -M 3 -i False
python3 ./test.py -f 3-enterprise/restore/restoreQnode.py -N 5 -M 3 -i False
python3 ./test.py -f 7-tmq/create_wrong_topic.py
python3 ./test.py -f 7-tmq/dropDbR3ConflictTransaction.py -N 3
python3 ./test.py -f 7-tmq/basic5.py
@ -112,16 +150,30 @@ python3 ./test.py -f 7-tmq/stbTagFilter-1ctb.py
python3 ./test.py -f 7-tmq/dataFromTsdbNWal.py
python3 ./test.py -f 7-tmq/dataFromTsdbNWal-multiCtb.py
python3 ./test.py -f 7-tmq/tmq_taosx.py
python3 ./test.py -f 7-tmq/tmq_replay.py
python3 ./test.py -f 7-tmq/tmqSeekAndCommit.py
python3 ./test.py -f 7-tmq/tmq_offset.py
python3 ./test.py -f 7-tmq/tmqDataPrecisionUnit.py
python3 ./test.py -f 7-tmq/raw_block_interface_test.py
python3 ./test.py -f 7-tmq/stbTagFilter-multiCtb.py
python3 ./test.py -f 7-tmq/tmqSubscribeStb-r3.py -N 5
python3 ./test.py -f 7-tmq/tmq3mnodeSwitch.py -N 6 -M 3
python3 ./test.py -f 7-tmq/tmq3mnodeSwitch.py -N 6 -M 3 -n 3
python3 ./test.py -f 7-tmq/tmq3mnodeSwitch.py -N 6 -M 3 -i True
python3 ./test.py -f 7-tmq/tmq3mnodeSwitch.py -N 6 -M 3 -n 3 -i True
python3 test.py -f 7-tmq/tmqVnodeTransform-stb.py -N 2 -n 1
python3 test.py -f 7-tmq/tmqVnodeTransform-stb.py -N 6 -n 3
python3 test.py -f 7-tmq/tmqVnodeSplit-stb-select.py -N 2 -n 1
python3 test.py -f 7-tmq/tmqVnodeSplit-stb-select-duplicatedata.py -N 3 -n 3
python3 test.py -f 7-tmq/tmqVnodeSplit-stb-select.py -N 3 -n 3
python3 test.py -f 7-tmq/tmqVnodeSplit-stb.py -N 3 -n 3
python3 test.py -f 7-tmq/tmqVnodeSplit-column.py -N 3 -n 3
python3 test.py -f 7-tmq/tmqVnodeSplit-db.py -N 3 -n 3
python3 test.py -f 7-tmq/tmqVnodeReplicate.py -M 3 -N 3 -n 3
python3 ./test.py -f 99-TDcase/TD-19201.py
python3 ./test.py -f 99-TDcase/TD-21561.py
python3 ./test.py -f 99-TDcase/TS-3404.py
python3 ./test.py -f 99-TDcase/TS-3581.py
python3 ./test.py -f 99-TDcase/TS-3311.py
python3 ./test.py -f 99-TDcase/TS-3821.py
python3 ./test.py -f 0-others/balance_vgroups_r1.py -N 6
python3 ./test.py -f 0-others/taosShell.py
python3 ./test.py -f 0-others/taosShellError.py
@ -139,10 +191,19 @@ python3 ./test.py -f 0-others/user_privilege_show.py
python3 ./test.py -f 0-others/user_privilege_all.py
python3 ./test.py -f 0-others/fsync.py
python3 ./test.py -f 0-others/multilevel.py
python3 ./test.py -f 0-others/ttl.py
python3 ./test.py -f 0-others/ttlChangeOnWrite.py
python3 ./test.py -f 0-others/compress_tsz1.py
python3 ./test.py -f 0-others/compress_tsz2.py
python3 ./test.py -f 0-others/view/non_marterial_view/test_view.py
python3 ./test.py -f 0-others/compatibility.py
python3 ./test.py -f 0-others/tag_index_basic.py
python3 ./test.py -N 3 -f 0-others/walRetention.py
python3 ./test.py -f 0-others/splitVGroupRep1.py -N 3
python3 ./test.py -f 0-others/splitVGroupRep3.py -N 3
python3 ./test.py -f 0-others/timeRangeWise.py -N 3
python3 ./test.py -f 0-others/delete_check.py
python3 ./test.py -f 0-others/test_hot_refresh_configurations.py
python3 ./test.py -f 1-insert/alter_database.py
python3 ./test.py -f 1-insert/alter_replica.py -N 3
python3 ./test.py -f 1-insert/influxdb_line_taosc_insert.py
@ -190,6 +251,7 @@ python3 ./test.py -f 1-insert/rowlength64k_4.py -Q 3
python3 ./test.py -f 1-insert/rowlength64k_4.py -Q 4
python3 ./test.py -f 1-insert/precisionUS.py
python3 ./test.py -f 1-insert/precisionNS.py
python3 ./test.py -f 1-insert/test_ts4219.py
python3 ./test.py -f 0-others/show.py
python3 ./test.py -f 0-others/show_tag_index.py
python3 ./test.py -f 0-others/information_schema.py
@ -325,6 +387,8 @@ python3 ./test.py -f 2-query/smaTest.py
python3 ./test.py -f 2-query/smaTest.py -R
python3 ./test.py -f 0-others/sma_index.py
python3 ./test.py -f 2-query/sml_TS-3724.py
python3 ./test.py -f 2-query/sml-TD19291.py
python3 ./test.py -f 2-query/varbinary.py
python3 ./test.py -f 2-query/sml.py
python3 ./test.py -f 2-query/sml.py -R
python3 ./test.py -f 2-query/spread.py
@ -426,7 +490,6 @@ python3 ./test.py -f 6-cluster/5dnode3mnodeRestartDnodeInsertData.py -N 6 -M 3
python3 ./test.py -f 6-cluster/5dnode3mnodeRestartDnodeInsertData.py -N 6 -M 3 -n 3
python3 ./test.py -f 6-cluster/5dnode3mnodeRestartDnodeInsertDataAsync.py -N 6 -M 3
python3 ./test.py -f 6-cluster/manually-test/6dnode3mnodeInsertLessDataAlterRep3to1to3.py -N 6 -M 3
python3 ./test.py -f 6-cluster/5dnode3mnodeRoll.py -N 3 -C 1
python3 ./test.py -f 6-cluster/5dnode3mnodeAdd1Ddnoe.py -N 7 -M 3 -C 6
python3 ./test.py -f 6-cluster/5dnode3mnodeAdd1Ddnoe.py -N 7 -M 3 -C 6 -n 3
python3 ./test.py -f 6-cluster/5dnode3mnodeRecreateMnode.py -N 6 -M 3
@ -731,7 +794,9 @@ python3 ./test.py -f 2-query/out_of_order.py -R
python3 ./test.py -f 2-query/blockSMA.py -Q 4
python3 ./test.py -f 2-query/projectionDesc.py -Q 4
python3 ./test.py -f 2-query/odbc.py
python3 ./test.py -f 2-query/fill_with_group.py
python3 ./test.py -f 99-TDcase/TD-21561.py -Q 4
python3 ./test.py -f 99-TDcase/TD-20582.py
python3 ./test.py -f 5-taos-tools/taosbenchmark/insertMix.py -N 3
python3 ./test.py -f 5-taos-tools/taosbenchmark/stt.py -N 3
python3 ./test.py -f eco-system/meta/database/keep_time_offset.py