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

Merge branch 'main' into FIX/TD-21043-main

This commit is contained in:
Shengliang Guan 2023-01-04 09:29:07 +08:00
parent 4031dc1af4 cae835bd86
commit 34260cd397
53 changed files with 888 additions and 337 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
cmake/cmake.define
View File

@ -117,12 +117,18 @@ ELSE ()
IF (${BUILD_SANITIZER})
SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Werror -Werror=return-type -fPIC -gdwarf-2 -fsanitize=address -fsanitize=undefined -fsanitize-recover=all -fsanitize=float-divide-by-zero -fsanitize=float-cast-overflow -fno-sanitize=shift-base -fno-sanitize=alignment -g3 -Wformat=0")
SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Werror -Wno-literal-suffix -Werror=return-type -fPIC -gdwarf-2 -fsanitize=address -fsanitize=undefined -fsanitize-recover=all -fsanitize=float-divide-by-zero -fsanitize=float-cast-overflow -fno-sanitize=shift-base -fno-sanitize=alignment -g3 -Wformat=0")
MESSAGE(STATUS "Will compile with Address Sanitizer!")
MESSAGE(STATUS "Compile with Address Sanitizer!")
ELSE ()
SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Werror -Werror=return-type -fPIC -gdwarf-2 -g3 -Wformat=2 -Wno-format-nonliteral -Wno-format-truncation -Wno-format-y2k")
SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Werror -Wno-literal-suffix -Werror=return-type -fPIC -gdwarf-2 -g3 -Wformat=2 -Wno-format-nonliteral -Wno-format-truncation -Wno-format-y2k")
ENDIF ()
# disable all assert
IF ((${DISABLE_ASSERT} MATCHES "true") OR (${DISABLE_ASSERTS} MATCHES "true"))
ADD_DEFINITIONS(-DDISABLE_ASSERT)
MESSAGE(STATUS "Disable all asserts")
ENDIF()
INCLUDE(CheckCCompilerFlag)
IF (TD_ARM_64 OR TD_ARM_32)
SET(COMPILER_SUPPORT_SSE42 false)
@ -155,7 +161,7 @@ ELSE ()
SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -mavx2")
SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -mavx2")
ENDIF()
MESSAGE(STATUS "SIMD instructions (AVX/AVX2) is ACTIVATED")
MESSAGE(STATUS "SIMD instructions (FMA/AVX/AVX2) is ACTIVATED")
ENDIF()
ENDIF ()

1
include/util/taoserror.h
View File

@ -345,6 +345,7 @@ int32_t* taosGetErrno();
#define TSDB_CODE_MND_TOPIC_SUBSCRIBED TAOS_DEF_ERROR_CODE(0, 0x03EB)
#define TSDB_CODE_MND_CGROUP_USED TAOS_DEF_ERROR_CODE(0, 0x03EC)
#define TSDB_CODE_MND_TOPIC_MUST_BE_DELETED TAOS_DEF_ERROR_CODE(0, 0x03ED)
#define TSDB_CODE_MND_INVALID_SUB_OPTION TAOS_DEF_ERROR_CODE(0, 0x03EE)
#define TSDB_CODE_MND_IN_REBALANCE TAOS_DEF_ERROR_CODE(0, 0x03EF)
// mnode-stream

7
include/util/tlog.h
View File

@ -85,12 +85,19 @@ void taosPrintLongString(const char *flags, ELogLevel level, int32_t dflag, cons
bool taosAssertDebug(bool condition, const char *file, int32_t line, const char *format, ...);
bool taosAssertRelease(bool condition);
// Disable all asserts that may compromise the performance.
#if defined DISABLE_ASSERT
#define ASSERT(condition)
#define ASSERTS(condition, ...)
#else
#define ASSERTS(condition, ...) taosAssertDebug(condition, __FILE__, __LINE__, __VA_ARGS__)
#ifdef NDEBUG
#define ASSERT(condition) taosAssertRelease(condition)
#else
#define ASSERT(condition) taosAssertDebug(condition, __FILE__, __LINE__, "assert info not provided")
#endif
#endif
// clang-format off
#define uFatal(...) { if (uDebugFlag & DEBUG_FATAL) { taosPrintLog("UTL FATAL", DEBUG_FATAL, tsLogEmbedded ? 255 : uDebugFlag, __VA_ARGS__); }}

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

@ -912,10 +912,12 @@ void tmqFreeImpl(void* handle) {
tmq_t* tmq = (tmq_t*)handle;
// TODO stop timer
tmqClearUnhandleMsg(tmq);
if (tmq->mqueue) taosCloseQueue(tmq->mqueue);
if (tmq->mqueue) {
tmqClearUnhandleMsg(tmq);
taosCloseQueue(tmq->mqueue);
}
if (tmq->delayedTask) taosCloseQueue(tmq->delayedTask);
if (tmq->qall) taosFreeQall(tmq->qall);
taosFreeQall(tmq->qall);
tsem_destroy(&tmq->rspSem);

12
source/dnode/mgmt/mgmt_snode/src/smWorker.c
View File

@ -58,11 +58,7 @@ static void smProcessStreamQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
dTrace("msg:%p, get from snode-stream queue", pMsg);
int32_t code = sndProcessStreamMsg(pMgmt->pSnode, pMsg);
if (code < 0) {
if (pMsg) {
dGError("snd, msg:%p failed to process stream msg %s since %s", pMsg, TMSG_INFO(pMsg->msgType), terrstr(code));
} else {
dGError("snd, msg:%p failed to process stream empty msg since %s", pMsg, terrstr(code));
}
dGError("snd, msg:%p failed to process stream msg %s since %s", pMsg, TMSG_INFO(pMsg->msgType), terrstr(code));
smSendRsp(pMsg, terrno);
}
@ -161,8 +157,10 @@ int32_t smPutMsgToQueue(SSnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) {
smPutNodeMsgToWriteQueue(pMgmt, pMsg);
break;
default:
ASSERTS(0, "msg:%p failed to put into snode queue since %s, type:%s qtype:%d", pMsg, terrstr(),
TMSG_INFO(pMsg->msgType), qtype);
terrno = TSDB_CODE_INVALID_PARA;
rpcFreeCont(pMsg->pCont);
taosFreeQitem(pMsg);
return -1;
}
return 0;
}

5
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
View File

@ -132,7 +132,7 @@ static void vmGenerateVnodeCfg(SCreateVnodeReq *pCreate, SVnodeCfg *pCfg) {
pCfg->syncCfg.myIndex = pCreate->selfIndex;
pCfg->syncCfg.replicaNum = pCreate->replica;
memset(&pCfg->syncCfg.nodeInfo, 0, sizeof(pCfg->syncCfg.nodeInfo));
for (int i = 0; i < pCreate->replica; ++i) {
for (int32_t i = 0; i < pCreate->replica; ++i) {
SNodeInfo *pNode = &pCfg->syncCfg.nodeInfo[i];
pNode->nodeId = pCreate->replicas[i].id;
pNode->nodePort = pCreate->replicas[i].port;
@ -288,7 +288,8 @@ int32_t vmProcessAlterVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
dInfo("vgId:%d, start to alter vnode, replica:%d selfIndex:%d strict:%d", alterReq.vgId, alterReq.replica,
alterReq.selfIndex, alterReq.strict);
for (int32_t i = 0; i < alterReq.replica; ++i) {
dInfo("vgId:%d, replica:%d ep:%s:%u", alterReq.vgId, i, alterReq.replicas[i].fqdn, alterReq.replicas[i].port);
SReplica *pReplica = &alterReq.replicas[i];
dInfo("vgId:%d, replica:%d ep:%s:%u dnode:%d", alterReq.vgId, i, pReplica->fqdn, pReplica->port, pReplica->port);
}
if (alterReq.replica <= 0 || alterReq.selfIndex < 0 || alterReq.selfIndex >= alterReq.replica) {

16
source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
View File

@ -86,12 +86,8 @@ static void vmProcessStreamQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
int32_t code = vnodeProcessFetchMsg(pVnode->pImpl, pMsg, pInfo);
if (code != 0) {
if (terrno != 0) code = terrno;
if (pMsg) {
dGError("vgId:%d, msg:%p failed to process stream msg %s since %s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType),
terrstr(code));
} else {
dGError("vgId:%d, msg:%p failed to process stream empty msg since %s", pVnode->vgId, pMsg, terrstr(code));
}
dGError("vgId:%d, msg:%p failed to process stream msg %s since %s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType),
terrstr(code));
vmSendRsp(pMsg, code);
}
@ -146,16 +142,16 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp
return -1;
}
SMsgHead *pHead = pMsg->pCont;
int32_t code = 0;
SMsgHead *pHead = pMsg->pCont;
int32_t code = 0;
pHead->contLen = ntohl(pHead->contLen);
pHead->vgId = ntohl(pHead->vgId);
SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId);
if (pVnode == NULL) {
dGError("vgId:%d, msg:%p failed to put into vnode queue since %s, type:%s qtype:%d contLen:%d", pHead->vgId, pMsg, terrstr(),
TMSG_INFO(pMsg->msgType), qtype, pHead->contLen);
dGError("vgId:%d, msg:%p failed to put into vnode queue since %s, type:%s qtype:%d contLen:%d", pHead->vgId, pMsg,
terrstr(), TMSG_INFO(pMsg->msgType), qtype, pHead->contLen);
return terrno != 0 ? terrno : -1;
}

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

@ -112,7 +112,6 @@ static SSdbRaw *mndDbActionEncode(SDbObj *pDb) {
SDB_SET_INT8(pRaw, dataPos, pDb->cfg.hashMethod, _OVER)
SDB_SET_INT32(pRaw, dataPos, pDb->cfg.numOfRetensions, _OVER)
for (int32_t i = 0; i < pDb->cfg.numOfRetensions; ++i) {
ASSERT(taosArrayGetSize(pDb->cfg.pRetensions) == pDb->cfg.numOfRetensions);
SRetention *pRetension = taosArrayGet(pDb->cfg.pRetensions, i);
SDB_SET_INT64(pRaw, dataPos, pRetension->freq, _OVER)
SDB_SET_INT64(pRaw, dataPos, pRetension->keep, _OVER)
@ -364,6 +363,7 @@ static int32_t mndCheckDbCfg(SMnode *pMnode, SDbCfg *pCfg) {
if (pCfg->hashPrefix < TSDB_MIN_HASH_PREFIX || pCfg->hashPrefix > TSDB_MAX_HASH_PREFIX) return -1;
if (pCfg->hashSuffix < TSDB_MIN_HASH_SUFFIX || pCfg->hashSuffix > TSDB_MAX_HASH_SUFFIX) return -1;
if (pCfg->tsdbPageSize < TSDB_MIN_TSDB_PAGESIZE || pCfg->tsdbPageSize > TSDB_MAX_TSDB_PAGESIZE) return -1;
if (taosArrayGetSize(pCfg->pRetensions) != pCfg->numOfRetensions) return -1;
terrno = 0;
return terrno;

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

@ -489,7 +489,7 @@ int32_t tEncodeSubscribeObj(void **buf, const SMqSubscribeObj *pSub) {
tlen += tEncodeSMqConsumerEp(buf, pConsumerEp);
cnt++;
}
ASSERT(cnt == sz);
if(cnt != sz) return -1;
tlen += taosEncodeArray(buf, pSub->unassignedVgs, (FEncode)tEncodeSMqVgEp);
tlen += taosEncodeString(buf, pSub->dbName);
return tlen;

1
source/dnode/mnode/impl/src/mndMnode.c
View File

@ -763,7 +763,6 @@ static void mndReloadSyncConfig(SMnode *pMnode) {
mInfo("vgId:1, mnode sync not reconfig since readyMnodes:%d updatingMnodes:%d", readyMnodes, updatingMnodes);
return;
}
// ASSERT(0);
if (cfg.myIndex == -1) {
#if 1

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

@ -111,7 +111,7 @@ static int32_t convertToRetrieveType(char *name, int32_t len) {
} else if (strncasecmp(name, TSDB_INS_TABLE_USER_PRIVILEGES, len) == 0) {
type = TSDB_MGMT_TABLE_PRIVILEGES;
} else {
// ASSERT(0);
mError("invalid show name:%s len:%d", name, len);
}
return type;

18
source/dnode/mnode/impl/src/mndSma.c
View File

@ -488,7 +488,7 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
memcpy(smaObj.db, pDb->name, TSDB_DB_FNAME_LEN);
smaObj.createdTime = taosGetTimestampMs();
smaObj.uid = mndGenerateUid(pCreate->name, TSDB_TABLE_FNAME_LEN);
ASSERT(smaObj.uid != 0);
char resultTbName[TSDB_TABLE_FNAME_LEN + 16] = {0};
snprintf(resultTbName, TSDB_TABLE_FNAME_LEN + 16, "%s_td_tsma_rst_tb", pCreate->name);
memcpy(smaObj.dstTbName, resultTbName, TSDB_TABLE_FNAME_LEN);
@ -558,13 +558,15 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
SNode *pAst = NULL;
if (nodesStringToNode(streamObj.ast, &pAst) < 0) {
ASSERT(0);
terrno = TSDB_CODE_MND_INVALID_SMA_OPTION;
mError("sma:%s, failed to create since parse ast error", smaObj.name);
return -1;
}
// extract output schema from ast
if (qExtractResultSchema(pAst, (int32_t *)&streamObj.outputSchema.nCols, &streamObj.outputSchema.pSchema) != 0) {
ASSERT(0);
terrno = TSDB_CODE_MND_INVALID_SMA_OPTION;
mError("sma:%s, failed to create since extract result schema error", smaObj.name);
return -1;
}
@ -579,15 +581,18 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
};
if (qCreateQueryPlan(&cxt, &pPlan, NULL) < 0) {
ASSERT(0);
terrno = TSDB_CODE_MND_INVALID_SMA_OPTION;
mError("sma:%s, failed to create since create query plan error", smaObj.name);
return -1;
}
// save physcial plan
if (nodesNodeToString((SNode *)pPlan, false, &streamObj.physicalPlan, NULL) != 0) {
ASSERT(0);
terrno = TSDB_CODE_MND_INVALID_SMA_OPTION;
mError("sma:%s, failed to create since save physcial plan error", smaObj.name);
return -1;
}
if (pAst != NULL) nodesDestroyNode(pAst);
nodesDestroyNode((SNode *)pPlan);
@ -826,14 +831,13 @@ static int32_t mndDropSma(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SSmaObj *p
if (mndDropStreamTasks(pMnode, pTrans, pStream) < 0) {
mError("stream:%s, failed to drop task since %s", pStream->name, terrstr());
sdbRelease(pMnode->pSdb, pStream);
ASSERT(0);
goto _OVER;
}
// drop stream
if (mndPersistDropStreamLog(pMnode, pTrans, pStream) < 0) {
mError("stream:%s, failed to drop log since %s", pStream->name, terrstr());
sdbRelease(pMnode->pSdb, pStream);
ASSERT(0);
goto _OVER;
}
}

16
source/dnode/mnode/impl/src/mndStb.c
View File

@ -1177,7 +1177,9 @@ static int32_t mndCheckAlterColForTopic(SMnode *pMnode, const char *stbFullName,
SNode *pAst = NULL;
if (nodesStringToNode(pTopic->ast, &pAst) != 0) {
ASSERT(0);
terrno = TSDB_CODE_MND_FIELD_CONFLICT_WITH_TOPIC;
mError("topic:%s, create ast error", pTopic->name);
sdbRelease(pSdb, pTopic);
return -1;
}
@ -1222,7 +1224,9 @@ static int32_t mndCheckAlterColForStream(SMnode *pMnode, const char *stbFullName
SNode *pAst = NULL;
if (nodesStringToNode(pStream->ast, &pAst) != 0) {
ASSERT(0);
terrno = TSDB_CODE_MND_INVALID_STREAM_OPTION;
mError("stream:%s, create ast error", pStream->name);
sdbRelease(pSdb, pStream);
return -1;
}
@ -2094,7 +2098,9 @@ static int32_t mndCheckDropStbForTopic(SMnode *pMnode, const char *stbFullName,
SNode *pAst = NULL;
if (nodesStringToNode(pTopic->ast, &pAst) != 0) {
ASSERT(0);
terrno = TSDB_CODE_MND_INVALID_TOPIC_OPTION;
mError("topic:%s, create ast error", pTopic->name);
sdbRelease(pSdb, pTopic);
return -1;
}
@ -2141,7 +2147,9 @@ static int32_t mndCheckDropStbForStream(SMnode *pMnode, const char *stbFullName,
SNode *pAst = NULL;
if (nodesStringToNode(pStream->ast, &pAst) != 0) {
ASSERT(0);
terrno = TSDB_CODE_MND_INVALID_STREAM_OPTION;
mError("stream:%s, create ast error", pStream->name);
sdbRelease(pSdb, pStream);
return -1;
}

90
source/dnode/mnode/impl/src/mndSubscribe.c
View File

@ -96,18 +96,12 @@ static SMqSubscribeObj *mndCreateSub(SMnode *pMnode, const SMqTopicObj *pTopic,
pSub->subType = pTopic->subType;
pSub->withMeta = pTopic->withMeta;
ASSERT(pSub->unassignedVgs->size == 0);
ASSERT(taosHashGetSize(pSub->consumerHash) == 0);
if (mndSchedInitSubEp(pMnode, pTopic, pSub) < 0) {
tDeleteSubscribeObj(pSub);
taosMemoryFree(pSub);
return NULL;
}
ASSERT(pSub->unassignedVgs->size > 0);
ASSERT(taosHashGetSize(pSub->consumerHash) == 0);
return pSub;
}
@ -144,7 +138,10 @@ static int32_t mndBuildSubChangeReq(void **pBuf, int32_t *pLen, const SMqSubscri
static int32_t mndPersistSubChangeVgReq(SMnode *pMnode, STrans *pTrans, const SMqSubscribeObj *pSub,
const SMqRebOutputVg *pRebVg) {
ASSERT(pRebVg->oldConsumerId != pRebVg->newConsumerId);
if (pRebVg->oldConsumerId == pRebVg->newConsumerId) {
terrno = TSDB_CODE_MND_INVALID_SUB_OPTION;
return -1;
}
void *buf;
int32_t tlen;
@ -155,8 +152,8 @@ static int32_t mndPersistSubChangeVgReq(SMnode *pMnode, STrans *pTrans, const SM
int32_t vgId = pRebVg->pVgEp->vgId;
SVgObj *pVgObj = mndAcquireVgroup(pMnode, vgId);
if (pVgObj == NULL) {
ASSERT(0);
taosMemoryFree(buf);
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
@ -206,9 +203,9 @@ static SMqRebInfo *mndGetOrCreateRebSub(SHashObj *pHash, const char *key) {
}
static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqRebOutputObj *pOutput) {
int32_t totalVgNum = pOutput->pSub->vgNum;
mInfo("mq rebalance: subscription: %s, vgNum: %d", pOutput->pSub->key, pOutput->pSub->vgNum);
int32_t totalVgNum = pOutput->pSub->vgNum;
const char *sub = pOutput->pSub->key;
mInfo("sub:%s, mq rebalance vgNum:%d", sub, pOutput->pSub->vgNum);
// 1. build temporary hash(vgId -> SMqRebOutputVg) to store modified vg
SHashObj *pHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);
@ -218,11 +215,10 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
int32_t actualRemoved = 0;
for (int32_t i = 0; i < removedNum; i++) {
int64_t consumerId = *(int64_t *)taosArrayGet(pInput->pRebInfo->removedConsumers, i);
ASSERT(consumerId > 0);
SMqConsumerEp *pConsumerEp = taosHashGet(pOutput->pSub->consumerHash, &consumerId, sizeof(int64_t));
ASSERT(pConsumerEp);
if (pConsumerEp) {
ASSERT(consumerId == pConsumerEp->consumerId);
actualRemoved++;
int32_t consumerVgNum = taosArrayGetSize(pConsumerEp->vgs);
for (int32_t j = 0; j < consumerVgNum; j++) {
@ -233,7 +229,7 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
.pVgEp = pVgEp,
};
taosHashPut(pHash, &pVgEp->vgId, sizeof(int32_t), &outputVg, sizeof(SMqRebOutputVg));
mInfo("mq rebalance: remove vgId:%d from consumer:%" PRId64, pVgEp->vgId, consumerId);
mInfo("sub:%s, mq rebalance remove vgId:%d from consumer:%" PRId64, sub, pVgEp->vgId, consumerId);
}
taosArrayDestroy(pConsumerEp->vgs);
taosHashRemove(pOutput->pSub->consumerHash, &consumerId, sizeof(int64_t));
@ -241,7 +237,10 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
taosArrayPush(pOutput->removedConsumers, &consumerId);
}
}
ASSERT(removedNum == actualRemoved);
if (removedNum != actualRemoved) {
mError("sub:%s, mq rebalance removedNum:%d not matched with actual:%d", sub, removedNum, actualRemoved);
}
// if previously no consumer, there are vgs not assigned
{
@ -254,7 +253,7 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
.pVgEp = pVgEp,
};
taosHashPut(pHash, &pVgEp->vgId, sizeof(int32_t), &rebOutput, sizeof(SMqRebOutputVg));
mInfo("mq rebalance: remove vgId:%d from unassigned", pVgEp->vgId);
mInfo("sub:%s, mq rebalance remove vgId:%d from unassigned", sub, pVgEp->vgId);
}
}
@ -268,8 +267,8 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
minVgCnt = totalVgNum / afterRebConsumerNum;
imbConsumerNum = totalVgNum % afterRebConsumerNum;
}
mInfo("mq rebalance: %d consumer after rebalance, at least %d vg each, %d consumer has more vg", afterRebConsumerNum,
minVgCnt, imbConsumerNum);
mInfo("sub:%s, mq rebalance %d consumer after rebalance, at least %d vg each, %d consumer has more vg", sub,
afterRebConsumerNum, minVgCnt, imbConsumerNum);
// 4. first scan: remove consumer more than wanted, put to remove hash
int32_t imbCnt = 0;
@ -278,7 +277,7 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
pIter = taosHashIterate(pOutput->pSub->consumerHash, pIter);
if (pIter == NULL) break;
SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIter;
ASSERT(pConsumerEp->consumerId > 0);
int32_t consumerVgNum = taosArrayGetSize(pConsumerEp->vgs);
// all old consumers still existing are touched
// TODO optimize: touch only consumer whose vgs changed
@ -298,7 +297,7 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
.pVgEp = pVgEp,
};
taosHashPut(pHash, &pVgEp->vgId, sizeof(int32_t), &outputVg, sizeof(SMqRebOutputVg));
mInfo("mq rebalance: remove vgId:%d from consumer:%" PRId64 ",(first scan)", pVgEp->vgId,
mInfo("sub:%s, mq rebalance remove vgId:%d from consumer:%" PRId64 ",(first scan)", sub, pVgEp->vgId,
pConsumerEp->consumerId);
}
imbCnt++;
@ -313,7 +312,7 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
.pVgEp = pVgEp,
};
taosHashPut(pHash, &pVgEp->vgId, sizeof(int32_t), &outputVg, sizeof(SMqRebOutputVg));
mInfo("mq rebalance: remove vgId:%d from consumer:%" PRId64 ",(first scan)", pVgEp->vgId,
mInfo("sub:%s, mq rebalance remove vgId:%d from consumer:%" PRId64 ",(first scan)", sub, pVgEp->vgId,
pConsumerEp->consumerId);
}
}
@ -325,13 +324,13 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
int32_t consumerNum = taosArrayGetSize(pInput->pRebInfo->newConsumers);
for (int32_t i = 0; i < consumerNum; i++) {
int64_t consumerId = *(int64_t *)taosArrayGet(pInput->pRebInfo->newConsumers, i);
ASSERT(consumerId > 0);
SMqConsumerEp newConsumerEp;
newConsumerEp.consumerId = consumerId;
newConsumerEp.vgs = taosArrayInit(0, sizeof(void *));
taosHashPut(pOutput->pSub->consumerHash, &consumerId, sizeof(int64_t), &newConsumerEp, sizeof(SMqConsumerEp));
taosArrayPush(pOutput->newConsumers, &consumerId);
mInfo("mq rebalance: add new consumer:%" PRId64, consumerId);
mInfo("sub:%s, mq rebalance add new consumer:%" PRId64, sub, consumerId);
}
}
@ -344,13 +343,16 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
pIter = taosHashIterate(pOutput->pSub->consumerHash, pIter);
if (pIter == NULL) break;
SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIter;
ASSERT(pConsumerEp->consumerId > 0);
// push until equal minVg
while (taosArrayGetSize(pConsumerEp->vgs) < minVgCnt) {
// iter hash and find one vg
pRemovedIter = taosHashIterate(pHash, pRemovedIter);
ASSERT(pRemovedIter);
if (pRemovedIter == NULL) {
mError("sub:%s, removed iter is null", sub);
continue;
}
pRebVg = (SMqRebOutputVg *)pRemovedIter;
// push
taosArrayPush(pConsumerEp->vgs, &pRebVg->pVgEp);
@ -361,7 +363,6 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
}
}
ASSERT(pIter == NULL);
// 7. handle unassigned vg
if (taosHashGetSize(pOutput->pSub->consumerHash) != 0) {
// if has consumer, assign all left vg
@ -377,9 +378,8 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
}
while (1) {
pIter = taosHashIterate(pOutput->pSub->consumerHash, pIter);
ASSERT(pIter);
pConsumerEp = (SMqConsumerEp *)pIter;
ASSERT(pConsumerEp->consumerId > 0);
if (taosArrayGetSize(pConsumerEp->vgs) == minVgCnt) {
break;
}
@ -404,19 +404,19 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
pIter = taosHashIterate(pHash, pIter);
if (pIter == NULL) break;
pRebOutput = (SMqRebOutputVg *)pIter;
ASSERT(pRebOutput->newConsumerId == -1);
taosArrayPush(pOutput->pSub->unassignedVgs, &pRebOutput->pVgEp);
taosArrayPush(pOutput->rebVgs, pRebOutput);
mInfo("mq rebalance: unassign vgId:%d (second scan)", pRebOutput->pVgEp->vgId);
mInfo("sub:%s, mq rebalance unassign vgId:%d (second scan)", sub, pRebOutput->pVgEp->vgId);
}
}
// 8. generate logs
mInfo("mq rebalance: calculation completed, rebalanced vg:");
mInfo("sub:%s, mq rebalance calculation completed, rebalanced vg", sub);
for (int32_t i = 0; i < taosArrayGetSize(pOutput->rebVgs); i++) {
SMqRebOutputVg *pOutputRebVg = taosArrayGet(pOutput->rebVgs, i);
mInfo("mq rebalance: vgId:%d, moved from consumer:%" PRId64 ", to consumer:%" PRId64, pOutputRebVg->pVgEp->vgId,
pOutputRebVg->oldConsumerId, pOutputRebVg->newConsumerId);
mInfo("sub:%s, mq rebalance vgId:%d, moved from consumer:%" PRId64 ", to consumer:%" PRId64, sub,
pOutputRebVg->pVgEp->vgId, pOutputRebVg->oldConsumerId, pOutputRebVg->newConsumerId);
}
{
void *pIter = NULL;
@ -425,10 +425,11 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
if (pIter == NULL) break;
SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIter;
int32_t sz = taosArrayGetSize(pConsumerEp->vgs);
mInfo("mq rebalance: final cfg: consumer %" PRId64 " has %d vg", pConsumerEp->consumerId, sz);
mInfo("sub:%s, mq rebalance final cfg: consumer %" PRId64 " has %d vg", sub, pConsumerEp->consumerId, sz);
for (int32_t i = 0; i < sz; i++) {
SMqVgEp *pVgEp = taosArrayGetP(pConsumerEp->vgs, i);
mInfo("mq rebalance: final cfg: vg %d to consumer %" PRId64 "", pVgEp->vgId, pConsumerEp->consumerId);
mInfo("sub:%s, mq rebalance final cfg: vg %d to consumer %" PRId64 "", sub, pVgEp->vgId,
pConsumerEp->consumerId);
}
}
}
@ -487,7 +488,7 @@ static int32_t mndPersistRebResult(SMnode *pMnode, SRpcMsg *pMsg, const SMqRebOu
consumerNum = taosArrayGetSize(pOutput->newConsumers);
for (int32_t i = 0; i < consumerNum; i++) {
int64_t consumerId = *(int64_t *)taosArrayGet(pOutput->newConsumers, i);
ASSERT(consumerId > 0);
SMqConsumerObj *pConsumerOld = mndAcquireConsumer(pMnode, consumerId);
SMqConsumerObj *pConsumerNew = tNewSMqConsumerObj(pConsumerOld->consumerId, pConsumerOld->cgroup);
pConsumerNew->updateType = CONSUMER_UPDATE__ADD;
@ -497,7 +498,6 @@ static int32_t mndPersistRebResult(SMnode *pMnode, SRpcMsg *pMsg, const SMqRebOu
taosArrayPush(pConsumerNew->rebNewTopics, &topic);
mndReleaseConsumer(pMnode, pConsumerOld);
if (mndSetConsumerCommitLogs(pMnode, pTrans, pConsumerNew) != 0) {
ASSERT(0);
tDeleteSMqConsumerObj(pConsumerNew);
taosMemoryFree(pConsumerNew);
goto REB_FAIL;
@ -510,7 +510,7 @@ static int32_t mndPersistRebResult(SMnode *pMnode, SRpcMsg *pMsg, const SMqRebOu
consumerNum = taosArrayGetSize(pOutput->removedConsumers);
for (int32_t i = 0; i < consumerNum; i++) {
int64_t consumerId = *(int64_t *)taosArrayGet(pOutput->removedConsumers, i);
ASSERT(consumerId > 0);
SMqConsumerObj *pConsumerOld = mndAcquireConsumer(pMnode, consumerId);
SMqConsumerObj *pConsumerNew = tNewSMqConsumerObj(pConsumerOld->consumerId, pConsumerOld->cgroup);
pConsumerNew->updateType = CONSUMER_UPDATE__REMOVE;
@ -520,7 +520,6 @@ static int32_t mndPersistRebResult(SMnode *pMnode, SRpcMsg *pMsg, const SMqRebOu
taosArrayPush(pConsumerNew->rebRemovedTopics, &topic);
mndReleaseConsumer(pMnode, pConsumerOld);
if (mndSetConsumerCommitLogs(pMnode, pTrans, pConsumerNew) != 0) {
ASSERT(0);
tDeleteSMqConsumerObj(pConsumerNew);
taosMemoryFree(pConsumerNew);
goto REB_FAIL;
@ -577,7 +576,6 @@ static int32_t mndProcessRebalanceReq(SRpcMsg *pMsg) {
char cgroup[TSDB_CGROUP_LEN];
mndSplitSubscribeKey(pRebInfo->key, topic, cgroup, true);
SMqTopicObj *pTopic = mndAcquireTopic(pMnode, topic);
/*ASSERT(pTopic);*/
if (pTopic == NULL) {
mError("mq rebalance %s failed since topic %s not exist, abort", pRebInfo->key, topic);
continue;
@ -585,8 +583,14 @@ static int32_t mndProcessRebalanceReq(SRpcMsg *pMsg) {
taosRLockLatch(&pTopic->lock);
rebOutput.pSub = mndCreateSub(pMnode, pTopic, pRebInfo->key);
if (rebOutput.pSub == NULL) {
mError("mq rebalance %s failed create sub since %s, abort", pRebInfo->key, terrstr());
taosRUnLockLatch(&pTopic->lock);
mndReleaseTopic(pMnode, pTopic);
continue;
}
memcpy(rebOutput.pSub->dbName, pTopic->db, TSDB_DB_FNAME_LEN);
ASSERT(taosHashGetSize(rebOutput.pSub->consumerHash) == 0);
taosRUnLockLatch(&pTopic->lock);
mndReleaseTopic(pMnode, pTopic);
@ -606,7 +610,6 @@ static int32_t mndProcessRebalanceReq(SRpcMsg *pMsg) {
// if add more consumer to balanced subscribe,
// possibly no vg is changed
/*ASSERT(taosArrayGetSize(rebOutput.rebVgs) != 0);*/
if (mndPersistRebResult(pMnode, pMsg, &rebOutput) < 0) {
mError("mq rebalance persist rebalance output error, possibly vnode splitted or dropped");
@ -693,6 +696,7 @@ static SSdbRaw *mndSubActionEncode(SMqSubscribeObj *pSub) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
void *buf = NULL;
int32_t tlen = tEncodeSubscribeObj(NULL, pSub);
if (tlen <= 0) goto SUB_ENCODE_OVER;
int32_t size = sizeof(int32_t) + tlen + MND_SUBSCRIBE_RESERVE_SIZE;
SSdbRaw *pRaw = sdbAllocRaw(SDB_SUBSCRIBE, MND_SUBSCRIBE_VER_NUMBER, size);

8
source/dnode/mnode/impl/src/mndTopic.c
View File

@ -384,7 +384,11 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
topicObj.subType = pCreate->subType;
topicObj.withMeta = pCreate->withMeta;
if (topicObj.withMeta) {
ASSERT(topicObj.subType != TOPIC_SUB_TYPE__COLUMN);
if (topicObj.subType == TOPIC_SUB_TYPE__COLUMN) {
terrno = TSDB_CODE_MND_INVALID_TOPIC_OPTION;
mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
return -1;
}
}
if (pCreate->subType == TOPIC_SUB_TYPE__COLUMN) {
@ -499,7 +503,6 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
if (code < 0) {
sdbRelease(pSdb, pVgroup);
mndTransDrop(pTrans);
ASSERT(0);
return -1;
}
void *buf = taosMemoryCalloc(1, sizeof(SMsgHead) + len);
@ -723,7 +726,6 @@ static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {
// TODO check if rebalancing
if (mndDropSubByTopic(pMnode, pTrans, dropReq.name) < 0) {
/*ASSERT(0);*/
mError("topic:%s, failed to drop since %s", pTopic->name, terrstr());
mndTransDrop(pTrans);
mndReleaseTopic(pMnode, pTopic);

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

@ -772,8 +772,8 @@ static FORCE_INLINE int32_t tsdbKeyCmprFn(const void *p1, const void *p2) {
return 0;
}
#define SL_NODE_FORWARD(n, l) ((n)->forwards[l])
#define SL_NODE_BACKWARD(n, l) ((n)->forwards[(n)->level + (l)])
// #define SL_NODE_FORWARD(n, l) ((n)->forwards[l])
// #define SL_NODE_BACKWARD(n, l) ((n)->forwards[(n)->level + (l)])
static FORCE_INLINE TSDBROW *tsdbTbDataIterGet(STbDataIter *pIter) {
if (pIter == NULL) return NULL;

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

@ -284,8 +284,7 @@ int32_t tqMetaRestoreHandle(STQ* pTq) {
handle.pRef = walOpenRef(pTq->pVnode->pWal);
if (handle.pRef == NULL) {
ASSERT(0);
return -1;
continue;
}
walRefVer(handle.pRef, handle.snapshotVer);

28
source/dnode/vnode/src/tq/tqOffset.c
View File

@ -46,21 +46,37 @@ int32_t tqOffsetRestoreFromFile(STqOffsetStore* pStore, const char* fname) {
}
int32_t size = htonl(head.size);
void* memBuf = taosMemoryCalloc(1, size);
if (memBuf == NULL) {
return -1;
}
if ((code = taosReadFile(pFile, memBuf, size)) != size) {
ASSERT(0);
// TODO handle error
taosMemoryFree(memBuf);
return -1;
}
STqOffset offset;
SDecoder decoder;
tDecoderInit(&decoder, memBuf, size);
if (tDecodeSTqOffset(&decoder, &offset) < 0) {
ASSERT(0);
taosMemoryFree(memBuf);
tDecoderClear(&decoder);
return -1;
}
tDecoderClear(&decoder);
if (taosHashPut(pStore->pHash, offset.subKey, strlen(offset.subKey), &offset, sizeof(STqOffset)) < 0) {
ASSERT(0);
// TODO
return -1;
}
if (offset.val.type == TMQ_OFFSET__LOG) {
STqHandle* pHandle = taosHashGet(pStore->pTq->pHandle, offset.subKey, strlen(offset.subKey));
if (pHandle) {
if (walRefVer(pHandle->pRef, offset.val.version) < 0) {
tqError("vgId: %d, tq handle %s ref ver %" PRId64 "error", pStore->pTq->pVnode->config.vgId,
pHandle->subKey, offset.val.version);
}
}
}
taosMemoryFree(memBuf);
}
@ -124,7 +140,7 @@ int32_t tqOffsetCommitFile(STqOffsetStore* pStore) {
const char* sysErrStr = strerror(errno);
tqError("vgId:%d, cannot open file %s when commit offset since %s", pStore->pTq->pVnode->config.vgId, fname,
sysErrStr);
ASSERT(0);
taosMemoryFree(fname);
return -1;
}
taosMemoryFree(fname);

2
source/dnode/vnode/src/tq/tqSink.c
View File

@ -64,7 +64,7 @@ int32_t tqBuildDeleteReq(SVnode* pVnode, const char* stbFullName, const SSDataBl
.startTs = startTs,
.endTs = endTs,
};
strncpy(req.tbname, name, TSDB_TABLE_NAME_LEN);
strncpy(req.tbname, name, TSDB_TABLE_NAME_LEN - 1);
taosMemoryFree(name);
/*tqDebug("stream delete msg, active: vgId:%d, ts:%" PRId64 " name:%s", pVnode->config.vgId, ts, name);*/
taosArrayPush(deleteReq->deleteReqs, &req);

4
source/dnode/vnode/src/tq/tqSnapshot.c
View File

@ -175,6 +175,8 @@ int32_t tqSnapWriterClose(STqSnapWriter** ppWriter, int8_t rollback) {
if (code) goto _err;
}
int vgId = TD_VID(pWriter->pTq->pVnode);
taosMemoryFree(pWriter);
*ppWriter = NULL;
@ -186,7 +188,7 @@ int32_t tqSnapWriterClose(STqSnapWriter** ppWriter, int8_t rollback) {
return code;
_err:
tqError("vgId:%d, tq snapshot writer close failed since %s", TD_VID(pWriter->pTq->pVnode), tstrerror(code));
tqError("vgId:%d, tq snapshot writer close failed since %s", vgId, tstrerror(code));
return code;
}

97
source/dnode/vnode/src/tsdb/tsdbMemTable.c
View File

@ -19,9 +19,13 @@
#define SL_MAX_LEVEL 5
// sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l) * 2
#define SL_NODE_SIZE(l) (sizeof(SMemSkipListNode) + ((l) << 4))
#define SL_NODE_FORWARD(n, l) ((n)->forwards[l])
#define SL_NODE_BACKWARD(n, l) ((n)->forwards[(n)->level + (l)])
#define SL_NODE_SIZE(l) (sizeof(SMemSkipListNode) + ((l) << 4))
#define SL_NODE_FORWARD(n, l) ((n)->forwards[l])
#define SL_NODE_BACKWARD(n, l) ((n)->forwards[(n)->level + (l)])
#define SL_GET_NODE_FORWARD(n, l) ((SMemSkipListNode *)atomic_load_64((int64_t *)&SL_NODE_FORWARD(n, l)))
#define SL_GET_NODE_BACKWARD(n, l) ((SMemSkipListNode *)atomic_load_64((int64_t *)&SL_NODE_BACKWARD(n, l)))
#define SL_SET_NODE_FORWARD(n, l, p) atomic_store_64((int64_t *)&SL_NODE_FORWARD(n, l), (int64_t)(p))
#define SL_SET_NODE_BACKWARD(n, l, p) atomic_store_64((int64_t *)&SL_NODE_BACKWARD(n, l), (int64_t)(p))
#define SL_MOVE_BACKWARD 0x1
#define SL_MOVE_FROM_POS 0x2
@ -246,18 +250,18 @@ void tsdbTbDataIterOpen(STbData *pTbData, TSDBKEY *pFrom, int8_t backward, STbDa
if (pFrom == NULL) {
// create from head or tail
if (backward) {
pIter->pNode = SL_NODE_BACKWARD(pTbData->sl.pTail, 0);
pIter->pNode = SL_GET_NODE_BACKWARD(pTbData->sl.pTail, 0);
} else {
pIter->pNode = SL_NODE_FORWARD(pTbData->sl.pHead, 0);
pIter->pNode = SL_GET_NODE_FORWARD(pTbData->sl.pHead, 0);
}
} else {
// create from a key
if (backward) {
tbDataMovePosTo(pTbData, pos, pFrom, SL_MOVE_BACKWARD);
pIter->pNode = SL_NODE_BACKWARD(pos[0], 0);
pIter->pNode = SL_GET_NODE_BACKWARD(pos[0], 0);
} else {
tbDataMovePosTo(pTbData, pos, pFrom, 0);
pIter->pNode = SL_NODE_FORWARD(pos[0], 0);
pIter->pNode = SL_GET_NODE_FORWARD(pos[0], 0);
}
}
}
@ -271,7 +275,7 @@ bool tsdbTbDataIterNext(STbDataIter *pIter) {
return false;
}
pIter->pNode = SL_NODE_BACKWARD(pIter->pNode, 0);
pIter->pNode = SL_GET_NODE_BACKWARD(pIter->pNode, 0);
if (pIter->pNode == pIter->pTbData->sl.pHead) {
return false;
}
@ -282,7 +286,7 @@ bool tsdbTbDataIterNext(STbDataIter *pIter) {
return false;
}
pIter->pNode = SL_NODE_FORWARD(pIter->pNode, 0);
pIter->pNode = SL_GET_NODE_FORWARD(pIter->pNode, 0);
if (pIter->pNode == pIter->pTbData->sl.pTail) {
return false;
}
@ -335,7 +339,7 @@ static int32_t tsdbGetOrCreateTbData(SMemTable *pMemTable, tb_uid_t suid, tb_uid
int8_t maxLevel = pMemTable->pTsdb->pVnode->config.tsdbCfg.slLevel;
ASSERT(pPool != NULL);
pTbData = vnodeBufPoolMalloc(pPool, sizeof(*pTbData) + SL_NODE_SIZE(maxLevel) * 2);
pTbData = vnodeBufPoolMallocAligned(pPool, sizeof(*pTbData) + SL_NODE_SIZE(maxLevel) * 2);
if (pTbData == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
@ -408,7 +412,7 @@ static void tbDataMovePosTo(STbData *pTbData, SMemSkipListNode **pos, TSDBKEY *p
if (fromPos) px = pos[pTbData->sl.level - 1];
for (int8_t iLevel = pTbData->sl.level - 1; iLevel >= 0; iLevel--) {
pn = SL_NODE_BACKWARD(px, iLevel);
pn = SL_GET_NODE_BACKWARD(px, iLevel);
while (pn != pTbData->sl.pHead) {
tKey.version = pn->version;
tKey.ts = pn->pTSRow->ts;
@ -418,7 +422,7 @@ static void tbDataMovePosTo(STbData *pTbData, SMemSkipListNode **pos, TSDBKEY *p
break;
} else {
px = pn;
pn = SL_NODE_BACKWARD(px, iLevel);
pn = SL_GET_NODE_BACKWARD(px, iLevel);
}
}
@ -438,7 +442,7 @@ static void tbDataMovePosTo(STbData *pTbData, SMemSkipListNode **pos, TSDBKEY *p
if (fromPos) px = pos[pTbData->sl.level - 1];
for (int8_t iLevel = pTbData->sl.level - 1; iLevel >= 0; iLevel--) {
pn = SL_NODE_FORWARD(px, iLevel);
pn = SL_GET_NODE_FORWARD(px, iLevel);
while (pn != pTbData->sl.pTail) {
tKey.version = pn->version;
tKey.ts = pn->pTSRow->ts;
@ -448,7 +452,7 @@ static void tbDataMovePosTo(STbData *pTbData, SMemSkipListNode **pos, TSDBKEY *p
break;
} else {
px = pn;
pn = SL_NODE_FORWARD(px, iLevel);
pn = SL_GET_NODE_FORWARD(px, iLevel);
}
}
@ -474,58 +478,53 @@ static int32_t tbDataDoPut(SMemTable *pMemTable, STbData *pTbData, SMemSkipListN
int8_t level;
SMemSkipListNode *pNode;
SVBufPool *pPool = pMemTable->pTsdb->pVnode->inUse;
int64_t nSize;
// node
// create node
level = tsdbMemSkipListRandLevel(&pTbData->sl);
ASSERT(pPool != NULL);
pNode = (SMemSkipListNode *)vnodeBufPoolMalloc(pPool, SL_NODE_SIZE(level));
nSize = SL_NODE_SIZE(level);
pNode = (SMemSkipListNode *)vnodeBufPoolMallocAligned(pPool, nSize + pRow->len);
if (pNode == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
pNode->level = level;
pNode->version = version;
pNode->pTSRow = vnodeBufPoolMalloc(pPool, pRow->len);
if (NULL == pNode->pTSRow) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
pNode->pTSRow = (STSRow *)((char *)pNode + nSize);
memcpy(pNode->pTSRow, pRow, pRow->len);
for (int8_t iLevel = level - 1; iLevel >= 0; iLevel--) {
SMemSkipListNode *pn = pos[iLevel];
SMemSkipListNode *px;
if (forward) {
px = SL_NODE_FORWARD(pn, iLevel);
SL_NODE_BACKWARD(pNode, iLevel) = pn;
SL_NODE_FORWARD(pNode, iLevel) = px;
} else {
px = SL_NODE_BACKWARD(pn, iLevel);
SL_NODE_BACKWARD(pNode, iLevel) = px;
SL_NODE_FORWARD(pNode, iLevel) = pn;
// set node
if (forward) {
for (int8_t iLevel = 0; iLevel < level; iLevel++) {
SL_NODE_FORWARD(pNode, iLevel) = SL_NODE_FORWARD(pos[iLevel], iLevel);
SL_NODE_BACKWARD(pNode, iLevel) = pos[iLevel];
}
} else {
for (int8_t iLevel = 0; iLevel < level; iLevel++) {
SL_NODE_FORWARD(pNode, iLevel) = pos[iLevel];
SL_NODE_BACKWARD(pNode, iLevel) = SL_NODE_BACKWARD(pos[iLevel], iLevel);
}
}
for (int8_t iLevel = level - 1; iLevel >= 0; iLevel--) {
SMemSkipListNode *pn = pos[iLevel];
SMemSkipListNode *px;
// set forward and backward
if (forward) {
for (int8_t iLevel = level - 1; iLevel >= 0; iLevel--) {
SMemSkipListNode *pNext = pos[iLevel]->forwards[iLevel];
if (forward) {
px = SL_NODE_FORWARD(pn, iLevel);
SL_SET_NODE_FORWARD(pos[iLevel], iLevel, pNode);
SL_SET_NODE_BACKWARD(pNext, iLevel, pNode);
SL_NODE_FORWARD(pn, iLevel) = pNode;
SL_NODE_BACKWARD(px, iLevel) = pNode;
} else {
px = SL_NODE_BACKWARD(pn, iLevel);
SL_NODE_FORWARD(px, iLevel) = pNode;
SL_NODE_BACKWARD(pn, iLevel) = pNode;
pos[iLevel] = pNode;
}
} else {
for (int8_t iLevel = level - 1; iLevel >= 0; iLevel--) {
SMemSkipListNode *pPrev = pos[iLevel]->forwards[pos[iLevel]->level + iLevel];
pos[iLevel] = pNode;
SL_SET_NODE_FORWARD(pPrev, iLevel, pNode);
SL_SET_NODE_BACKWARD(pos[iLevel], iLevel, pNode);
pos[iLevel] = pNode;
}
}
pTbData->sl.size++;

107
source/dnode/vnode/src/tsdb/tsdbSnapshot.c
View File

@ -76,6 +76,7 @@ static int32_t tFDataIterCmprFn(const SRBTreeNode* pNode1, const SRBTreeNode* pN
static int32_t tsdbSnapReadOpenFile(STsdbSnapReader* pReader) {
int32_t code = 0;
int32_t lino = 0;
SDFileSet dFileSet = {.fid = pReader->fid};
SDFileSet* pSet = taosArraySearch(pReader->fs.aDFileSet, &dFileSet, tDFileSetCmprFn, TD_GT);
@ -83,7 +84,7 @@ static int32_t tsdbSnapReadOpenFile(STsdbSnapReader* pReader) {
pReader->fid = pSet->fid;
code = tsdbDataFReaderOpen(&pReader->pDataFReader, pReader->pTsdb, pSet);
if (code) goto _err;
TSDB_CHECK_CODE(code, lino, _exit);
pReader->pIter = NULL;
tRBTreeCreate(&pReader->rbt, tFDataIterCmprFn);
@ -93,13 +94,13 @@ static int32_t tsdbSnapReadOpenFile(STsdbSnapReader* pReader) {
pIter->type = SNAP_DATA_FILE_ITER;
code = tsdbReadBlockIdx(pReader->pDataFReader, pIter->aBlockIdx);
if (code) goto _err;
TSDB_CHECK_CODE(code, lino, _exit);
for (pIter->iBlockIdx = 0; pIter->iBlockIdx < taosArrayGetSize(pIter->aBlockIdx); pIter->iBlockIdx++) {
pIter->pBlockIdx = (SBlockIdx*)taosArrayGet(pIter->aBlockIdx, pIter->iBlockIdx);
code = tsdbReadDataBlk(pReader->pDataFReader, pIter->pBlockIdx, &pIter->mBlock);
if (code) goto _err;
TSDB_CHECK_CODE(code, lino, _exit);
for (pIter->iBlock = 0; pIter->iBlock < pIter->mBlock.nItem; pIter->iBlock++) {
SDataBlk dataBlk;
@ -108,7 +109,7 @@ static int32_t tsdbSnapReadOpenFile(STsdbSnapReader* pReader) {
if (dataBlk.minVer > pReader->ever || dataBlk.maxVer < pReader->sver) continue;
code = tsdbReadDataBlockEx(pReader->pDataFReader, &dataBlk, &pIter->bData);
if (code) goto _err;
TSDB_CHECK_CODE(code, lino, _exit);
ASSERT(pIter->pBlockIdx->suid == pIter->bData.suid);
ASSERT(pIter->pBlockIdx->uid == pIter->bData.uid);
@ -139,7 +140,7 @@ static int32_t tsdbSnapReadOpenFile(STsdbSnapReader* pReader) {
pIter->iStt = iStt;
code = tsdbReadSttBlk(pReader->pDataFReader, iStt, pIter->aSttBlk);
if (code) goto _err;
TSDB_CHECK_CODE(code, lino, _exit);
for (pIter->iSttBlk = 0; pIter->iSttBlk < taosArrayGetSize(pIter->aSttBlk); pIter->iSttBlk++) {
SSttBlk* pSttBlk = (SSttBlk*)taosArrayGet(pIter->aSttBlk, pIter->iSttBlk);
@ -148,7 +149,7 @@ static int32_t tsdbSnapReadOpenFile(STsdbSnapReader* pReader) {
if (pSttBlk->maxVer < pReader->sver) continue;
code = tsdbReadSttBlockEx(pReader->pDataFReader, iStt, pSttBlk, &pIter->bData);
if (code) goto _err;
TSDB_CHECK_CODE(code, lino, _exit);
for (pIter->iRow = 0; pIter->iRow < pIter->bData.nRow; pIter->iRow++) {
int64_t rowVer = pIter->bData.aVersion[pIter->iRow];
@ -169,13 +170,13 @@ static int32_t tsdbSnapReadOpenFile(STsdbSnapReader* pReader) {
pIter++;
}
tsdbInfo("vgId:%d, vnode snapshot tsdb open data file to read for %s, fid:%d", TD_VID(pReader->pTsdb->pVnode),
pReader->pTsdb->path, pReader->fid);
return code;
_err:
tsdbError("vgId:%d, vnode snapshot tsdb snap read open file failed since %s", TD_VID(pReader->pTsdb->pVnode),
tstrerror(code));
_exit:
if (code) {
tsdbError("vgId:%d, %s failed since %s", TD_VID(pReader->pTsdb->pVnode), __func__, tstrerror(code));
} else {
tsdbInfo("vgId:%d, %s done, path:%s, fid:%d", TD_VID(pReader->pTsdb->pVnode), __func__, pReader->pTsdb->path,
pReader->fid);
}
return code;
}
@ -318,12 +319,14 @@ _exit:
static int32_t tsdbSnapReadData(STsdbSnapReader* pReader, uint8_t** ppData) {
int32_t code = 0;
STsdb* pTsdb = pReader->pTsdb;
int32_t lino = 0;
STsdb* pTsdb = pReader->pTsdb;
while (true) {
if (pReader->pDataFReader == NULL) {
code = tsdbSnapReadOpenFile(pReader);
if (code) goto _err;
TSDB_CHECK_CODE(code, lino, _exit);
}
if (pReader->pDataFReader == NULL) break;
@ -338,17 +341,17 @@ static int32_t tsdbSnapReadData(STsdbSnapReader* pReader, uint8_t** ppData) {
SBlockData* pBlockData = &pReader->bData;
code = tsdbUpdateTableSchema(pTsdb->pVnode->pMeta, id.suid, id.uid, &pReader->skmTable);
if (code) goto _err;
TSDB_CHECK_CODE(code, lino, _exit);
code = tBlockDataInit(pBlockData, &id, pReader->skmTable.pTSchema, NULL, 0);
if (code) goto _err;
TSDB_CHECK_CODE(code, lino, _exit);
while (pRowInfo->suid == id.suid && pRowInfo->uid == id.uid) {
code = tBlockDataAppendRow(pBlockData, &pRowInfo->row, NULL, pRowInfo->uid);
if (code) goto _err;
TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbSnapNextRow(pReader);
if (code) goto _err;
TSDB_CHECK_CODE(code, lino, _exit);
pRowInfo = tsdbSnapGetRow(pReader);
if (pRowInfo == NULL) {
@ -360,21 +363,22 @@ static int32_t tsdbSnapReadData(STsdbSnapReader* pReader, uint8_t** ppData) {
}
code = tsdbSnapCmprData(pReader, ppData);
if (code) goto _err;
TSDB_CHECK_CODE(code, lino, _exit);
break;
}
return code;
_err:
tsdbError("vgId:%d, vnode snapshot tsdb read data for %s failed since %s", TD_VID(pTsdb->pVnode), pTsdb->path,
tstrerror(code));
_exit:
if (code) {
tsdbError("vgId:%d, %s failed since %s, path:%s", TD_VID(pTsdb->pVnode), __func__, tstrerror(code), pTsdb->path);
}
return code;
}
static int32_t tsdbSnapReadDel(STsdbSnapReader* pReader, uint8_t** ppData) {
int32_t code = 0;
int32_t code = 0;
int32_t lino = 0;
STsdb* pTsdb = pReader->pTsdb;
SDelFile* pDelFile = pReader->fs.pDelFile;
@ -385,11 +389,11 @@ static int32_t tsdbSnapReadDel(STsdbSnapReader* pReader, uint8_t** ppData) {
// open
code = tsdbDelFReaderOpen(&pReader->pDelFReader, pDelFile, pTsdb);
if (code) goto _err;
TSDB_CHECK_CODE(code, lino, _exit);
// read index
code = tsdbReadDelIdx(pReader->pDelFReader, pReader->aDelIdx);
if (code) goto _err;
TSDB_CHECK_CODE(code, lino, _exit);
pReader->iDelIdx = 0;
}
@ -405,7 +409,7 @@ static int32_t tsdbSnapReadDel(STsdbSnapReader* pReader, uint8_t** ppData) {
pReader->iDelIdx++;
code = tsdbReadDelData(pReader->pDelFReader, pDelIdx, pReader->aDelData);
if (code) goto _err;
TSDB_CHECK_CODE(code, lino, _exit);
int32_t size = 0;
for (int32_t iDelData = 0; iDelData < taosArrayGetSize(pReader->aDelData); iDelData++) {
@ -422,7 +426,7 @@ static int32_t tsdbSnapReadDel(STsdbSnapReader* pReader, uint8_t** ppData) {
*ppData = taosMemoryMalloc(sizeof(SSnapDataHdr) + size);
if (*ppData == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
TSDB_CHECK_CODE(code, lino, _exit);
}
SSnapDataHdr* pHdr = (SSnapDataHdr*)(*ppData);
@ -449,11 +453,9 @@ static int32_t tsdbSnapReadDel(STsdbSnapReader* pReader, uint8_t** ppData) {
}
_exit:
return code;
_err:
tsdbError("vgId:%d, vnode snapshot tsdb read del for %s failed since %s", TD_VID(pTsdb->pVnode), pTsdb->path,
tstrerror(code));
if (code) {
tsdbError("vgId:%d, %s failed since %s, path:%s", TD_VID(pTsdb->pVnode), __func__, tstrerror(code), pTsdb->path);
}
return code;
}
@ -591,44 +593,39 @@ int32_t tsdbSnapReaderClose(STsdbSnapReader** ppReader) {
int32_t tsdbSnapRead(STsdbSnapReader* pReader, uint8_t** ppData) {
int32_t code = 0;
int32_t lino = 0;
*ppData = NULL;
// read data file
if (!pReader->dataDone) {
code = tsdbSnapReadData(pReader, ppData);
if (code) {
goto _err;
TSDB_CHECK_CODE(code, lino, _exit);
if (*ppData) {
goto _exit;
} else {
if (*ppData) {
goto _exit;
} else {
pReader->dataDone = 1;
}
pReader->dataDone = 1;
}
}
// read del file
if (!pReader->delDone) {
code = tsdbSnapReadDel(pReader, ppData);
if (code) {
goto _err;
TSDB_CHECK_CODE(code, lino, _exit);
if (*ppData) {
goto _exit;
} else {
if (*ppData) {
goto _exit;
} else {
pReader->delDone = 1;
}
pReader->delDone = 1;
}
}
_exit:
tsdbDebug("vgId:%d, vnode snapshot tsdb read for %s", TD_VID(pReader->pTsdb->pVnode), pReader->pTsdb->path);
return code;
_err:
tsdbError("vgId:%d, vnode snapshot tsdb read for %s failed since %s", TD_VID(pReader->pTsdb->pVnode),
pReader->pTsdb->path, tstrerror(code));
if (code) {
tsdbError("vgId:%d, %s failed since %s, path:%s", TD_VID(pReader->pTsdb->pVnode), __func__, tstrerror(code),
pReader->pTsdb->path);
} else {
tsdbDebug("vgId:%d, %s done, path:%s", TD_VID(pReader->pTsdb->pVnode), __func__, pReader->pTsdb->path);
}
return code;
}

30
source/dnode/vnode/src/vnd/vnodeCfg.c
View File

@ -128,14 +128,19 @@ int vnodeEncodeConfig(const void *pObj, SJson *pJson) {
SJson *nodeInfo = tjsonCreateArray();
if (nodeInfo == NULL) return -1;
if (tjsonAddItemToObject(pJson, "syncCfg.nodeInfo", nodeInfo) < 0) return -1;
vDebug("vgId:%d, encode config, replicas:%d selfIndex:%d", pCfg->vgId, pCfg->syncCfg.replicaNum,
pCfg->syncCfg.myIndex);
for (int i = 0; i < pCfg->syncCfg.replicaNum; ++i) {
SJson *info = tjsonCreateObject();
SJson *info = tjsonCreateObject();
SNodeInfo *pNode = (SNodeInfo *)&pCfg->syncCfg.nodeInfo[i];
if (info == NULL) return -1;
if (tjsonAddIntegerToObject(info, "nodePort", pCfg->syncCfg.nodeInfo[i].nodePort) < 0) return -1;
if (tjsonAddStringToObject(info, "nodeFqdn", pCfg->syncCfg.nodeInfo[i].nodeFqdn) < 0) return -1;
if (tjsonAddIntegerToObject(info, "nodeId", pCfg->syncCfg.nodeInfo[i].nodeId) < 0) return -1;
if (tjsonAddIntegerToObject(info, "clusterId", pCfg->syncCfg.nodeInfo[i].clusterId) < 0) return -1;
if (tjsonAddIntegerToObject(info, "nodePort", pNode->nodePort) < 0) return -1;
if (tjsonAddStringToObject(info, "nodeFqdn", pNode->nodeFqdn) < 0) return -1;
if (tjsonAddIntegerToObject(info, "nodeId", pNode->nodeId) < 0) return -1;
if (tjsonAddIntegerToObject(info, "clusterId", pNode->clusterId) < 0) return -1;
if (tjsonAddItemToArray(nodeInfo, info) < 0) return -1;
vDebug("vgId:%d, encode config, replica:%d ep:%s:%u dnode:%d", pCfg->vgId, i, pNode->nodeFqdn, pNode->nodePort,
pNode->nodeId);
}
return 0;
@ -248,15 +253,20 @@ int vnodeDecodeConfig(const SJson *pJson, void *pObj) {
int arraySize = tjsonGetArraySize(nodeInfo);
if (arraySize != pCfg->syncCfg.replicaNum) return -1;
vDebug("vgId:%d, decode config, replicas:%d selfIndex:%d", pCfg->vgId, pCfg->syncCfg.replicaNum,
pCfg->syncCfg.myIndex);
for (int i = 0; i < arraySize; ++i) {
SJson *info = tjsonGetArrayItem(nodeInfo, i);
SJson *info = tjsonGetArrayItem(nodeInfo, i);
SNodeInfo *pNode = &pCfg->syncCfg.nodeInfo[i];
if (info == NULL) return -1;
tjsonGetNumberValue(info, "nodePort", pCfg->syncCfg.nodeInfo[i].nodePort, code);
tjsonGetNumberValue(info, "nodePort", pNode->nodePort, code);
if (code < 0) return -1;
tjsonGetStringValue(info, "nodeFqdn", pCfg->syncCfg.nodeInfo[i].nodeFqdn);
tjsonGetStringValue(info, "nodeFqdn", pNode->nodeFqdn);
if (code < 0) return -1;
tjsonGetNumberValue(info, "nodeId", pCfg->syncCfg.nodeInfo[i].nodeId, code);
tjsonGetNumberValue(info, "clusterId", pCfg->syncCfg.nodeInfo[i].clusterId, code);
tjsonGetNumberValue(info, "nodeId", pNode->nodeId, code);
tjsonGetNumberValue(info, "clusterId", pNode->clusterId, code);
vDebug("vgId:%d, decode config, replica:%d ep:%s:%u dnode:%d", pCfg->vgId, i, pNode->nodeFqdn, pNode->nodePort,
pNode->nodeId);
}
tjsonGetNumberValue(pJson, "tsdbPageSize", pCfg->tsdbPageSize, code);

6
source/dnode/vnode/src/vnd/vnodeCommit.c
View File

@ -123,8 +123,8 @@ int vnodeSaveInfo(const char *dir, const SVnodeInfo *pInfo) {
// free info binary
taosMemoryFree(data);
vInfo("vgId:%d, vnode info is saved, fname:%s replica:%d", pInfo->config.vgId, fname,
pInfo->config.syncCfg.replicaNum);
vInfo("vgId:%d, vnode info is saved, fname:%s replica:%d selfIndex:%d", pInfo->config.vgId, fname,
pInfo->config.syncCfg.replicaNum, pInfo->config.syncCfg.myIndex);
return 0;
@ -224,6 +224,8 @@ static int32_t vnodePrepareCommit(SVnode *pVnode, SCommitInfo *pInfo) {
} else {
snprintf(dir, TSDB_FILENAME_LEN, "%s", pVnode->path);
}
vDebug("vgId:%d, save config while prepare commit", TD_VID(pVnode));
if (vnodeSaveInfo(dir, &pInfo->info) < 0) {
code = terrno;
TSDB_CHECK_CODE(code, lino, _exit);

5
source/dnode/vnode/src/vnd/vnodeOpen.c
View File

@ -48,6 +48,7 @@ int32_t vnodeCreate(const char *path, SVnodeCfg *pCfg, STfs *pTfs) {
info.state.applied = -1;
info.state.commitID = 0;
vInfo("vgId:%d, save config while create", pCfg->vgId);
if (vnodeSaveInfo(dir, &info) < 0 || vnodeCommitInfo(dir, &info) < 0) {
vError("vgId:%d, failed to save vnode config since %s", pCfg ? pCfg->vgId : 0, tstrerror(terrno));
return -1;
@ -79,14 +80,14 @@ int32_t vnodeAlter(const char *path, SAlterVnodeReplicaReq *pReq, STfs *pTfs) {
pCfg->replicaNum = pReq->replica;
memset(&pCfg->nodeInfo, 0, sizeof(pCfg->nodeInfo));
vInfo("vgId:%d, save config, replicas:%d selfIndex:%d", pReq->vgId, pCfg->replicaNum, pCfg->myIndex);
vInfo("vgId:%d, save config while alter, replicas:%d selfIndex:%d", pReq->vgId, pCfg->replicaNum, pCfg->myIndex);
for (int i = 0; i < pReq->replica; ++i) {
SNodeInfo *pNode = &pCfg->nodeInfo[i];
pNode->nodeId = pReq->replicas[i].id;
pNode->nodePort = pReq->replicas[i].port;
tstrncpy(pNode->nodeFqdn, pReq->replicas[i].fqdn, sizeof(pNode->nodeFqdn));
(void)tmsgUpdateDnodeInfo(&pNode->nodeId, &pNode->clusterId, pNode->nodeFqdn, &pNode->nodePort);
vInfo("vgId:%d, save config, replica:%d ep:%s:%u", pReq->vgId, i, pNode->nodeFqdn, pNode->nodePort);
vInfo("vgId:%d, replica:%d ep:%s:%u dnode:%d", pReq->vgId, i, pNode->nodeFqdn, pNode->nodePort, pNode->nodeId);
}
info.config.syncCfg = *pCfg;

4
source/dnode/vnode/src/vnd/vnodeSnapshot.c
View File

@ -405,6 +405,10 @@ static int32_t vnodeSnapWriteInfo(SVSnapWriter *pWriter, uint8_t *pData, uint32_
} else {
snprintf(dir, TSDB_FILENAME_LEN, "%s", pWriter->pVnode->path);
}
SVnode *pVnode = pWriter->pVnode;
pWriter->info.config = pVnode->config;
vDebug("vgId:%d, save config while write snapshot", pWriter->pVnode->config.vgId);
if (vnodeSaveInfo(dir, &pWriter->info) < 0) {
code = terrno;
goto _exit;

4
source/libs/executor/inc/executil.h
View File

@ -115,6 +115,10 @@ struct SResultRowEntryInfo* getResultEntryInfo(const SResultRow* pRow, int32_t i
static FORCE_INLINE SResultRow* getResultRowByPos(SDiskbasedBuf* pBuf, SResultRowPosition* pos, bool forUpdate) {
SFilePage* bufPage = (SFilePage*)getBufPage(pBuf, pos->pageId);
if (NULL == bufPage) {
return NULL;
}
if (forUpdate) {
setBufPageDirty(bufPage, true);
}

13
source/libs/executor/src/dataInserter.c
View File

@ -41,6 +41,7 @@ typedef struct SDataInserterHandle {
SHashObj* pCols;
int32_t status;
bool queryEnd;
bool fullOrderColList;
uint64_t useconds;
uint64_t cachedSize;
TdThreadMutex mutex;
@ -125,7 +126,6 @@ int32_t dataBlockToSubmit(SDataInserterHandle* pInserter, SSubmitReq** pReq) {
int64_t uid = pInserter->pNode->tableId;
int64_t suid = pInserter->pNode->stableId;
int32_t vgId = pInserter->pNode->vgId;
bool fullCol = (pInserter->pNode->pCols->length == pTSchema->numOfCols);
SSubmitReq* ret = NULL;
int32_t sz = taosArrayGetSize(pBlocks);
@ -176,7 +176,7 @@ int32_t dataBlockToSubmit(SDataInserterHandle* pInserter, SSubmitReq** pReq) {
const STColumn* pColumn = &pTSchema->columns[k];
SColumnInfoData* pColData = NULL;
int16_t colIdx = k;
if (!fullCol) {
if (!pInserter->fullOrderColList) {
int16_t* slotId = taosHashGet(pInserter->pCols, &pColumn->colId, sizeof(pColumn->colId));
if (NULL == slotId) {
continue;
@ -212,7 +212,7 @@ int32_t dataBlockToSubmit(SDataInserterHandle* pInserter, SSubmitReq** pReq) {
tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NORM, data, true, pColumn->offset, k);
}
}
if (!fullCol) {
if (!pInserter->fullOrderColList) {
rb.hasNone = true;
}
tdSRowEnd(&rb);
@ -346,12 +346,19 @@ int32_t createDataInserter(SDataSinkManager* pManager, const SDataSinkNode* pDat
return TSDB_CODE_OUT_OF_MEMORY;
}
inserter->fullOrderColList = pInserterNode->pCols->length == inserter->pSchema->numOfCols;
inserter->pCols = taosHashInit(pInserterNode->pCols->length, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT),
false, HASH_NO_LOCK);
SNode* pNode = NULL;
int32_t i = 0;
FOREACH(pNode, pInserterNode->pCols) {
SColumnNode* pCol = (SColumnNode*)pNode;
taosHashPut(inserter->pCols, &pCol->colId, sizeof(pCol->colId), &pCol->slotId, sizeof(pCol->slotId));
if (inserter->fullOrderColList && pCol->colId != inserter->pSchema->columns[i].colId) {
inserter->fullOrderColList = false;
}
++i;
}
tsem_init(&inserter->ready, 0, 0);

6
source/libs/executor/src/executil.c
View File

@ -1726,8 +1726,10 @@ STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowI
return w;
}
w = getResultRowByPos(pBuf, &pResultRowInfo->cur, false)->win;
SResultRow* pRow = getResultRowByPos(pBuf, &pResultRowInfo->cur, false);
if (pRow) {
w = pRow->win;
}
// in case of typical time window, we can calculate time window directly.
if (w.skey > ts || w.ekey < ts) {
w = doCalculateTimeWindow(ts, pInterval);

36
source/libs/executor/src/executorimpl.c
View File

@ -150,6 +150,11 @@ SResultRow* getNewResultRow(SDiskbasedBuf* pResultBuf, int32_t* currentPageId, i
pData->num = sizeof(SFilePage);
} else {
pData = getBufPage(pResultBuf, *currentPageId);
if (pData == NULL) {
qError("failed to get buffer, code:%s", tstrerror(terrno));
return NULL;
}
pageId = *currentPageId;
if (pData->num + interBufSize > getBufPageSize(pResultBuf)) {
@ -200,6 +205,10 @@ SResultRow* doSetResultOutBufByKey(SDiskbasedBuf* pResultBuf, SResultRowInfo* pR
if (isIntervalQuery) {
if (p1 != NULL) { // the *p1 may be NULL in case of sliding+offset exists.
pResult = getResultRowByPos(pResultBuf, p1, true);
if (NULL == pResult) {
T_LONG_JMP(pTaskInfo->env, terrno);
}
ASSERT(pResult->pageId == p1->pageId && pResult->offset == p1->offset);
}
} else {
@ -208,6 +217,10 @@ SResultRow* doSetResultOutBufByKey(SDiskbasedBuf* pResultBuf, SResultRowInfo* pR
if (p1 != NULL) {
// todo
pResult = getResultRowByPos(pResultBuf, p1, true);
if (NULL == pResult) {
T_LONG_JMP(pTaskInfo->env, terrno);
}
ASSERT(pResult->pageId == p1->pageId && pResult->offset == p1->offset);
}
}
@ -216,6 +229,10 @@ SResultRow* doSetResultOutBufByKey(SDiskbasedBuf* pResultBuf, SResultRowInfo* pR
if (pResultRowInfo->cur.pageId != -1 && ((pResult == NULL) || (pResult->pageId != pResultRowInfo->cur.pageId))) {
SResultRowPosition pos = pResultRowInfo->cur;
SFilePage* pPage = getBufPage(pResultBuf, pos.pageId);
if (pPage == NULL) {
qError("failed to get buffer, code:%s, %s", tstrerror(terrno), GET_TASKID(pTaskInfo));
T_LONG_JMP(pTaskInfo->env, terrno);
}
releaseBufPage(pResultBuf, pPage);
}
@ -223,6 +240,9 @@ SResultRow* doSetResultOutBufByKey(SDiskbasedBuf* pResultBuf, SResultRowInfo* pR
if (pResult == NULL) {
ASSERT(pSup->resultRowSize > 0);
pResult = getNewResultRow(pResultBuf, &pSup->currentPageId, pSup->resultRowSize);
if (pResult == NULL) {
T_LONG_JMP(pTaskInfo->env, terrno);
}
// add a new result set for a new group
SResultRowPosition pos = {.pageId = pResult->pageId, .offset = pResult->offset};
@ -260,6 +280,11 @@ static int32_t addNewWindowResultBuf(SResultRow* pWindowRes, SDiskbasedBuf* pRes
} else {
SPageInfo* pi = getLastPageInfo(list);
pData = getBufPage(pResultBuf, getPageId(pi));
if (pData == NULL) {
qError("failed to get buffer, code:%s", tstrerror(terrno));
return terrno;
}
pageId = getPageId(pi);
if (pData->num + size > getBufPageSize(pResultBuf)) {
@ -912,7 +937,7 @@ void doSetTableGroupOutputBuf(SOperatorInfo* pOperator, int32_t numOfOutput, uin
if (pResultRow->pageId == -1) {
int32_t ret = addNewWindowResultBuf(pResultRow, pAggInfo->aggSup.pResultBuf, pAggInfo->binfo.pRes->info.rowSize);
if (ret != TSDB_CODE_SUCCESS) {
return;
T_LONG_JMP(pTaskInfo->env, terrno);
}
}
@ -993,6 +1018,11 @@ static void doCopyResultToDataBlock(SExprInfo* pExprInfo, int32_t numOfExprs, SR
int32_t finalizeResultRows(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPosition, SExprSupp* pSup,
SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo) {
SFilePage* page = getBufPage(pBuf, resultRowPosition->pageId);
if (page == NULL) {
qError("failed to get buffer, code:%s, %s", tstrerror(terrno), GET_TASKID(pTaskInfo));
T_LONG_JMP(pTaskInfo->env, terrno);
}
SResultRow* pRow = (SResultRow*)((char*)page + resultRowPosition->offset);
SqlFunctionCtx* pCtx = pSup->pCtx;
@ -1036,6 +1066,10 @@ int32_t doCopyToSDataBlock(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock, SExprS
for (int32_t i = pGroupResInfo->index; i < numOfRows; i += 1) {
SResKeyPos* pPos = taosArrayGetP(pGroupResInfo->pRows, i);
SFilePage* page = getBufPage(pBuf, pPos->pos.pageId);
if (page == NULL) {
qError("failed to get buffer, code:%s, %s", tstrerror(terrno), GET_TASKID(pTaskInfo));
T_LONG_JMP(pTaskInfo->env, terrno);
}
SResultRow* pRow = (SResultRow*)((char*)page + pPos->pos.offset);

19
source/libs/executor/src/groupoperator.c
View File

@ -492,13 +492,17 @@ _error:
static void doHashPartition(SOperatorInfo* pOperator, SSDataBlock* pBlock) {
SPartitionOperatorInfo* pInfo = pOperator->info;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
for (int32_t j = 0; j < pBlock->info.rows; ++j) {
recordNewGroupKeys(pInfo->pGroupCols, pInfo->pGroupColVals, pBlock, j);
int32_t len = buildGroupKeys(pInfo->keyBuf, pInfo->pGroupColVals);
SDataGroupInfo* pGroupInfo = NULL;
void* pPage = getCurrentDataGroupInfo(pInfo, &pGroupInfo, len);
if (pPage == NULL) {
T_LONG_JMP(pTaskInfo->env, terrno);
}
pGroupInfo->numOfRows += 1;
@ -595,6 +599,10 @@ void* getCurrentDataGroupInfo(const SPartitionOperatorInfo* pInfo, SDataGroupInf
} else {
int32_t* curId = taosArrayGetLast(p->pPageList);
pPage = getBufPage(pInfo->pBuf, *curId);
if (pPage == NULL) {
qError("failed to get buffer, code:%s", tstrerror(terrno));
return pPage;
}
int32_t* rows = (int32_t*)pPage;
if (*rows >= pInfo->rowCapacity) {
@ -674,7 +682,8 @@ static int compareDataGroupInfo(const void* group1, const void* group2) {
static SSDataBlock* buildPartitionResult(SOperatorInfo* pOperator) {
SPartitionOperatorInfo* pInfo = pOperator->info;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SDataGroupInfo* pGroupInfo =
(pInfo->groupIndex != -1) ? taosArrayGet(pInfo->sortedGroupArray, pInfo->groupIndex) : NULL;
if (pInfo->groupIndex == -1 || pInfo->pageIndex >= taosArrayGetSize(pGroupInfo->pPageList)) {
@ -692,7 +701,11 @@ static SSDataBlock* buildPartitionResult(SOperatorInfo* pOperator) {
int32_t* pageId = taosArrayGet(pGroupInfo->pPageList, pInfo->pageIndex);
void* page = getBufPage(pInfo->pBuf, *pageId);
if (page == NULL) {
qError("failed to get buffer, code:%s, %s", tstrerror(terrno), GET_TASKID(pTaskInfo));
T_LONG_JMP(pTaskInfo->env, terrno);
}
blockDataEnsureCapacity(pInfo->binfo.pRes, pInfo->rowCapacity);
blockDataFromBuf1(pInfo->binfo.pRes, page, pInfo->rowCapacity);

4
source/libs/executor/src/scanoperator.c
View File

@ -170,6 +170,10 @@ static SResultRow* getTableGroupOutputBuf(SOperatorInfo* pOperator, uint64_t gro
}
*pPage = getBufPage(pTableScanInfo->base.pdInfo.pAggSup->pResultBuf, p1->pageId);
if (NULL == *pPage) {
return NULL;
}
return (SResultRow*)((char*)(*pPage) + p1->offset);
}

14
source/libs/executor/src/timewindowoperator.c
View File

@ -636,6 +636,10 @@ static void doInterpUnclosedTimeWindow(SOperatorInfo* pOperatorInfo, int32_t num
}
SResultRow* pr = getResultRowByPos(pInfo->aggSup.pResultBuf, p1, false);
if (NULL == pr) {
T_LONG_JMP(pTaskInfo->env, terrno);
}
ASSERT(pr->offset == p1->offset && pr->pageId == p1->pageId);
if (pr->closed) {
@ -1315,6 +1319,10 @@ static void setInverFunction(SqlFunctionCtx* pCtx, int32_t num, EStreamType type
static void doClearWindowImpl(SResultRowPosition* p1, SDiskbasedBuf* pResultBuf, SExprSupp* pSup, int32_t numOfOutput) {
SResultRow* pResult = getResultRowByPos(pResultBuf, p1, false);
if (NULL == pResult) {
return;
}
SqlFunctionCtx* pCtx = pSup->pCtx;
for (int32_t i = 0; i < numOfOutput; ++i) {
pCtx[i].resultInfo = getResultEntryInfo(pResult, i, pSup->rowEntryInfoOffset);
@ -1328,6 +1336,9 @@ static void doClearWindowImpl(SResultRowPosition* p1, SDiskbasedBuf* pResultBuf,
}
}
SFilePage* bufPage = getBufPage(pResultBuf, p1->pageId);
if (NULL == bufPage) {
return;
}
setBufPageDirty(bufPage, true);
releaseBufPage(pResultBuf, bufPage);
}
@ -4114,6 +4125,9 @@ void destroyMAIOperatorInfo(void* param) {
static SResultRow* doSetSingleOutputTupleBuf(SResultRowInfo* pResultRowInfo, SAggSupporter* pSup) {
SResultRow* pResult = getNewResultRow(pSup->pResultBuf, &pSup->currentPageId, pSup->resultRowSize);
if (NULL == pResult) {
return pResult;
}
pResultRowInfo->cur = (SResultRowPosition){.pageId = pResult->pageId, .offset = pResult->offset};
return pResult;
}

9
source/libs/executor/src/tsort.c
View File

@ -270,6 +270,10 @@ static int32_t sortComparInit(SMsortComparParam* pParam, SArray* pSources, int32
int32_t* pPgId = taosArrayGet(pSource->pageIdList, pSource->pageIndex);
void* pPage = getBufPage(pHandle->pBuf, *pPgId);
if (NULL == pPage) {
return terrno;
}
code = blockDataFromBuf(pSource->src.pBlock, pPage);
if (code != TSDB_CODE_SUCCESS) {
return code;
@ -337,6 +341,11 @@ static int32_t adjustMergeTreeForNextTuple(SSortSource* pSource, SMultiwayMergeT
int32_t* pPgId = taosArrayGet(pSource->pageIdList, pSource->pageIndex);
void* pPage = getBufPage(pHandle->pBuf, *pPgId);
if (pPage == NULL) {
qError("failed to get buffer, code:%s", tstrerror(terrno));
return terrno;
}
int32_t code = blockDataFromBuf(pSource->src.pBlock, pPage);
if (code != TSDB_CODE_SUCCESS) {
return code;

2
source/libs/function/src/builtinsimpl.c
View File

@ -784,7 +784,7 @@ int32_t minmaxFunctionFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
pEntryInfo->isNullRes = (pEntryInfo->numOfRes == 0) ? 1 : 0;
if (pCol->info.type == TSDB_DATA_TYPE_FLOAT) {
float v = *(float*)&pRes->v;
float v = GET_FLOAT_VAL(&pRes->v);
colDataAppend(pCol, currentRow, (const char*)&v, pEntryInfo->isNullRes);
} else {
colDataAppend(pCol, currentRow, (const char*)&pRes->v, pEntryInfo->isNullRes);

15
source/libs/function/src/detail/tminmax.c
View File

@ -737,7 +737,12 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
}
if (!pBuf->assign) {
pBuf->v = *(int64_t*)tval;
if (type == TSDB_DATA_TYPE_FLOAT) {
GET_FLOAT_VAL(&pBuf->v) = GET_DOUBLE_VAL(tval);
} else {
pBuf->v = GET_INT64_VAL(tval);
}
if (pCtx->subsidiaries.num > 0) {
index = findRowIndex(pInput->startRowIndex, pInput->numOfRows, pCol, tval);
if (index >= 0) {
@ -751,7 +756,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
int64_t val = GET_INT64_VAL(tval);
if ((prev < val) ^ isMinFunc) {
*(int64_t*)&pBuf->v = val;
GET_INT64_VAL(&pBuf->v) = val;
if (pCtx->subsidiaries.num > 0) {
index = findRowIndex(pInput->startRowIndex, pInput->numOfRows, pCol, tval);
if (index >= 0) {
@ -765,7 +770,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
uint64_t val = GET_UINT64_VAL(tval);
if ((prev < val) ^ isMinFunc) {
*(uint64_t*)&pBuf->v = val;
GET_UINT64_VAL(&pBuf->v) = val;
if (pCtx->subsidiaries.num > 0) {
index = findRowIndex(pInput->startRowIndex, pInput->numOfRows, pCol, tval);
if (index >= 0) {
@ -779,7 +784,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
double val = GET_DOUBLE_VAL(tval);
if ((prev < val) ^ isMinFunc) {
*(double*)&pBuf->v = val;
GET_DOUBLE_VAL(&pBuf->v) = val;
if (pCtx->subsidiaries.num > 0) {
index = findRowIndex(pInput->startRowIndex, pInput->numOfRows, pCol, tval);
if (index >= 0) {
@ -793,7 +798,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
float val = GET_DOUBLE_VAL(tval);
if ((prev < val) ^ isMinFunc) {
*(float*)&pBuf->v = val;
GET_FLOAT_VAL(&pBuf->v) = val;
}
if (pCtx->subsidiaries.num > 0) {

5
source/libs/function/src/udfd.c
View File

@ -460,13 +460,14 @@ void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
#else
snprintf(path, sizeof(path), "%s/lib%s.so", tsTempDir, pFuncInfo->name);
#endif
TdFilePtr file =
taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC | TD_FILE_AUTO_DEL);
TdFilePtr file = taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC);
if (file == NULL) {
fnError("udfd write udf shared library: %s failed, error: %d %s", path, errno, strerror(errno));
msgInfo->code = TSDB_CODE_FILE_CORRUPTED;
goto _return;
}
int64_t count = taosWriteFile(file, pFuncInfo->pCode, pFuncInfo->codeSize);
if (count != pFuncInfo->codeSize) {
fnError("udfd write udf shared library failed");

3
source/libs/sync/src/syncMain.c
View File

@ -851,6 +851,7 @@ SSyncNode* syncNodeOpen(SSyncInfo* pSyncInfo) {
if (!taosCheckExistFile(pSyncNode->configPath)) {
// create a new raft config file
sInfo("vgId:%d, create a new raft config file", pSyncNode->vgId);
pSyncNode->raftCfg.isStandBy = pSyncInfo->isStandBy;
pSyncNode->raftCfg.snapshotStrategy = pSyncInfo->snapshotStrategy;
pSyncNode->raftCfg.lastConfigIndex = SYNC_INDEX_INVALID;
@ -894,7 +895,6 @@ SSyncNode* syncNodeOpen(SSyncInfo* pSyncInfo) {
pNode->nodeId, pNode->clusterId);
}
pSyncNode->pWal = pSyncInfo->pWal;
pSyncNode->msgcb = pSyncInfo->msgcb;
pSyncNode->syncSendMSg = pSyncInfo->syncSendMSg;
@ -1658,7 +1658,6 @@ void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncInde
// persist cfg
syncWriteCfgFile(pSyncNode);
// change isStandBy to normal (election timeout)
if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
syncNodeBecomeLeader(pSyncNode, "");

6
source/os/CMakeLists.txt
View File

@ -59,4 +59,8 @@ endif()
IF (JEMALLOC_ENABLED)
target_link_libraries(os PUBLIC -ljemalloc)
ENDIF ()
ENDIF ()
if(${BUILD_TEST})
add_subdirectory(test)
endif(${BUILD_TEST})

91
source/os/src/osFile.c
View File

@ -39,14 +39,6 @@
#define _SEND_FILE_STEP_ 1000
#endif
#if defined(WINDOWS)
typedef int32_t FileFd;
typedef int32_t SocketFd;
#else
typedef int32_t FileFd;
typedef int32_t SocketFd;
#endif
typedef int32_t FileFd;
typedef struct TdFile {
@ -54,19 +46,10 @@ typedef struct TdFile {
int refId;
FileFd fd;
FILE *fp;
} * TdFilePtr, TdFile;
} TdFile;
#define FILE_WITH_LOCK 1
typedef struct AutoDelFile *AutoDelFilePtr;
typedef struct AutoDelFile {
char *name;
AutoDelFilePtr lastAutoDelFilePtr;
} AutoDelFile;
static TdThreadMutex autoDelFileLock;
static AutoDelFilePtr nowAutoDelFilePtr = NULL;
static TdThreadOnce autoDelFileInit = PTHREAD_ONCE_INIT;
void taosGetTmpfilePath(const char *inputTmpDir, const char *fileNamePrefix, char *dstPath) {
#ifdef WINDOWS
const char *tdengineTmpFileNamePrefix = "tdengine-";
@ -268,34 +251,6 @@ int32_t taosDevInoFile(TdFilePtr pFile, int64_t *stDev, int64_t *stIno) {
return 0;
}
void autoDelFileList() {
taosThreadMutexLock(&autoDelFileLock);
while (nowAutoDelFilePtr != NULL) {
taosRemoveFile(nowAutoDelFilePtr->name);
AutoDelFilePtr tmp = nowAutoDelFilePtr->lastAutoDelFilePtr;
taosMemoryFree(nowAutoDelFilePtr->name);
taosMemoryFree(nowAutoDelFilePtr);
nowAutoDelFilePtr = tmp;
}
taosThreadMutexUnlock(&autoDelFileLock);
taosThreadMutexDestroy(&autoDelFileLock);
}
void autoDelFileListInit() {
taosThreadMutexInit(&autoDelFileLock, NULL);
atexit(autoDelFileList);
}
void autoDelFileListAdd(const char *path) {
taosThreadOnce(&autoDelFileInit, autoDelFileListInit);
taosThreadMutexLock(&autoDelFileLock);
AutoDelFilePtr tmp = taosMemoryMalloc(sizeof(AutoDelFile));
tmp->lastAutoDelFilePtr = nowAutoDelFilePtr;
tmp->name = taosMemoryStrDup(path);
nowAutoDelFilePtr = tmp;
taosThreadMutexUnlock(&autoDelFileLock);
}
TdFilePtr taosOpenFile(const char *path, int32_t tdFileOptions) {
int fd = -1;
FILE *fp = NULL;
@ -313,7 +268,6 @@ TdFilePtr taosOpenFile(const char *path, int32_t tdFileOptions) {
assert(!(tdFileOptions & TD_FILE_EXCL));
fp = fopen(path, mode);
if (fp == NULL) {
// terrno = TAOS_SYSTEM_ERROR(errno);
return NULL;
}
} else {
@ -331,32 +285,44 @@ TdFilePtr taosOpenFile(const char *path, int32_t tdFileOptions) {
access |= (tdFileOptions & TD_FILE_TEXT) ? O_TEXT : 0;
access |= (tdFileOptions & TD_FILE_EXCL) ? O_EXCL : 0;
#ifdef WINDOWS
fd = _open(path, access, _S_IREAD | _S_IWRITE);
int32_t pmode = _S_IREAD | _S_IWRITE;
if (tdFileOptions & TD_FILE_AUTO_DEL) {
pmode |= _O_TEMPORARY;
}
fd = _open(path, access, pmode);
#else
fd = open(path, access, S_IRWXU | S_IRWXG | S_IRWXO);
#endif
if (fd == -1) {
// terrno = TAOS_SYSTEM_ERROR(errno);
return NULL;
}
}
TdFilePtr pFile = (TdFilePtr)taosMemoryMalloc(sizeof(TdFile));
if (pFile == NULL) {
// terrno = TSDB_CODE_OUT_OF_MEMORY;
if (fd >= 0) close(fd);
if (fp != NULL) fclose(fp);
return NULL;
}
#if FILE_WITH_LOCK
taosThreadRwlockInit(&(pFile->rwlock), NULL);
#endif
pFile->fd = fd;
pFile->fp = fp;
pFile->refId = 0;
if (tdFileOptions & TD_FILE_AUTO_DEL) {
autoDelFileListAdd(path);
#ifdef WINDOWS
// do nothing, since the property of pmode is set with _O_TEMPORARY; the OS will recycle
// the file handle, as well as the space on disk.
#else
// Remove it instantly, so when the program exits normally/abnormally, the file
// will be automatically remove by OS.
unlink(path);
#endif
}
return pFile;
}
@ -560,6 +526,21 @@ int32_t taosFStatFile(TdFilePtr pFile, int64_t *size, int32_t *mtime) {
int32_t taosLockFile(TdFilePtr pFile) {
#ifdef WINDOWS
BOOL fSuccess = FALSE;
LARGE_INTEGER fileSize;
OVERLAPPED overlapped = {0};
HANDLE hFile = (HANDLE)_get_osfhandle(pFile->fd);
fSuccess = LockFileEx(hFile, LOCKFILE_EXCLUSIVE_LOCK | LOCKFILE_FAIL_IMMEDIATELY,
0, // reserved
~0, // number of bytes to lock low
~0, // number of bytes to lock high
&overlapped // overlapped structure
);
if (!fSuccess) {
return GetLastError();
}
return 0;
#else
assert(pFile->fd >= 0); // Please check if you have closed the file.
@ -570,6 +551,14 @@ int32_t taosLockFile(TdFilePtr pFile) {
int32_t taosUnLockFile(TdFilePtr pFile) {
#ifdef WINDOWS
BOOL fSuccess = FALSE;
OVERLAPPED overlapped = {0};
HANDLE hFile = (HANDLE)_get_osfhandle(pFile->fd);
fSuccess = UnlockFileEx(hFile, 0, ~0, ~0, &overlapped);
if (!fSuccess) {
return GetLastError();
}
return 0;
#else
assert(pFile->fd >= 0); // Please check if you have closed the file.

94
source/os/test/osTests.cpp
View File

@ -27,6 +27,7 @@
#pragma GCC diagnostic ignored "-Wpointer-arith"
#include "os.h"
#include "tlog.h"
TEST(osTest, osSystem) {
const char *flags = "UTL FATAL ";
@ -35,4 +36,97 @@ TEST(osTest, osSystem) {
taosPrintTrace(flags, level, dflag);
}
void fileOperateOnFree(void *param) {
char * fname = (char *)param;
TdFilePtr pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE);
printf("On free thread open file\n");
ASSERT_NE(pFile, nullptr);
int ret = taosLockFile(pFile);
printf("On free thread lock file ret:%d\n", ret);
ASSERT_EQ(ret, 0);
ret = taosUnLockFile(pFile);
printf("On free thread unlock file ret:%d\n", ret);
ASSERT_EQ(ret, 0);
ret = taosCloseFile(&pFile);
ASSERT_EQ(ret, 0);
printf("On free thread close file ret:%d\n", ret);
}
void *fileOperateOnFreeThread(void *param) {
fileOperateOnFree(param);
return NULL;
}
void fileOperateOnBusy(void *param) {
char * fname = (char *)param;
TdFilePtr pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE);
printf("On busy thread open file\n");
ASSERT_NE(pFile, nullptr);
int ret = taosLockFile(pFile);
printf("On busy thread lock file ret:%d\n", ret);
ASSERT_NE(ret, 0);
ret = taosUnLockFile(pFile);
printf("On busy thread unlock file ret:%d\n", ret);
#ifdef _TD_DARWIN_64
ASSERT_EQ(ret, 0);
#else
ASSERT_NE(ret, 0);
#endif
ret = taosCloseFile(&pFile);
printf("On busy thread close file ret:%d\n", ret);
ASSERT_EQ(ret, 0);
}
void *fileOperateOnBusyThread(void *param) {
fileOperateOnBusy(param);
return NULL;
}
TEST(osTest, osFile) {
char *fname = "./osfiletest1.txt";
TdFilePtr pOutFD = taosCreateFile(fname, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
ASSERT_NE(pOutFD, nullptr);
printf("create file success\n");
TdFilePtr pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE);
printf("open file\n");
ASSERT_NE(pFile, nullptr);
int ret = taosLockFile(pFile);
printf("lock file ret:%d\n", ret);
ASSERT_EQ(ret, 0);
TdThreadAttr thattr;
taosThreadAttrInit(&thattr);
TdThread thread1, thread2;
taosThreadCreate(&(thread1), &thattr, fileOperateOnBusyThread, (void *)fname);
taosThreadAttrDestroy(&thattr);
taosThreadJoin(thread1, NULL);
taosThreadClear(&thread1);
ret = taosUnLockFile(pFile);
printf("unlock file ret:%d\n", ret);
ASSERT_EQ(ret, 0);
ret = taosCloseFile(&pFile);
printf("close file ret:%d\n", ret);
ASSERT_EQ(ret, 0);
taosThreadCreate(&(thread2), &thattr, fileOperateOnFreeThread, (void *)fname);
taosThreadAttrDestroy(&thattr);
taosThreadJoin(thread2, NULL);
taosThreadClear(&thread2);
//int ret = taosRemoveFile(fname);
//ASSERT_EQ(ret, 0);
//printf("remove file success");
}
#pragma GCC diagnostic pop

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

@ -273,42 +273,90 @@ int32_t tsDecompressINTImp(const char *const input, const int32_t nelements, cha
char bit = bit_per_integer[(int32_t)selector]; // bit = 3
int32_t elems = selector_to_elems[(int32_t)selector];
for (int32_t i = 0; i < elems; i++) {
uint64_t zigzag_value;
// Optimize the performance, by remove the constantly switch operation.
int32_t v = 0;
uint64_t zigzag_value;
if (selector == 0 || selector == 1) {
zigzag_value = 0;
} else {
zigzag_value = ((w >> (4 + bit * i)) & INT64MASK(bit));
}
int64_t diff = ZIGZAG_DECODE(int64_t, zigzag_value);
int64_t curr_value = diff + prev_value;
prev_value = curr_value;
switch (type) {
case TSDB_DATA_TYPE_BIGINT: {
for (int32_t i = 0; i < elems; i++) {
if (selector == 0 || selector == 1) {
zigzag_value = 0;
} else {
zigzag_value = ((w >> (4 + v)) & INT64MASK(bit));
}
int64_t diff = ZIGZAG_DECODE(int64_t, zigzag_value);
int64_t curr_value = diff + prev_value;
prev_value = curr_value;
switch (type) {
case TSDB_DATA_TYPE_BIGINT:
*((int64_t *)output + _pos) = (int64_t)curr_value;
_pos++;
break;
case TSDB_DATA_TYPE_INT:
v += bit;
if ((++count) == nelements) break;
}
} break;
case TSDB_DATA_TYPE_INT: {
for (int32_t i = 0; i < elems; i++) {
if (selector == 0 || selector == 1) {
zigzag_value = 0;
} else {
zigzag_value = ((w >> (4 + v)) & INT64MASK(bit));
}
int64_t diff = ZIGZAG_DECODE(int64_t, zigzag_value);
int64_t curr_value = diff + prev_value;
prev_value = curr_value;
*((int32_t *)output + _pos) = (int32_t)curr_value;
_pos++;
break;
case TSDB_DATA_TYPE_SMALLINT:
v += bit;
if ((++count) == nelements) break;
}
} break;
case TSDB_DATA_TYPE_SMALLINT: {
for (int32_t i = 0; i < elems; i++) {
if (selector == 0 || selector == 1) {
zigzag_value = 0;
} else {
zigzag_value = ((w >> (4 + v)) & INT64MASK(bit));
}
int64_t diff = ZIGZAG_DECODE(int64_t, zigzag_value);
int64_t curr_value = diff + prev_value;
prev_value = curr_value;
*((int16_t *)output + _pos) = (int16_t)curr_value;
_pos++;
break;
case TSDB_DATA_TYPE_TINYINT:
v += bit;
if ((++count) == nelements) break;
}
} break;
case TSDB_DATA_TYPE_TINYINT: {
for (int32_t i = 0; i < elems; i++) {
if (selector == 0 || selector == 1) {
zigzag_value = 0;
} else {
zigzag_value = ((w >> (4 + v)) & INT64MASK(bit));
}
int64_t diff = ZIGZAG_DECODE(int64_t, zigzag_value);
int64_t curr_value = diff + prev_value;
prev_value = curr_value;
*((int8_t *)output + _pos) = (int8_t)curr_value;
_pos++;
break;
default:
perror("Wrong integer types.\n");
return -1;
}
count++;
if (count == nelements) break;
v += bit;
if ((++count) == nelements) break;
}
} break;
}
ip += LONG_BYTES;
}
@ -758,7 +806,7 @@ int32_t tsDecompressDoubleImp(const char *const input, const int32_t nelements,
uint64_t prev_value = 0;
for (int32_t i = 0; i < nelements; i++) {
if (i % 2 == 0) {
if ((i & 0x01) == 0) {
flags = input[ipos++];
}

3
source/util/src/terror.c
View File

@ -276,8 +276,9 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_SUBSCRIBE_NOT_EXIST, "Subcribe not exist")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_OFFSET_NOT_EXIST, "Offset not exist")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_CONSUMER_NOT_READY, "Consumer not ready")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOPIC_SUBSCRIBED, "Topic subscribed cannot be dropped")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOPIC_MUST_BE_DELETED, "Topic must be dropped first")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_CGROUP_USED, "Consumer group being used by some consumer")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOPIC_MUST_BE_DELETED, "Topic must be dropped first")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_SUB_OPTION, "Invalid subscribe option")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_IN_REBALANCE, "Topic being rebalanced")
// mnode-stream

14
tests/parallel_test/cases.task
View File

@ -175,6 +175,7 @@
,,y,script,./test.sh -f tsim/query/session.sim
,,y,script,./test.sh -f tsim/query/udf.sim
,,y,script,./test.sh -f tsim/query/udf_with_const.sim
,,y,script,./test.sh -f tsim/query/sys_tbname.sim
,,y,script,./test.sh -f tsim/query/groupby.sim
,,y,script,./test.sh -f tsim/qnode/basic1.sim
,,y,script,./test.sh -f tsim/snode/basic1.sim
@ -617,6 +618,8 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/varchar.py -R
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/case_when.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/case_when.py -R
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/blockSMA.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/blockSMA.py -R
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/update_data.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/tb_100w_data_order.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/delete_stable.py
@ -699,8 +702,8 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqConsumerGroup.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqShow.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqAlterSchema.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqConsFromTsdb.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqConsFromTsdb1.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqConsFromTsdb.py -N 3 -n 3
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqConsFromTsdb1.py -N 3 -n 3
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqConsFromTsdb-mutilVg.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqConsFromTsdb1-mutilVg.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqConsFromTsdb-1ctb.py
@ -715,11 +718,11 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDnodeRestart.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDnodeRestart1.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUpdate-1ctb.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUpdateWithConsume.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUpdateWithConsume.py -N 3 -n 3
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUpdate-multiCtb-snapshot0.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUpdate-multiCtb-snapshot1.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDelete-1ctb.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDelete-multiCtb.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDelete-multiCtb.py -N 3 -n 3
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDropStb.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDropStbCtb.py
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDropNtb-snapshot0.py
@ -832,6 +835,7 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/tsbsQuery.py -Q 2
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/sml.py -Q 2
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/case_when.py -Q 2
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/blockSMA.py -Q 2
,,y,system-test,./pytest.sh python3 ./test.py -f 99-TDcase/TD-21561.py -Q 2
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/between.py -Q 3
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/distinct.py -Q 3
@ -928,6 +932,7 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/sml.py -Q 3
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/interp.py -Q 3
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/case_when.py -Q 3
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/blockSMA.py -Q 3
,,y,system-test,./pytest.sh python3 ./test.py -f 99-TDcase/TD-21561.py -Q 3
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/between.py -Q 4
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/distinct.py -Q 4
@ -1034,6 +1039,7 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/insert_null_none.py -Q 2
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/insert_null_none.py -Q 3
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/insert_null_none.py -Q 4
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/blockSMA.py -Q 4
,,y,system-test,./pytest.sh python3 ./test.py -f 99-TDcase/TD-21561.py -Q 4
,,y,system-test,./pytest.sh python3 ./test.py -f 99-TDcase/TD-20582.py

20
tests/script/tsim/insert/insert_select.sim
View File

@ -42,4 +42,24 @@ if $rows != 2 then
return -1
endi
print ======== step2
sql drop database if exists db1;
sql create database db1 vgroups 1;
sql use db1;
sql create table t1(ts timestamp, a int, b int );
sql create table t2(ts timestamp, a int, b int );
sql insert into t1 values(1648791211000,1,2);
sql insert into t2 (ts, b, a) select ts, a, b from t1;
sql select * from t2;
if $rows != 1 then
return -1
endi
if $data01 != 2 then
return -1
endi
if $data02 != 1 then
return -1
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT

89
tests/script/tsim/query/sys_tbname.sim Normal file
View File

@ -0,0 +1,89 @@
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/exec.sh -n dnode1 -s start
sql connect
sql create database sys_tbname;
sql use sys_tbname;
sql create stable st(ts timestamp, f int) tags(t int);
sql create table ct1 using st tags(1);
sql create table ct2 using st tags(2);
sql create table t (ts timestamp, f int);
sql insert into t values(now, 1)(now+1s, 2);
sql create table t2 (ts timestamp, f1 int, f2 int);
sql insert into t2 values(now, 0, 0)(now+1s, 1, 1);
sql select tbname from information_schema.ins_databases;
print $rows $data00
if $rows != 3 then
return -1
endi
if $data00 != @ins_databases@ then
return -1
endi
sql select distinct tbname from information_schema.ins_databases;
print $rows $data00
if $rows != 1 then
return -1
endi
if $data00 != @ins_databases@ then
return -1
endi
sql select tbname from information_schema.ins_stables;
print $rows $data00
if $rows != 1 then
return -1
endi
if $data00 != @ins_stables@ then
return -1
endi
sql select distinct tbname from information_schema.ins_stables;
print $rows $data00
if $rows != 1 then
return -1
endi
if $data00 != @ins_stables@ then
return -1
endi
sql select tbname from information_schema.ins_tables;
print $rows $data00
if $rows != 32 then
return -1
endi
if $data00 != @ins_tables@ then
return -1
endi
sql select distinct tbname from information_schema.ins_tables;
print $rows $data00
if $rows != 1 then
return -1
endi
if $data00 != @ins_tables@ then
return -1
endi
sql select tbname from information_schema.ins_tags;
print $rows $data00
if $rows != 2 then
return -1
endi
if $data00 != @ins_tags@ then
return -1
endi
sql select distinct tbname from information_schema.ins_tags;
print $rows $data00
if $rows != 1 then
return -1
endi
if $data00 != @ins_tags@ then
return -1
endi
#system sh/exec.sh -n dnode1 -s stop -x SIGINT

146
tests/system-test/2-query/blockSMA.py Normal file
View File

@ -0,0 +1,146 @@
from wsgiref.headers import tspecials
from util.log import *
from util.cases import *
from util.sql import *
import numpy as np
class TDTestCase:
def init(self, conn, logSql, replicaVar=1):
self.replicaVar = int(replicaVar)
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor())
self.rowNum = 10000
self.ts = 1537146000000
def run(self):
dbname = "db"
tdSql.prepare()
tdSql.execute(f'''create table {dbname}.ntb(ts timestamp, col1 tinyint, col2 smallint, col3 int, col4 bigint, col5 float, col6 double,
col7 bool, col8 binary(20), col9 nchar(20), col11 tinyint unsigned, col12 smallint unsigned, col13 int unsigned, col14 bigint unsigned)''')
for i in range(self.rowNum):
tdSql.execute(f"insert into {dbname}.ntb values(%d, %d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d', %d, %d, %d, %d)"
% (self.ts + i, i % 127 + 1, i + 1, i + 1, i + 1, i + 0.1, i + 0.1, i % 2, i + 1, i + 1, i % 255 + 1, i + 1, i + 1, i + 1))
tdSql.execute('flush database db')
# test functions using sma result
tdSql.query(f"select count(col1),min(col1),max(col1),avg(col1),sum(col1),spread(col1),percentile(col1, 0),first(col1),last(col1) from {dbname}.ntb")
tdSql.checkData(0, 0, 10000)
tdSql.checkData(0, 1, 1)
tdSql.checkData(0, 2, 127)
tdSql.checkData(0, 3, 63.8449)
tdSql.checkData(0, 4, 638449)
tdSql.checkData(0, 5, 126.0)
tdSql.checkData(0, 6, 1.0)
tdSql.checkData(0, 7, 1)
tdSql.checkData(0, 8, 94)
tdSql.query(f"select count(col2),min(col2),max(col2),avg(col2),sum(col2),spread(col2),percentile(col2, 0),first(col2),last(col2) from {dbname}.ntb")
tdSql.checkData(0, 0, 10000)
tdSql.checkData(0, 1, 1)
tdSql.checkData(0, 2, 10000)
tdSql.checkData(0, 3, 5000.5)
tdSql.checkData(0, 4, 50005000)
tdSql.checkData(0, 5, 9999.0)
tdSql.checkData(0, 6, 1.0)
tdSql.checkData(0, 7, 1)
tdSql.checkData(0, 8, 10000)
tdSql.query(f"select count(col3),min(col3),max(col3),avg(col3),sum(col3),spread(col3),percentile(col3, 0),first(col3),last(col3) from {dbname}.ntb")
tdSql.checkData(0, 0, 10000)
tdSql.checkData(0, 1, 1)
tdSql.checkData(0, 2, 10000)
tdSql.checkData(0, 3, 5000.5)
tdSql.checkData(0, 4, 50005000)
tdSql.checkData(0, 5, 9999.0)
tdSql.checkData(0, 6, 1.0)
tdSql.checkData(0, 7, 1)
tdSql.checkData(0, 8, 10000)
tdSql.query(f"select count(col4),min(col4),max(col4),avg(col4),sum(col4),spread(col4),percentile(col4, 0),first(col4),last(col4) from {dbname}.ntb")
tdSql.checkData(0, 0, 10000)
tdSql.checkData(0, 1, 1)
tdSql.checkData(0, 2, 10000)
tdSql.checkData(0, 3, 5000.5)
tdSql.checkData(0, 4, 50005000)
tdSql.checkData(0, 5, 9999.0)
tdSql.checkData(0, 6, 1.0)
tdSql.checkData(0, 7, 1)
tdSql.checkData(0, 8, 10000)
tdSql.query(f"select count(col5),min(col5),max(col5),avg(col5),sum(col5),spread(col5),percentile(col5, 0),first(col5),last(col5) from {dbname}.ntb")
tdSql.checkData(0, 0, 10000)
tdSql.checkData(0, 1, 0.1)
tdSql.checkData(0, 2, 9999.09961)
tdSql.checkData(0, 3, 4999.599985846)
tdSql.checkData(0, 4, 49995999.858455874)
tdSql.checkData(0, 5, 9998.999609374)
tdSql.checkData(0, 6, 0.100000001)
tdSql.checkData(0, 7, 0.1)
tdSql.checkData(0, 8, 9999.09961)
tdSql.query(f"select count(col6),min(col6),max(col6),avg(col6),sum(col6),spread(col6),percentile(col6, 0),first(col6),last(col6) from {dbname}.ntb")
tdSql.checkData(0, 0, 10000)
tdSql.checkData(0, 1, 0.1)
tdSql.checkData(0, 2, 9999.100000000)
tdSql.checkData(0, 3, 4999.600000001)
tdSql.checkData(0, 4, 49996000.000005305)
tdSql.checkData(0, 5, 9999.000000000)
tdSql.checkData(0, 6, 0.1)
tdSql.checkData(0, 7, 0.1)
tdSql.checkData(0, 8, 9999.1)
tdSql.query(f"select count(col11),min(col11),max(col11),avg(col11),sum(col11),spread(col11),percentile(col11, 0),first(col11),last(col11) from {dbname}.ntb")
tdSql.checkData(0, 0, 10000)
tdSql.checkData(0, 1, 1)
tdSql.checkData(0, 2, 255)
tdSql.checkData(0, 3, 127.45)
tdSql.checkData(0, 4, 1274500)
tdSql.checkData(0, 5, 254.000000000)
tdSql.checkData(0, 6, 1.0)
tdSql.checkData(0, 7, 1)
tdSql.checkData(0, 8, 55)
tdSql.query(f"select count(col12),min(col12),max(col12),avg(col12),sum(col12),spread(col12),percentile(col12, 0),first(col12),last(col12) from {dbname}.ntb")
tdSql.checkData(0, 0, 10000)
tdSql.checkData(0, 1, 1)
tdSql.checkData(0, 2, 10000)
tdSql.checkData(0, 3, 5000.5)
tdSql.checkData(0, 4, 50005000)
tdSql.checkData(0, 5, 9999.0)
tdSql.checkData(0, 6, 1.0)
tdSql.checkData(0, 7, 1)
tdSql.checkData(0, 8, 10000)
tdSql.query(f"select count(col13),min(col13),max(col13),avg(col13),sum(col13),spread(col13),percentile(col13, 0),first(col13),last(col13) from {dbname}.ntb")
tdSql.checkData(0, 0, 10000)
tdSql.checkData(0, 1, 1)
tdSql.checkData(0, 2, 10000)
tdSql.checkData(0, 3, 5000.5)
tdSql.checkData(0, 4, 50005000)
tdSql.checkData(0, 5, 9999.0)
tdSql.checkData(0, 6, 1.0)
tdSql.checkData(0, 7, 1)
tdSql.checkData(0, 8, 10000)
tdSql.query(f"select count(col14),min(col14),max(col14),avg(col14),sum(col14),spread(col14),percentile(col14, 0),first(col14),last(col14) from {dbname}.ntb")
tdSql.checkData(0, 0, 10000)
tdSql.checkData(0, 1, 1)
tdSql.checkData(0, 2, 10000)
tdSql.checkData(0, 3, 5000.5)
tdSql.checkData(0, 4, 50005000)
tdSql.checkData(0, 5, 9999.0)
tdSql.checkData(0, 6, 1.0)
tdSql.checkData(0, 7, 1)
tdSql.checkData(0, 8, 10000)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

16
tests/system-test/7-tmq/subscribeDb1.py
View File

@ -61,7 +61,7 @@ class TDTestCase:
def insertConsumerInfo(self,consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifmanualcommit,cdbName='cdb'):
sql = "insert into %s.consumeinfo values "%cdbName
sql += "(now, %d, '%s', '%s', %d, %d, %d)"%(consumerId, topicList, keyList, expectrowcnt, ifcheckdata, ifmanualcommit)
sql += "(now + %ds, %d, '%s', '%s', %d, %d, %d)"%(consumerId, consumerId, topicList, keyList, expectrowcnt, ifcheckdata, ifmanualcommit)
tdLog.info("consume info sql: %s"%sql)
tdSql.query(sql)
@ -174,12 +174,13 @@ class TDTestCase:
'ctbNum': 10, \
'rowsPerTbl': 5000, \
'batchNum': 100, \
'replica': self.replicaVar, \
'startTs': 1640966400000} # 2022-01-01 00:00:00.000
parameterDict['cfg'] = cfgPath
self.initConsumerTable()
tdSql.execute("create database if not exists %s vgroups %d" %(parameterDict['dbName'], parameterDict['vgroups']))
tdLog.info("create database if not exists %s vgroups %d replica %d" %(parameterDict['dbName'], parameterDict['vgroups'], parameterDict['replica']))
tdSql.execute("create database if not exists %s vgroups %d replica %d" %(parameterDict['dbName'], parameterDict['vgroups'], parameterDict['replica']))
prepareEnvThread = threading.Thread(target=self.prepareEnv, kwargs=parameterDict)
prepareEnvThread.start()
@ -191,10 +192,11 @@ class TDTestCase:
'ctbNum': 10, \
'rowsPerTbl': 5000, \
'batchNum': 100, \
'replica': self.replicaVar, \
'startTs': 1640966400000} # 2022-01-01 00:00:00.000
parameterDict['cfg'] = cfgPath
tdSql.execute("create database if not exists %s vgroups %d" %(parameterDict2['dbName'], parameterDict2['vgroups']))
tdSql.execute("create database if not exists %s vgroups %d replica %d" %(parameterDict2['dbName'], parameterDict2['vgroups'], parameterDict2['replica']))
prepareEnvThread2 = threading.Thread(target=self.prepareEnv, kwargs=parameterDict2)
prepareEnvThread2.start()
@ -259,12 +261,13 @@ class TDTestCase:
'ctbNum': 10, \
'rowsPerTbl': 5000, \
'batchNum': 100, \
'replica': self.replicaVar, \
'startTs': 1640966400000} # 2022-01-01 00:00:00.000
parameterDict['cfg'] = cfgPath
self.initConsumerTable()
tdSql.execute("create database if not exists %s vgroups %d" %(parameterDict['dbName'], parameterDict['vgroups']))
tdSql.execute("create database if not exists %s vgroups %d replica %d" %(parameterDict['dbName'], parameterDict['vgroups'], parameterDict['replica']))
prepareEnvThread = threading.Thread(target=self.prepareEnv, kwargs=parameterDict)
prepareEnvThread.start()
@ -276,10 +279,11 @@ class TDTestCase:
'ctbNum': 10, \
'rowsPerTbl': 5000, \
'batchNum': 100, \
'replica': self.replicaVar, \
'startTs': 1640966400000} # 2022-01-01 00:00:00.000
parameterDict['cfg'] = cfgPath
tdSql.execute("create database if not exists %s vgroups %d" %(parameterDict2['dbName'], parameterDict2['vgroups']))
tdSql.execute("create database if not exists %s vgroups %d replica %d" %(parameterDict2['dbName'], parameterDict2['vgroups'], parameterDict2['replica']))
prepareEnvThread2 = threading.Thread(target=self.prepareEnv, kwargs=parameterDict2)
prepareEnvThread2.start()

4
tests/system-test/7-tmq/tmqConsFromTsdb-mutilVg.py
View File

@ -117,7 +117,7 @@ class TDTestCase:
topicList = topicNameList[0]
ifcheckdata = 1
ifManualCommit = 1
keyList = 'group.id:cgrp1, enable.auto.commit:true, auto.commit.interval.ms:1000, auto.offset.reset:earliest'
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")
@ -186,7 +186,7 @@ class TDTestCase:
topicList = topicNameList[0]
ifcheckdata = 1
ifManualCommit = 1
keyList = 'group.id:cgrp1, enable.auto.commit:true, auto.commit.interval.ms:1000, auto.offset.reset:earliest'
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 0")

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

@ -53,7 +53,7 @@ class TDTestCase:
paraDict['rowsPerTbl'] = self.rowsPerTbl
tmqCom.initConsumerTable()
tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=1)
tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=self.replicaVar)
tdLog.info("create stb")
tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"])
tdLog.info("create ctb")
@ -116,7 +116,7 @@ class TDTestCase:
topicList = topicNameList[0]
ifcheckdata = 1
ifManualCommit = 1
keyList = 'group.id:cgrp1, enable.auto.commit:true, auto.commit.interval.ms:1000, auto.offset.reset:earliest'
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")
@ -185,7 +185,7 @@ class TDTestCase:
topicList = topicNameList[0]
ifcheckdata = 1
ifManualCommit = 1
keyList = 'group.id:cgrp1, enable.auto.commit:true, auto.commit.interval.ms:1000, auto.offset.reset:earliest'
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 0")

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

@ -53,7 +53,7 @@ class TDTestCase:
paraDict['rowsPerTbl'] = self.rowsPerTbl
tmqCom.initConsumerTable()
tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=1)
tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=self.replicaVar)
tdLog.info("create stb")
tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"])
tdLog.info("create ctb")
@ -116,7 +116,7 @@ class TDTestCase:
topicList = topicNameList[0]
ifcheckdata = 1
ifManualCommit = 1
keyList = 'group.id:cgrp1, enable.auto.commit:true, auto.commit.interval.ms:1000, auto.offset.reset:earliest'
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)
consumerId = 4
@ -188,7 +188,7 @@ class TDTestCase:
topicList = topicNameList[0]
ifcheckdata = 1
ifManualCommit = 1
keyList = 'group.id:cgrp1, enable.auto.commit:true, auto.commit.interval.ms:1000, auto.offset.reset:earliest'
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 0")

8
tests/system-test/7-tmq/tmqDelete-multiCtb.py
View File

@ -53,7 +53,7 @@ class TDTestCase:
paraDict['rowsPerTbl'] = self.rowsPerTbl
tmqCom.initConsumerTable()
tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=1,wal_retention_size=-1, wal_retention_period=-1)
tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=self.replicaVar,wal_retention_size=-1, wal_retention_period=-1)
tdLog.info("create stb")
tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"])
tdLog.info("create ctb")
@ -146,7 +146,7 @@ class TDTestCase:
ifManualCommit = 1
keyList = 'group.id:cgrp1,\
enable.auto.commit:true,\
auto.commit.interval.ms:1000,\
auto.commit.interval.ms:200,\
auto.offset.reset:earliest'
tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
@ -247,7 +247,7 @@ class TDTestCase:
ifManualCommit = 1
keyList = 'group.id:cgrp1,\
enable.auto.commit:true,\
auto.commit.interval.ms:1000,\
auto.commit.interval.ms:200,\
auto.offset.reset:earliest'
tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
@ -335,7 +335,7 @@ class TDTestCase:
ifManualCommit = 1
keyList = 'group.id:cgrp1,\
enable.auto.commit:true,\
auto.commit.interval.ms:1000,\
auto.commit.interval.ms:200,\
auto.offset.reset:earliest'
tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)

4
tests/system-test/7-tmq/tmqUpdateWithConsume.py
View File

@ -53,7 +53,7 @@ class TDTestCase:
paraDict['rowsPerTbl'] = self.rowsPerTbl
tmqCom.initConsumerTable()
tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=1, wal_retention_size=-1, wal_retention_period=-1)
tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=self.replicaVar, wal_retention_size=-1, wal_retention_period=-1)
tdLog.info("create stb")
tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"])
tdLog.info("create ctb")
@ -124,7 +124,7 @@ class TDTestCase:
ifManualCommit = 1
keyList = 'group.id:cgrp1,\
enable.auto.commit:true,\
auto.commit.interval.ms:1000,\
auto.commit.interval.ms:200,\
auto.offset.reset:earliest'
tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)