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

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

This commit is contained in:
Minghao Li 2022-10-27 16:22:08 +08:00
parent 04c0684149 8f88339988
commit bbb624b63e
19 changed files with 380 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cmake/taostools_CMakeLists.txt.in
View File

@ -2,7 +2,7 @@
# taos-tools
ExternalProject_Add(taos-tools
GIT_REPOSITORY https://github.com/taosdata/taos-tools.git
GIT_TAG f9c1d32
GIT_TAG f20eb34
SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools"
BINARY_DIR ""
#BUILD_IN_SOURCE TRUE

10
include/libs/executor/executor.h
View File

@ -89,6 +89,16 @@ qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* readers, int32_t* n
*/
int32_t qSetMultiStreamInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numOfBlocks, int32_t type);
/**
* Set block for sma
* @param tinfo
* @param pBlocks
* @param numOfInputBlock
* @param type
* @return
*/
int32_t qSetSMAInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numOfBlocks, int32_t type);
/**
* Update the table id list, add or remove.
*

5
source/dnode/mnode/impl/src/mndDnode.c
View File

@ -776,11 +776,6 @@ static int32_t mndProcessDropDnodeReq(SRpcMsg *pReq) {
}
}
if (numOfVnodes > 0) {
terrno = TSDB_CODE_OPS_NOT_SUPPORT;
goto _OVER;
}
code = mndDropDnode(pMnode, pReq, pDnode, pMObj, pQObj, pSObj, numOfVnodes);
if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;

44
source/dnode/mnode/impl/src/mndVgroup.c
View File

@ -1112,7 +1112,9 @@ int32_t mndSetMoveVgroupInfoToTrans(SMnode *pMnode, STrans *pTrans, SDbObj *pDb,
mInfo("vgId:%d, will add 1 vnodes", pVgroup->vgId);
if (mndAddVnodeToVgroup(pMnode, &newVg, pArray) != 0) return -1;
if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, &newVg, &newVg.vnodeGid[newVg.replica - 1]) != 0) return -1;
if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, &newVg, -1) != 0) return -1;
for (int32_t i = 0; i < newVg.replica - 1; ++i) {
if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, &newVg, newVg.vnodeGid[i].dnodeId) != 0) return -1;
}
if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, &newVg) != 0) return -1;
mInfo("vgId:%d, will remove 1 vnodes", pVgroup->vgId);
@ -1120,8 +1122,10 @@ int32_t mndSetMoveVgroupInfoToTrans(SMnode *pMnode, STrans *pTrans, SDbObj *pDb,
SVnodeGid del = newVg.vnodeGid[vnIndex];
newVg.vnodeGid[vnIndex] = newVg.vnodeGid[newVg.replica];
memset(&newVg.vnodeGid[newVg.replica], 0, sizeof(SVnodeGid));
if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, &newVg, -1) != 0) return -1;
if (mndAddDropVnodeAction(pMnode, pTrans, pDb, &newVg, &del, true) != 0) return -1;
for (int32_t i = 0; i < newVg.replica; ++i) {
if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, &newVg, newVg.vnodeGid[i].dnodeId) != 0) return -1;
}
if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, &newVg) != 0) return -1;
{
@ -1193,21 +1197,11 @@ static int32_t mndAddIncVgroupReplicaToTrans(SMnode *pMnode, STrans *pTrans, SDb
pGid->dnodeId = newDnodeId;
pGid->syncState = TAOS_SYNC_STATE_ERROR;
if (pVgroup->replica == 2) {
if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, pVgroup, pGid) != 0) return -1;
if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, pVgroup, pVgroup->vnodeGid[0].dnodeId) != 0) return -1;
if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, pVgroup) != 0) return -1;
} else if (pVgroup->replica == 4) {
if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, pVgroup, pGid) != 0) return -1;
if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, pVgroup, pVgroup->vnodeGid[0].dnodeId) != 0) return -1;
if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, pVgroup, pVgroup->vnodeGid[1].dnodeId) != 0) return -1;
if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, pVgroup, pVgroup->vnodeGid[2].dnodeId) != 0) return -1;
if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, pVgroup) != 0) return -1;
} else {
mError("vgId:%d, failed to add 1 vnode since invalid replica:%d", pVgroup->vgId, pVgroup->replica);
terrno = TSDB_CODE_MND_APP_ERROR;
return -1;
for (int32_t i = 0; i < pVgroup->replica - 1; ++i) {
if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, pVgroup, pVgroup->vnodeGid[i].dnodeId) != 0) return -1;
}
if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, pVgroup) != 0) return -1;
return 0;
}
@ -1232,21 +1226,11 @@ static int32_t mndAddDecVgroupReplicaFromTrans(SMnode *pMnode, STrans *pTrans, S
memcpy(pGid, &pVgroup->vnodeGid[pVgroup->replica], sizeof(SVnodeGid));
memset(&pVgroup->vnodeGid[pVgroup->replica], 0, sizeof(SVnodeGid));
if (pVgroup->replica == 1) {
if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, pVgroup, pVgroup->vnodeGid[0].dnodeId) != 0) return -1;
if (mndAddDropVnodeAction(pMnode, pTrans, pDb, pVgroup, &delGid, true) != 0) return -1;
if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, pVgroup) != 0) return -1;
} else if (pVgroup->replica == 3) {
if (mndAddDropVnodeAction(pMnode, pTrans, pDb, pVgroup, &delGid, true) != 0) return -1;
if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, pVgroup, pVgroup->vnodeGid[0].dnodeId) != 0) return -1;
if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, pVgroup, pVgroup->vnodeGid[1].dnodeId) != 0) return -1;
if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, pVgroup, pVgroup->vnodeGid[2].dnodeId) != 0) return -1;
if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, pVgroup) != 0) return -1;
} else {
mError("vgId:%d, failed to remove 1 vnode since invalid replica:%d", pVgroup->vgId, pVgroup->replica);
terrno = TSDB_CODE_MND_APP_ERROR;
return -1;
for (int32_t i = 0; i < pVgroup->replica; ++i) {
if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, pVgroup, pVgroup->vnodeGid[i].dnodeId) != 0) return -1;
}
if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, pVgroup) != 0) return -1;
return 0;
}
@ -1890,9 +1874,6 @@ _OVER:
}
static int32_t mndProcessBalanceVgroupMsg(SRpcMsg *pReq) {
#if 1
return TSDB_CODE_OPS_NOT_SUPPORT;
#else
SMnode *pMnode = pReq->info.node;
int32_t code = -1;
SArray *pArray = NULL;
@ -1941,7 +1922,6 @@ _OVER:
taosArrayDestroy(pArray);
return code;
#endif
}
bool mndVgroupInDb(SVgObj *pVgroup, int64_t dbUid) { return !pVgroup->isTsma && pVgroup->dbUid == dbUid; }

4
source/dnode/vnode/src/sma/smaRollup.c
View File

@ -839,7 +839,7 @@ static int32_t tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int32_t msgSize,
tdRsmaPrintSubmitReq(pSma, pReq);
}
#endif
if (qSetMultiStreamInput(qTaskInfo, pMsg, msgSize, inputType) < 0) {
if (qSetSMAInput(qTaskInfo, pMsg, msgSize, inputType) < 0) {
smaError("vgId:%d, rsma %" PRIi8 " qSetStreamInput failed since %s", SMA_VID(pSma), level, tstrerror(terrno));
return TSDB_CODE_FAILED;
}
@ -1404,7 +1404,7 @@ static int32_t tdRSmaFetchAllResult(SSma *pSma, SRSmaInfo *pInfo) {
pItem->nScanned = 0;
if ((terrno = qSetMultiStreamInput(taskInfo, &dataBlock, 1, STREAM_INPUT__DATA_BLOCK)) < 0) {
if ((terrno = qSetSMAInput(taskInfo, &dataBlock, 1, STREAM_INPUT__DATA_BLOCK)) < 0) {
goto _err;
}
if (tdRSmaExecAndSubmitResult(pSma, taskInfo, pItem, pInfo->pTSchema, pInfo->suid) < 0) {

55
source/dnode/vnode/src/tsdb/tsdbRead.c
View File

@ -169,14 +169,14 @@ static TSDBROW* getValidMemRow(SIterInfo* pIter, const SArray* pDelList, STsdbRe
static int32_t doMergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pScanInfo, STsdbReader* pReader,
SRowMerger* pMerger);
static int32_t doMergeRowsInLastBlock(SLastBlockReader* pLastBlockReader, STableBlockScanInfo* pScanInfo, int64_t ts,
SRowMerger* pMerger);
SRowMerger* pMerger, SVersionRange* pVerRange);
static int32_t doMergeRowsInBuf(SIterInfo* pIter, uint64_t uid, int64_t ts, SArray* pDelList, SRowMerger* pMerger,
STsdbReader* pReader);
static int32_t doAppendRowFromTSRow(SSDataBlock* pBlock, STsdbReader* pReader, STSRow* pTSRow, uint64_t uid);
static int32_t doAppendRowFromFileBlock(SSDataBlock* pResBlock, STsdbReader* pReader, SBlockData* pBlockData,
int32_t rowIndex);
static void setComposedBlockFlag(STsdbReader* pReader, bool composed);
static bool hasBeenDropped(const SArray* pDelList, int32_t* index, TSDBKEY* pKey, int32_t order);
static bool hasBeenDropped(const SArray* pDelList, int32_t* index, TSDBKEY* pKey, int32_t order, SVersionRange* pRange);
static int32_t doMergeMemTableMultiRows(TSDBROW* pRow, uint64_t uid, SIterInfo* pIter, SArray* pDelList,
STSRow** pTSRow, STsdbReader* pReader, bool* freeTSRow);
@ -1466,7 +1466,8 @@ static bool tryCopyDistinctRowFromFileBlock(STsdbReader* pReader, SBlockData* pB
return false;
}
static bool nextRowFromLastBlocks(SLastBlockReader* pLastBlockReader, STableBlockScanInfo* pBlockScanInfo) {
static bool nextRowFromLastBlocks(SLastBlockReader* pLastBlockReader, STableBlockScanInfo* pBlockScanInfo,
SVersionRange* pVerRange) {
while (1) {
bool hasVal = tMergeTreeNext(&pLastBlockReader->mergeTree);
if (!hasVal) {
@ -1475,7 +1476,8 @@ static bool nextRowFromLastBlocks(SLastBlockReader* pLastBlockReader, STableBloc
TSDBROW row = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
TSDBKEY k = TSDBROW_KEY(&row);
if (!hasBeenDropped(pBlockScanInfo->delSkyline, &pBlockScanInfo->lastBlockDelIndex, &k, pLastBlockReader->order)) {
if (!hasBeenDropped(pBlockScanInfo->delSkyline, &pBlockScanInfo->lastBlockDelIndex, &k, pLastBlockReader->order,
pVerRange)) {
return true;
}
}
@ -1483,7 +1485,7 @@ static bool nextRowFromLastBlocks(SLastBlockReader* pLastBlockReader, STableBloc
static bool tryCopyDistinctRowFromSttBlock(TSDBROW* fRow, SLastBlockReader* pLastBlockReader,
STableBlockScanInfo* pScanInfo, int64_t ts, STsdbReader* pReader) {
bool hasVal = nextRowFromLastBlocks(pLastBlockReader, pScanInfo);
bool hasVal = nextRowFromLastBlocks(pLastBlockReader, pScanInfo, &pReader->verRange);
if (hasVal) {
int64_t next1 = getCurrentKeyInLastBlock(pLastBlockReader);
if (next1 != ts) {
@ -1602,7 +1604,7 @@ static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo*
return code;
}
}
doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLast, &merge);
doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLast, &merge, &pReader->verRange);
}
if (minKey == k.ts) {
@ -1647,7 +1649,7 @@ static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo*
return code;
}
}
doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLast, &merge);
doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLast, &merge, &pReader->verRange);
}
if (minKey == key) {
@ -1699,7 +1701,7 @@ static int32_t doMergeFileBlockAndLastBlock(SLastBlockReader* pLastBlockReader,
TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
tRowMerge(&merge, &fRow1);
doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLastBlock, &merge);
doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLastBlock, &merge, &pReader->verRange);
code = tRowMergerGetRow(&merge, &pTSRow);
if (code != TSDB_CODE_SUCCESS) {
@ -1717,7 +1719,7 @@ static int32_t doMergeFileBlockAndLastBlock(SLastBlockReader* pLastBlockReader,
return code;
}
doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLastBlock, &merge);
doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLastBlock, &merge, &pReader->verRange);
ASSERT(mergeBlockData);
// merge with block data if ts == key
@ -1771,7 +1773,7 @@ static int32_t mergeFileBlockAndLastBlock(STsdbReader* pReader, SLastBlockReader
TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
tRowMerge(&merge, &fRow1);
doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, ts, &merge);
doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, ts, &merge, &pReader->verRange);
code = tRowMergerGetRow(&merge, &pTSRow);
if (code != TSDB_CODE_SUCCESS) {
@ -1882,7 +1884,7 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo*
}
}
doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLast, &merge);
doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLast, &merge, &pReader->verRange);
}
if (minKey == ik.ts) {
@ -1901,8 +1903,8 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo*
}
}
code = doMergeRowsInBuf(&pBlockScanInfo->iiter, pBlockScanInfo->uid, ik.ts, pBlockScanInfo->delSkyline,
&merge, pReader);
code = doMergeRowsInBuf(&pBlockScanInfo->iiter, pBlockScanInfo->uid, ik.ts, pBlockScanInfo->delSkyline, &merge,
pReader);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
@ -1973,7 +1975,7 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo*
return code;
}
}
doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLast, &merge);
doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLast, &merge, &pReader->verRange);
}
if (minKey == key) {
@ -2095,7 +2097,8 @@ static bool isValidFileBlockRow(SBlockData* pBlockData, SFileBlockDumpInfo* pDum
}
TSDBKEY k = {.ts = ts, .version = ver};
if (hasBeenDropped(pBlockScanInfo->delSkyline, &pBlockScanInfo->fileDelIndex, &k, pReader->order)) {
if (hasBeenDropped(pBlockScanInfo->delSkyline, &pBlockScanInfo->fileDelIndex, &k, pReader->order,
&pReader->verRange)) {
return false;
}
@ -2130,7 +2133,7 @@ static bool initLastBlockReader(SLastBlockReader* pLBlockReader, STableBlockScan
return false;
}
return nextRowFromLastBlocks(pLBlockReader, pScanInfo);
return nextRowFromLastBlocks(pLBlockReader, pScanInfo, &pReader->verRange);
}
static int64_t getCurrentKeyInLastBlock(SLastBlockReader* pLastBlockReader) {
@ -2225,8 +2228,8 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
pBlockScanInfo = taosHashGet(pReader->status.pTableMap, &pBlockInfo->uid, sizeof(pBlockInfo->uid));
if (pBlockScanInfo == NULL) {
code = TSDB_CODE_INVALID_PARA;
tsdbError("failed to locate the uid:%"PRIu64" in query table uid list, total tables:%d, %s",
pBlockInfo->uid, taosHashGetSize(pReader->status.pTableMap), pReader->idStr);
tsdbError("failed to locate the uid:%" PRIu64 " in query table uid list, total tables:%d, %s", pBlockInfo->uid,
taosHashGetSize(pReader->status.pTableMap), pReader->idStr);
goto _end;
}
@ -2859,7 +2862,7 @@ SVersionRange getQueryVerRange(SVnode* pVnode, SQueryTableDataCond* pCond, int8_
return (SVersionRange){.minVer = startVer, .maxVer = endVer};
}
bool hasBeenDropped(const SArray* pDelList, int32_t* index, TSDBKEY* pKey, int32_t order) {
bool hasBeenDropped(const SArray* pDelList, int32_t* index, TSDBKEY* pKey, int32_t order, SVersionRange* pVerRange) {
ASSERT(pKey != NULL);
if (pDelList == NULL) {
return false;
@ -2887,7 +2890,8 @@ bool hasBeenDropped(const SArray* pDelList, int32_t* index, TSDBKEY* pKey, int32
return false;
}
if (pCurrent->ts <= pKey->ts && pNext->ts >= pKey->ts && pCurrent->version >= pKey->version) {
if (pCurrent->ts <= pKey->ts && pNext->ts >= pKey->ts && pCurrent->version >= pKey->version &&
pVerRange->maxVer >= pCurrent->version) {
return true;
}
@ -2903,7 +2907,8 @@ bool hasBeenDropped(const SArray* pDelList, int32_t* index, TSDBKEY* pKey, int32
continue;
}
if (pCurrent->ts <= pKey->ts && pNext->ts >= pKey->ts && pCurrent->version >= pKey->version) {
if (pCurrent->ts <= pKey->ts && pNext->ts >= pKey->ts && pCurrent->version >= pKey->version &&
pVerRange->maxVer >= pCurrent->version) {
return true;
}
}
@ -2973,7 +2978,7 @@ TSDBROW* getValidMemRow(SIterInfo* pIter, const SArray* pDelList, STsdbReader* p
// it is a valid data version
if ((key.version <= pReader->verRange.maxVer && key.version >= pReader->verRange.minVer) &&
(!hasBeenDropped(pDelList, &pIter->index, &key, pReader->order))) {
(!hasBeenDropped(pDelList, &pIter->index, &key, pReader->order, &pReader->verRange))) {
return pRow;
}
@ -2992,7 +2997,7 @@ TSDBROW* getValidMemRow(SIterInfo* pIter, const SArray* pDelList, STsdbReader* p
}
if (key.version <= pReader->verRange.maxVer && key.version >= pReader->verRange.minVer &&
(!hasBeenDropped(pDelList, &pIter->index, &key, pReader->order))) {
(!hasBeenDropped(pDelList, &pIter->index, &key, pReader->order, &pReader->verRange))) {
return pRow;
}
}
@ -3130,9 +3135,9 @@ int32_t doMergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pSc
}
int32_t doMergeRowsInLastBlock(SLastBlockReader* pLastBlockReader, STableBlockScanInfo* pScanInfo, int64_t ts,
SRowMerger* pMerger) {
SRowMerger* pMerger, SVersionRange* pVerRange) {
pScanInfo->lastKey = ts;
while (nextRowFromLastBlocks(pLastBlockReader, pScanInfo)) {
while (nextRowFromLastBlocks(pLastBlockReader, pScanInfo, pVerRange)) {
int64_t next1 = getCurrentKeyInLastBlock(pLastBlockReader);
if (next1 == ts) {
TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);

61
source/libs/executor/src/executor.c
View File

@ -30,6 +30,46 @@ static void cleanupRefPool() {
taosCloseRef(ref);
}
static int32_t doSetSMABlock(SOperatorInfo* pOperator, void* input, size_t numOfBlocks, int32_t type, char* id) {
ASSERT(pOperator != NULL);
if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
if (pOperator->numOfDownstream == 0) {
qError("failed to find stream scan operator to set the input data block, %s" PRIx64, id);
return TSDB_CODE_QRY_APP_ERROR;
}
if (pOperator->numOfDownstream > 1) { // not handle this in join query
qError("join not supported for stream block scan, %s" PRIx64, id);
return TSDB_CODE_QRY_APP_ERROR;
}
pOperator->status = OP_NOT_OPENED;
return doSetSMABlock(pOperator->pDownstream[0], input, numOfBlocks, type, id);
} else {
pOperator->status = OP_NOT_OPENED;
SStreamScanInfo* pInfo = pOperator->info;
if (type == STREAM_INPUT__MERGED_SUBMIT) {
for (int32_t i = 0; i < numOfBlocks; i++) {
SSubmitReq* pReq = *(void**)POINTER_SHIFT(input, i * sizeof(void*));
taosArrayPush(pInfo->pBlockLists, &pReq);
}
pInfo->blockType = STREAM_INPUT__DATA_SUBMIT;
} else if (type == STREAM_INPUT__DATA_SUBMIT) {
taosArrayPush(pInfo->pBlockLists, &input);
pInfo->blockType = STREAM_INPUT__DATA_SUBMIT;
} else if (type == STREAM_INPUT__DATA_BLOCK) {
for (int32_t i = 0; i < numOfBlocks; ++i) {
SSDataBlock* pDataBlock = &((SSDataBlock*)input)[i];
taosArrayPush(pInfo->pBlockLists, &pDataBlock);
}
pInfo->blockType = STREAM_INPUT__DATA_BLOCK;
}
return TSDB_CODE_SUCCESS;
}
}
static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t numOfBlocks, int32_t type, char* id) {
ASSERT(pOperator != NULL);
if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
@ -100,6 +140,27 @@ int32_t qSetMultiStreamInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numO
return code;
}
int32_t qSetSMAInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numOfBlocks, int32_t type) {
if (tinfo == NULL) {
return TSDB_CODE_QRY_APP_ERROR;
}
if (pBlocks == NULL || numOfBlocks == 0) {
return TSDB_CODE_SUCCESS;
}
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
int32_t code = doSetSMABlock(pTaskInfo->pRoot, (void*)pBlocks, numOfBlocks, type, GET_TASKID(pTaskInfo));
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed to set the sma block data", GET_TASKID(pTaskInfo));
} else {
qDebug("%s set the sma block successfully", GET_TASKID(pTaskInfo));
}
return code;
}
qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* readers, int32_t* numOfCols, SSchemaWrapper** pSchema) {
if (msg == NULL) {
// create raw scan

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

@ -1738,8 +1738,10 @@ static int32_t filterDelBlockByUid(SSDataBlock* pDst, const SSDataBlock* pSrc, S
j++;
}
}
uint32_t cap = pDst->info.capacity;
pDst->info = pSrc->info;
pDst->info.rows = j;
pDst->info.capacity = cap;
return 0;
}

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

@ -1148,6 +1148,8 @@ SSyncNode* syncNodeOpen(SSyncInfo* pSyncInfo) {
pSyncNode->pRaftCfg = NULL;
}
// init by SSyncInfo
pSyncNode->vgId = pSyncInfo->vgId;
SSyncCfg* pCfg = &pSyncInfo->syncCfg;
sDebug("vgId:%d, replica:%d selfIndex:%d", pSyncNode->vgId, pCfg->replicaNum, pCfg->myIndex);
for (int32_t i = 0; i < pCfg->replicaNum; ++i) {
@ -1155,8 +1157,6 @@ SSyncNode* syncNodeOpen(SSyncInfo* pSyncInfo) {
sDebug("vgId:%d, index:%d ep:%s:%u", pSyncNode->vgId, i, pNode->nodeFqdn, pNode->nodePort);
}
// init by SSyncInfo
pSyncNode->vgId = pSyncInfo->vgId;
memcpy(pSyncNode->path, pSyncInfo->path, sizeof(pSyncNode->path));
snprintf(pSyncNode->raftStorePath, sizeof(pSyncNode->raftStorePath), "%s%sraft_store.json", pSyncInfo->path,
TD_DIRSEP);
@ -3429,7 +3429,7 @@ int32_t syncNodeDoCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endInde
// config change finish
if (pEntry->originalRpcType == TDMT_SYNC_CONFIG_CHANGE_FINISH) {
if (rpcMsg.pCont != NULL) {
if (rpcMsg.pCont != NULL && rpcMsg.contLen > 0) {
code = syncNodeConfigChangeFinish(ths, &rpcMsg, pEntry);
ASSERT(code == 0);
}

2
source/libs/sync/src/syncMessage.c
View File

@ -3026,7 +3026,7 @@ void syncReconfigFinishFromRpcMsg(const SRpcMsg* pRpcMsg, SyncReconfigFinish* pM
}
SyncReconfigFinish* syncReconfigFinishFromRpcMsg2(const SRpcMsg* pRpcMsg) {
SyncReconfigFinish* pMsg = syncReconfigFinishDeserialize2(pRpcMsg->pCont, pRpcMsg->contLen);
SyncReconfigFinish* pMsg = syncReconfigFinishDeserialize2(pRpcMsg->pCont, (uint32_t)(pRpcMsg->contLen));
ASSERT(pMsg != NULL);
return pMsg;
}

22
tests/script/jenkins/basic.txt
View File

@ -36,26 +36,26 @@
./test.sh -f tsim/db/taosdlog.sim
# ---- dnode
# unsupport ./test.sh -f tsim/dnode/balance_replica1.sim
# unsupport ./test.sh -f tsim/dnode/balance_replica3.sim
# unsupport ./test.sh -f tsim/dnode/balance1.sim
# unsupport ./test.sh -f tsim/dnode/balance2.sim
# unsupport ./test.sh -f tsim/dnode/balance3.sim
# unsupport ./test.sh -f tsim/dnode/balancex.sim
./test.sh -f tsim/dnode/balance_replica1.sim
./test.sh -f tsim/dnode/balance_replica3.sim
./test.sh -f tsim/dnode/balance1.sim
./test.sh -f tsim/dnode/balance2.sim
./test.sh -f tsim/dnode/balance3.sim
./test.sh -f tsim/dnode/balancex.sim
./test.sh -f tsim/dnode/create_dnode.sim
./test.sh -f tsim/dnode/drop_dnode_has_mnode.sim
# unsupport ./test.sh -f tsim/dnode/drop_dnode_has_qnode_snode.sim
# unsupport ./test.sh -f tsim/dnode/drop_dnode_has_vnode_replica1.sim
# unsupport ./test.sh -f tsim/dnode/drop_dnode_has_vnode_replica3.sim
# unsupport ./test.sh -f tsim/dnode/drop_dnode_has_multi_vnode_replica1.sim
# unsupport ./test.sh -f tsim/dnode/drop_dnode_has_multi_vnode_replica3.sim
./test.sh -f tsim/dnode/drop_dnode_has_vnode_replica1.sim
./test.sh -f tsim/dnode/drop_dnode_has_vnode_replica3.sim
./test.sh -f tsim/dnode/drop_dnode_has_multi_vnode_replica1.sim
./test.sh -f tsim/dnode/drop_dnode_has_multi_vnode_replica3.sim
./test.sh -f tsim/dnode/offline_reason.sim
./test.sh -f tsim/dnode/redistribute_vgroup_replica1.sim
./test.sh -f tsim/dnode/redistribute_vgroup_replica3_v1_leader.sim
./test.sh -f tsim/dnode/redistribute_vgroup_replica3_v1_follower.sim
./test.sh -f tsim/dnode/redistribute_vgroup_replica3_v2.sim
./test.sh -f tsim/dnode/redistribute_vgroup_replica3_v3.sim
# unsupport ./test.sh -f tsim/dnode/vnode_clean.sim
./test.sh -f tsim/dnode/vnode_clean.sim
./test.sh -f tsim/dnode/use_dropped_dnode.sim
# ---- import ----

108
tests/script/tsim/dnode/balance1.sim
View File

@ -81,6 +81,42 @@ if $data(2)[2] != 2 then
return -1
endi
sql select * from d1.t1 order by t desc
print $data01 $data11 $data21 $data31 $data41
if $data01 != 11 then
return -1
endi
if $data11 != 12 then
return -1
endi
if $data21 != 13 then
return -1
endi
if $data31 != 14 then
return -1
endi
if $data41 != 15 then
return -1
endi
sql select * from d2.t2 order by t desc
print $data01 $data11 $data21 $data31 $data41
if $data01 != 21 then
return -1
endi
if $data11 != 22 then
return -1
endi
if $data21 != 23 then
return -1
endi
if $data31 != 24 then
return -1
endi
if $data41 != 25 then
return -1
endi
print ========== step4
sql drop dnode 2
sql select * from information_schema.ins_dnodes
@ -93,6 +129,42 @@ if $data(2)[2] != null then
return -1
endi
sql select * from d1.t1 order by t desc
print $data01 $data11 $data21 $data31 $data41
if $data01 != 11 then
return -1
endi
if $data11 != 12 then
return -1
endi
if $data21 != 13 then
return -1
endi
if $data31 != 14 then
return -1
endi
if $data41 != 15 then
return -1
endi
sql select * from d2.t2 order by t desc
print $data01 $data11 $data21 $data31 $data41
if $data01 != 21 then
return -1
endi
if $data11 != 22 then
return -1
endi
if $data21 != 23 then
return -1
endi
if $data31 != 24 then
return -1
endi
if $data41 != 25 then
return -1
endi
system sh/exec.sh -n dnode2 -s stop -x SIGINT
print ========== step5
@ -131,6 +203,42 @@ if $data(3)[2] != 1 then
return -1
endi
sql select * from d1.t1 order by t desc
print $data01 $data11 $data21 $data31 $data41
if $data01 != 11 then
return -1
endi
if $data11 != 12 then
return -1
endi
if $data21 != 13 then
return -1
endi
if $data31 != 14 then
return -1
endi
if $data41 != 15 then
return -1
endi
sql select * from d2.t2 order by t desc
print $data01 $data11 $data21 $data31 $data41
if $data01 != 21 then
return -1
endi
if $data11 != 22 then
return -1
endi
if $data21 != 23 then
return -1
endi
if $data31 != 24 then
return -1
endi
if $data41 != 25 then
return -1
endi
print ========== step6
sql create database d3 vgroups 1
sql create table d3.t3 (t timestamp, i int)

52
tests/system-test/2-query/cast.py
View File

@ -653,6 +653,58 @@ class TDTestCase:
tdSql.query(f"select cast(c9 as timestamp ) as b from {self.dbname}.ct4")
tdSql.query(f"select cast(c9 as binary(64) ) as b from {self.dbname}.ct4")
# enh of cast function about coverage
tdSql.query(f"select cast(c1 as int) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(c1 as bool) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(c1 as tinyint) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(c1 as smallint) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(c1 as float) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(c1 as double) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(c1 as tinyint unsigned) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(c1 as smallint unsigned) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(c1 as int unsigned) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(c2 as int) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(c3 as bool) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(c4 as tinyint) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(c5 as smallint) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(c6 as float) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(c7 as double) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(c8 as tinyint unsigned) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(c8 as timestamp ) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(c9 as timestamp ) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(c9 as binary(64) ) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(abs(c2) as int) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(c3 as bool) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(floor(c4) as tinyint) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(c5+2 as smallint) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(2 as float) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(c7 as double) as b from {self.dbname}.stb1")
tdSql.query(f"select cast('123' as tinyint unsigned) as b from {self.dbname}.stb1")
tdSql.query(f"select max(cast(abs(c2) as int)) as b from {self.dbname}.stb1")
tdSql.query(f"select log(cast(c3 as int),2) as b from {self.dbname}.stb1")
tdSql.query(f"select abs(cast(floor(c4) as tinyint)) as b from {self.dbname}.stb1")
tdSql.query(f"select last(cast(c5+2 as smallint)) as b from {self.dbname}.stb1")
tdSql.query(f"select mavg(cast(2 as float),3) as b from {self.dbname}.stb1 partition by tbname")
tdSql.query(f"select cast(c7 as double) as b from {self.dbname}.stb1 partition by tbname order by tbname")
tdSql.query(f"select cast('123' as tinyint unsigned) as b from {self.dbname}.stb1 partition by tbname")
# uion with cast and common cols
tdSql.query(f"select cast(c2 as int) as b from {self.dbname}.stb1 union all select c1 from {self.dbname}.stb1 ")
tdSql.query(f"select cast(c3 as bool) as b from {self.dbname}.stb1 union all select c7 from {self.dbname}.ct1 ")
tdSql.query(f"select cast(c4 as tinyint) as b from {self.dbname}.stb1 union all select c4 from {self.dbname}.stb1")
tdSql.query(f"select cast(c5 as smallint) as b from {self.dbname}.stb1 union all select cast(c5 as smallint) as b from {self.dbname}.stb1")
tdSql.query(f"select cast(c6 as float) as b from {self.dbname}.stb1 union all select c5 from {self.dbname}.stb1")
tdSql.query(f"select cast(c7 as double) as b from {self.dbname}.stb1 union all select 123 from {self.dbname}.stb1 ")
tdSql.query(f"select cast(c8 as tinyint unsigned) as b from {self.dbname}.stb1 union all select last(cast(c8 as tinyint unsigned)) from {self.dbname}.stb1")
def run(self):
tdSql.prepare()

11
tests/system-test/2-query/last.py
View File

@ -17,6 +17,7 @@ class TDTestCase:
self.ts = 1537146000000
self.binary_str = 'taosdata'
self.nchar_str = '涛思数据'
self.cachemodel = None
def set_create_normaltable_sql(self, ntbname, column_dict):
column_sql = ''
@ -36,7 +37,8 @@ class TDTestCase:
return create_stb_sql
def last_check_stb_tb_base(self):
tdSql.prepare()
tdSql.execute(
f'create database if not exists db cachemodel "{self.cachemodel}"')
stbname = f'db.{tdCom.getLongName(5, "letters")}'
column_dict = {
'col1': 'tinyint',
@ -112,7 +114,8 @@ class TDTestCase:
tdSql.execute('drop database db')
def last_check_ntb_base(self):
tdSql.prepare()
tdSql.execute(
f'create database if not exists db cachemodel "{self.cachemodel}"')
ntbname = f'db.{tdCom.getLongName(5, "letters")}'
column_dict = {
'col1': 'tinyint',
@ -191,7 +194,7 @@ class TDTestCase:
}
tdSql.execute(
f"create database if not exists {dbname} vgroups {vgroup_num}")
f'create database if not exists {dbname} vgroups {vgroup_num} cachemodel "{self.cachemodel}"')
tdSql.execute(f'use {dbname}')
# build 20 child tables,every table insert 10 rows
@ -244,6 +247,8 @@ class TDTestCase:
tdSql.execute(f'drop database {dbname}')
def run(self):
for cachemodel in ["None", "last_row", "last_value", "both"]:
self.cachemodel = cachemodel
self.last_check_stb_tb_base()
self.last_check_ntb_base()
self.last_check_stb_distribute()

31
tests/system-test/2-query/log.py
View File

@ -229,6 +229,36 @@ class TDTestCase:
tdSql.query(f"select log(c6 ,2) from {dbname}.ct3")
tdSql.checkRows(0)
# log used for different param types
tdSql.query(f"select log(c1,c2) from {dbname}.ct1;")
tdSql.query(f"select log(c1,c2) from {dbname}.stb1 partition by tbname order by tbname;")
tdSql.query(f"select log(c1,2) from {dbname}.ct1;")
tdSql.query(f"select log(c1,2) from {dbname}.stb1 partition by tbname order by tbname;")
tdSql.query(f"select log(2,c2) from {dbname}.ct1;")
tdSql.query(f"select log(2,c2) from {dbname}.stb1 partition by tbname order by tbname;")
tdSql.query(f"select log(2,1) from {dbname}.ct1;")
tdSql.query(f"select log(2,2) from {dbname}.stb1 partition by tbname order by tbname;")
tdSql.query(f"select log(2,floor(1)) from {dbname}.ct1;")
tdSql.query(f"select log(2,floor(2)) from {dbname}.stb1 partition by tbname order by tbname;")
tdSql.query(f"select log(abs(2),floor(1)) from {dbname}.ct1;")
tdSql.query(f"select log(abs(2),floor(2)) from {dbname}.stb1 partition by tbname order by tbname;")
tdSql.query(f"select log(abs(c2),c1) from {dbname}.ct1;")
tdSql.query(f"select log(abs(c2),c1) from {dbname}.stb1 partition by tbname order by tbname;")
tdSql.query(f"select log(c2,abs(c1)) from {dbname}.ct1;")
tdSql.query(f"select log(c2,abs(c1)) from {dbname}.stb1 partition by tbname order by tbname;")
tdSql.query(f"select log(abs(c2),2) from {dbname}.ct1;")
tdSql.query(f"select log(abs(c2),2) from {dbname}.stb1 partition by tbname order by tbname;")
# # used for regular table
tdSql.query(f"select log(c1 ,2) from {dbname}.t1")
@ -292,6 +322,7 @@ class TDTestCase:
tdSql.checkRows(25)
# used for not exists table
tdSql.error(f"select log(c1, 2) from {dbname}.stbbb1")
tdSql.error(f"select log(c1, 2) from {dbname}tbname")

30
tests/system-test/2-query/pow.py
View File

@ -216,6 +216,36 @@ class TDTestCase:
tdSql.checkRows(0)
# pow used for different param types
tdSql.query(f"select pow(c1,c2) from {dbname}.ct1;")
tdSql.query(f"select pow(c1,c2) from {dbname}.stb1 partition by tbname order by tbname;")
tdSql.query(f"select pow(c1,2) from {dbname}.ct1;")
tdSql.query(f"select pow(c1,2) from {dbname}.stb1 partition by tbname order by tbname;")
tdSql.query(f"select pow(2,c2) from {dbname}.ct1;")
tdSql.query(f"select pow(2,c2) from {dbname}.stb1 partition by tbname order by tbname;")
tdSql.query(f"select pow(2,1) from {dbname}.ct1;")
tdSql.query(f"select pow(2,2) from {dbname}.stb1 partition by tbname order by tbname;")
tdSql.query(f"select pow(2,floor(1)) from {dbname}.ct1;")
tdSql.query(f"select pow(2,floor(2)) from {dbname}.stb1 partition by tbname order by tbname;")
tdSql.query(f"select pow(abs(2),floor(1)) from {dbname}.ct1;")
tdSql.query(f"select pow(abs(2),floor(2)) from {dbname}.stb1 partition by tbname order by tbname;")
tdSql.query(f"select pow(abs(c2),c1) from {dbname}.ct1;")
tdSql.query(f"select pow(abs(c2),c1) from {dbname}.stb1 partition by tbname order by tbname;")
tdSql.query(f"select pow(c2,abs(c1)) from {dbname}.ct1;")
tdSql.query(f"select pow(c2,abs(c1)) from {dbname}.stb1 partition by tbname order by tbname;")
tdSql.query(f"select pow(abs(c2),2) from {dbname}.ct1;")
tdSql.query(f"select pow(abs(c2),2) from {dbname}.stb1 partition by tbname order by tbname;")
# # used for regular table
tdSql.query(f"select pow(c1 ,2) from {dbname}.t1")
tdSql.checkData(0, 0, None)

2
tests/system-test/fulltest.sh
View File

@ -551,7 +551,7 @@ python3 ./test.py -f 2-query/upper.py -Q 4
python3 ./test.py -f 2-query/lower.py -Q 4
python3 ./test.py -f 2-query/join.py -Q 4
python3 ./test.py -f 2-query/join2.py -Q 4
python3 ./test.py -f 2-query/cast.py -Q 4
#python3 ./test.py -f 2-query/cast.py -Q 4
python3 ./test.py -f 2-query/substr.py -Q 4
python3 ./test.py -f 2-query/union.py -Q 4
python3 ./test.py -f 2-query/union1.py -Q 4

2
tools/CMakeLists.txt
View File

@ -12,7 +12,7 @@ IF (TD_WEBSOCKET)
ExternalProject_Add(taosws-rs
PREFIX "taosws-rs"
SOURCE_DIR ${CMAKE_CURRENT_SOURCE_DIR}/taosws-rs
BUILD_ALWAYS on
BUILD_ALWAYS off
DEPENDS taos
BUILD_IN_SOURCE 1
CONFIGURE_COMMAND cmake -E echo "taosws-rs no need cmake to config"