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

Merge branch '3.0' into fix/TD-20645

This commit is contained in:
Shengliang Guan 2022-11-25 00:12:33 +08:00
parent 0ea961b659 96df0cdd0d
commit 8239b5526c
44 changed files with 1059 additions and 375 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
docs/en/05-get-started/index.md
View File

@ -8,6 +8,7 @@ import DiscordSVG from './discord.svg'
import TwitterSVG from './twitter.svg'
import YouTubeSVG from './youtube.svg'
import LinkedInSVG from './linkedin.svg'
import StackOverflowSVG from './stackoverflow.svg'
You can install and run TDengine on Linux/Windows/macOS machines as well as Docker containers. You can also deploy TDengine as a managed service with TDengine Cloud.
@ -35,10 +36,19 @@ The TDengine Knowledge Map covers the various knowledge points of TDengine, reve
<table width="100%">
<tr align="center" style={{border:0}}>
<td width="20%" style={{border:0}}><a href="https://github.com/taosdata/TDengine" target="_blank"><GitHubSVG /><p>Star GitHub</p></a></td>
<td width="20%" style={{border:0}}><a href="https://discord.com/invite/VZdSuUg4pS" target="_blank"><DiscordSVG /><p>Join Discord</p></a></td>
<td width="20%" style={{border:0}}><a href="https://twitter.com/TDengineDB" target="_blank"><TwitterSVG /><p>Follow Twitter</p></a></td>
<td width="20%" style={{border:0}}><a href="https://www.youtube.com/@tdengine" target="_blank"><YouTubeSVG /><p>Subscribe YouTube</p></a></td>
<td width="20%" style={{border:0}}><a href="https://www.linkedin.com/company/tdengine" target="_blank"><LinkedInSVG /><p>Follow LinkedIn</p></a></td>
<td width="16%" style={{border:0}}><a href="https://github.com/taosdata/TDengine" target="_blank"><GitHubSVG /></a></td>
<td width="16%" style={{border:0}}><a href="https://discord.com/invite/VZdSuUg4pS" target="_blank"><DiscordSVG /></a></td>
<td width="16%" style={{border:0}}><a href="https://twitter.com/TDengineDB" target="_blank"><TwitterSVG /></a></td>
<td width="16%" style={{border:0}}><a href="https://www.youtube.com/@tdengine" target="_blank"><YouTubeSVG /></a></td>
<td width="16%" style={{border:0}}><a href="https://www.linkedin.com/company/tdengine" target="_blank"><LinkedInSVG /></a></td>
<td width="16%" style={{border:0}}><a href="https://stackoverflow.com/questions/tagged/tdengine" target="_blank"><StackOverflowSVG /></a></td>
</tr>
<tr align="center" style={{border:0,backgroundColor:'transparent'}}>
<td width="16%" style={{border:0,padding:0}}><a href="https://github.com/taosdata/TDengine" target="_blank">Star GitHub</a></td>
<td width="16%" style={{border:0,padding:0}}><a href="https://discord.com/invite/VZdSuUg4pS" target="_blank">Join Discord</a></td>
<td width="16%" style={{border:0,padding:0}}><a href="https://twitter.com/TDengineDB" target="_blank">Follow Twitter</a></td>
<td width="16%" style={{border:0,padding:0}}><a href="https://www.youtube.com/@tdengine" target="_blank">Subscribe YouTube</a></td>
<td width="16%" style={{border:0,padding:0}}><a href="https://www.linkedin.com/company/tdengine" target="_blank">Follow LinkedIn</a></td>
<td width="16%" style={{border:0,padding:0}}><a href="https://stackoverflow.com/questions/tagged/tdengine" target="_blank">Ask StackOverflow</a></td>
</tr>
</table>

7
docs/en/05-get-started/stackoverflow.svg Normal file
View File

@ -0,0 +1,7 @@
<svg xmlns="http://www.w3.org/2000/svg" viewBox="-8 0 48 48" width="50" height="50">
<path d="M26 41v-9h4v13H0V32h4v9h22z" fill="#BCBBBB" />
<path
d="M23 34l.8-3-16.1-3.3L7 31l16 3zM9.2 23.2l15 7 1.4-3-15-7-1.4 3zm4.2-7.4L26 26.4l2.1-2.5-12.7-10.6-2.1 2.5zM21.5 8l-2.7 2 9.9 13.3 2.7-2L21.5 8zM7 38h16v-3H7v3z"
fill="#F48024"
/>
</svg>
Side by Side

After

Width:  |  Height:  |  Size: 350 B

2
include/libs/nodes/plannodes.h
View File

@ -171,6 +171,7 @@ typedef struct SExchangeLogicNode {
SLogicNode node;
int32_t srcStartGroupId;
int32_t srcEndGroupId;
bool seqRecvData;
} SExchangeLogicNode;
typedef struct SMergeLogicNode {
@ -416,6 +417,7 @@ typedef struct SExchangePhysiNode {
int32_t srcEndGroupId;
bool singleChannel;
SNodeList* pSrcEndPoints; // element is SDownstreamSource, scheduler fill by calling qSetSuplanExecutionNode
bool seqRecvData;
} SExchangePhysiNode;
typedef struct SMergePhysiNode {

38
source/client/src/clientStmt.c
View File

@ -152,13 +152,13 @@ int32_t stmtRestoreQueryFields(STscStmt* pStmt) {
return TSDB_CODE_SUCCESS;
}
int32_t stmtUpdateBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, char* tbFName, const char* sTableName) {
int32_t stmtUpdateBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, char* tbFName, const char* sTableName, bool autoCreateTbl) {
STscStmt* pStmt = (STscStmt*)stmt;
strncpy(pStmt->bInfo.tbFName, tbFName, sizeof(pStmt->bInfo.tbFName) - 1);
pStmt->bInfo.tbFName[sizeof(pStmt->bInfo.tbFName) - 1] = 0;
pStmt->bInfo.tbUid = pTableMeta->uid;
pStmt->bInfo.tbUid = autoCreateTbl ? 0 : pTableMeta->uid;
pStmt->bInfo.tbSuid = pTableMeta->suid;
pStmt->bInfo.tbType = pTableMeta->tableType;
pStmt->bInfo.boundTags = tags;
@ -182,7 +182,7 @@ int32_t stmtUpdateInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, char
SHashObj* pVgHash, SHashObj* pBlockHash, const char* sTableName) {
STscStmt* pStmt = (STscStmt*)stmt;
STMT_ERR_RET(stmtUpdateBindInfo(stmt, pTableMeta, tags, tbFName, sTableName));
STMT_ERR_RET(stmtUpdateBindInfo(stmt, pTableMeta, tags, tbFName, sTableName, autoCreateTbl));
STMT_ERR_RET(stmtUpdateExecInfo(stmt, pVgHash, pBlockHash, autoCreateTbl));
pStmt->sql.autoCreateTbl = autoCreateTbl;
@ -623,6 +623,8 @@ int stmtSetTbTags(TAOS_STMT* stmt, TAOS_MULTI_BIND* tags) {
pStmt->bInfo.sname.tname, tags, pStmt->exec.pRequest->msgBuf,
pStmt->exec.pRequest->msgBufLen));
pStmt->exec.autoCreateTbl = true;
return TSDB_CODE_SUCCESS;
}
@ -771,10 +773,6 @@ int stmtAddBatch(TAOS_STMT* stmt) {
int stmtUpdateTableUid(STscStmt* pStmt, SSubmitRsp* pRsp) {
tscDebug("stmt start to update tbUid, blockNum: %d", pRsp->nBlocks);
if (pRsp->nBlocks <= 0) {
return TSDB_CODE_SUCCESS;
}
size_t keyLen = 0;
STableDataBlocks** pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
while (pIter) {
@ -809,8 +807,30 @@ int stmtUpdateTableUid(STscStmt* pStmt, SSubmitRsp* pRsp) {
pMeta->uid = blkRsp->uid;
pStmt->bInfo.tbUid = blkRsp->uid;
} else {
tscError("table %s not found in submit rsp", pStmt->bInfo.tbFName);
STMT_ERR_RET(TSDB_CODE_TSC_APP_ERROR);
tscDebug("table %s not found in submit rsp, will update from catalog", pStmt->bInfo.tbFName);
if (NULL == pStmt->pCatalog) {
STMT_ERR_RET(catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog));
}
STMT_ERR_RET(stmtCreateRequest(pStmt));
STableMeta* pTableMeta = NULL;
SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
.requestId = pStmt->exec.pRequest->requestId,
.requestObjRefId = pStmt->exec.pRequest->self,
.mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
int32_t code = catalogGetTableMeta(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &pTableMeta);
taos_free_result(pStmt->exec.pRequest);
pStmt->exec.pRequest = NULL;
if (TSDB_CODE_PAR_TABLE_NOT_EXIST == code) {
tscDebug("tb %s not exist", pStmt->bInfo.tbFName);
return TSDB_CODE_SUCCESS;
}
pMeta->uid = pTableMeta->uid;
pStmt->bInfo.tbUid = pTableMeta->uid;
}
pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);

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

@ -692,7 +692,6 @@ TEST(testCase, insert_test) {
taos_free_result(pRes);
taos_close(pConn);
}
#endif
TEST(testCase, projection_query_tables) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
@ -752,9 +751,6 @@ TEST(testCase, projection_query_tables) {
taos_close(pConn);
}
#if 0
TEST(testCase, tsbs_perf_test) {
TdThread qid[20] = {0};
@ -764,15 +760,16 @@ TEST(testCase, tsbs_perf_test) {
getchar();
}
#endif
TEST(testCase, projection_query_stables) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
ASSERT_NE(pConn, nullptr);
TAOS_RES* pRes = taos_query(pConn, "use abc1");
TAOS_RES* pRes = taos_query(pConn, "use test");
taos_free_result(pRes);
pRes = taos_query(pConn, "select ts from st1");
pRes = taos_query(pConn, "select * from meters limit 50000000");
if (taos_errno(pRes) != 0) {
printf("failed to select from table, reason:%s\n", taos_errstr(pRes));
taos_free_result(pRes);
@ -785,14 +782,15 @@ TEST(testCase, projection_query_stables) {
char str[512] = {0};
while ((pRow = taos_fetch_row(pRes)) != NULL) {
int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
printf("%s\n", str);
// int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
// printf("%s\n", str);
}
taos_free_result(pRes);
taos_close(pConn);
}
#if 0
TEST(testCase, agg_query_tables) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
ASSERT_NE(pConn, nullptr);

6
source/dnode/mnode/impl/src/mndScheduler.c
View File

@ -224,7 +224,7 @@ int32_t mndAddShuffleSinkTasksToStream(SMnode* pMnode, SStreamObj* pStream) {
ASSERT(taosArrayGetSize(pStream->tasks) == 1);
while (1) {
SVgObj* pVgroup;
SVgObj* pVgroup = NULL;
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void**)&pVgroup);
if (pIter == NULL) break;
if (!mndVgroupInDb(pVgroup, pStream->targetDbUid)) {
@ -258,6 +258,7 @@ int32_t mndAddShuffleSinkTasksToStream(SMnode* pMnode, SStreamObj* pStream) {
pTask->tbSink.pSchemaWrapper = tCloneSSchemaWrapper(&pStream->outputSchema);
ASSERT(pTask->tbSink.pSchemaWrapper);
}
sdbRelease(pSdb, pVgroup);
}
return 0;
}
@ -382,6 +383,7 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
qDestroyQueryPlan(pPlan);
return -1;
}
sdbRelease(pSdb, pVgroup);
} else {
if (mndAssignTaskToSnode(pMnode, pInnerTask, plan, pSnode) < 0) {
sdbRelease(pSdb, pSnode);
@ -396,6 +398,7 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
qDestroyQueryPlan(pPlan);
return -1;
}
sdbRelease(pSdb, pVgroup);
}
}
@ -459,6 +462,7 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
pEpInfo->nodeId = pTask->nodeId;
pEpInfo->taskId = pTask->taskId;
taosArrayPush(pInnerTask->childEpInfo, &pEpInfo);
sdbRelease(pSdb, pVgroup);
}
}

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

@ -1187,11 +1187,11 @@ static int32_t mndCheckAlterColForTopic(SMnode *pMnode, const char *stbFullName,
goto NEXT;
}
if (pCol->colId > 0 && pCol->colId == colId) {
sdbRelease(pSdb, pTopic);
nodesDestroyNode(pAst);
nodesDestroyList(pNodeList);
terrno = TSDB_CODE_MND_FIELD_CONFLICT_WITH_TOPIC;
mError("topic:%s, check colId:%d conflicted", pTopic->name, pCol->colId);
nodesDestroyNode(pAst);
nodesDestroyList(pNodeList);
sdbRelease(pSdb, pTopic);
return -1;
}
mInfo("topic:%s, check colId:%d passed", pTopic->name, pCol->colId);
@ -1230,11 +1230,11 @@ static int32_t mndCheckAlterColForStream(SMnode *pMnode, const char *stbFullName
goto NEXT;
}
if (pCol->colId > 0 && pCol->colId == colId) {
sdbRelease(pSdb, pStream);
nodesDestroyNode(pAst);
nodesDestroyList(pNodeList);
terrno = TSDB_CODE_MND_STREAM_MUST_BE_DELETED;
mError("stream:%s, check colId:%d conflicted", pStream->name, pCol->colId);
nodesDestroyNode(pAst);
nodesDestroyList(pNodeList);
sdbRelease(pSdb, pStream);
return -1;
}
mInfo("stream:%s, check colId:%d passed", pStream->name, pCol->colId);
@ -1279,11 +1279,11 @@ static int32_t mndCheckAlterColForTSma(SMnode *pMnode, const char *stbFullName,
goto NEXT;
}
if ((pCol->colId) > 0 && (pCol->colId == colId)) {
sdbRelease(pSdb, pSma);
nodesDestroyNode(pAst);
nodesDestroyList(pNodeList);
terrno = TSDB_CODE_MND_FIELD_CONFLICT_WITH_TSMA;
mError("tsma:%s, check colId:%d conflicted", pSma->name, pCol->colId);
nodesDestroyNode(pAst);
nodesDestroyList(pNodeList);
sdbRelease(pSdb, pSma);
return -1;
}
mInfo("tsma:%s, check colId:%d passed", pSma->name, pCol->colId);

5
source/dnode/vnode/src/meta/metaTable.c
View File

@ -981,6 +981,11 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA
/* get stbEntry*/
tdbTbGet(pMeta->pUidIdx, &ctbEntry.ctbEntry.suid, sizeof(tb_uid_t), &pVal, &nVal);
if (!pVal) {
terrno = TSDB_CODE_INVALID_MSG;
goto _err;
}
tdbTbGet(pMeta->pTbDb, &((STbDbKey){.uid = ctbEntry.ctbEntry.suid, .version = ((SUidIdxVal *)pVal)[0].version}),
sizeof(STbDbKey), (void **)&stbEntry.pBuf, &nVal);
tdbFree(pVal);

39
source/dnode/vnode/src/vnd/vnodeSync.c
View File

@ -298,35 +298,24 @@ static int32_t vnodeSyncGetSnapshot(const SSyncFSM *pFsm, SSnapshot *pSnapshot)
static void vnodeSyncApplyMsg(const SSyncFSM *pFsm, const SRpcMsg *pMsg, const SFsmCbMeta *pMeta) {
SVnode *pVnode = pFsm->data;
if (pMeta->code == 0) {
SRpcMsg rpcMsg = {.msgType = pMsg->msgType, .contLen = pMsg->contLen};
rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
memcpy(rpcMsg.pCont, pMsg->pCont, pMsg->contLen);
rpcMsg.info = pMsg->info;
rpcMsg.info.conn.applyIndex = pMeta->index;
rpcMsg.info.conn.applyTerm = pMeta->term;
SRpcMsg rpcMsg = {.msgType = pMsg->msgType, .contLen = pMsg->contLen};
rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
memcpy(rpcMsg.pCont, pMsg->pCont, pMsg->contLen);
rpcMsg.info = pMsg->info;
rpcMsg.info.conn.applyIndex = pMeta->index;
rpcMsg.info.conn.applyTerm = pMeta->term;
const STraceId *trace = &pMsg->info.traceId;
vGTrace("vgId:%d, commit-cb is excuted, fsm:%p, index:%" PRId64 ", term:%" PRIu64 ", msg-index:%" PRId64
", weak:%d, code:%d, state:%d %s, type:%s",
pVnode->config.vgId, pFsm, pMeta->index, pMeta->term, rpcMsg.info.conn.applyIndex, pMeta->isWeak,
pMeta->code, pMeta->state, syncStr(pMeta->state), TMSG_INFO(pMsg->msgType));
const STraceId *trace = &pMsg->info.traceId;
vGTrace("vgId:%d, commit-cb is excuted, fsm:%p, index:%" PRId64 ", term:%" PRIu64 ", msg-index:%" PRId64
", weak:%d, code:%d, state:%d %s, type:%s",
pVnode->config.vgId, pFsm, pMeta->index, pMeta->term, rpcMsg.info.conn.applyIndex, pMeta->isWeak, pMeta->code,
pMeta->state, syncStr(pMeta->state), TMSG_INFO(pMsg->msgType));
tmsgPutToQueue(&pVnode->msgCb, APPLY_QUEUE, &rpcMsg);
} else {
SRpcMsg rsp = {.code = pMeta->code, .info = pMsg->info};
vError("vgId:%d, commit-cb execute error, type:%s, index:%" PRId64 ", error:0x%x %s", pVnode->config.vgId,
TMSG_INFO(pMsg->msgType), pMeta->index, pMeta->code, tstrerror(pMeta->code));
if (rsp.info.handle != NULL) {
tmsgSendRsp(&rsp);
}
}
tmsgPutToQueue(&pVnode->msgCb, APPLY_QUEUE, &rpcMsg);
}
static void vnodeSyncCommitMsg(const SSyncFSM *pFsm, const SRpcMsg *pMsg, const SFsmCbMeta *pMeta) {
if (pMeta->isWeak == 0) {
vnodeSyncApplyMsg(pFsm, pMsg, pMeta);
}
vnodeSyncApplyMsg(pFsm, pMsg, pMeta);
}
static void vnodeSyncPreCommitMsg(const SSyncFSM *pFsm, const SRpcMsg *pMsg, const SFsmCbMeta *pMeta) {
@ -420,7 +409,7 @@ static void vnodeRestoreFinish(const SSyncFSM *pFsm) {
static void vnodeBecomeFollower(const SSyncFSM *pFsm) {
SVnode *pVnode = pFsm->data;
vDebug("vgId:%d, become follower", pVnode->config.vgId);
vInfo("vgId:%d, become follower", pVnode->config.vgId);
taosThreadMutexLock(&pVnode->lock);
if (pVnode->blocked) {

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

@ -138,7 +138,9 @@ static int32_t putDataBlock(SDataSinkHandle* pHandle, const SInputData* pInput,
toDataCacheEntry(pDispatcher, pInput, pBuf);
taosWriteQitem(pDispatcher->pDataBlocks, pBuf);
*pContinue = (DS_BUF_LOW == updateStatus(pDispatcher) ? true : false);
int32_t status = updateStatus(pDispatcher);
*pContinue = (status == DS_BUF_LOW || status == DS_BUF_EMPTY);
return TSDB_CODE_SUCCESS;
}

121
source/libs/executor/src/exchangeoperator.c
View File

@ -51,9 +51,9 @@ typedef struct SSourceDataInfo {
const char* taskId;
} SSourceDataInfo;
static void destroyExchangeOperatorInfo(void* param);
static void freeBlock(void* pParam);
static void freeSourceDataInfo(void* param);
static void destroyExchangeOperatorInfo(void* param);
static void freeBlock(void* pParam);
static void freeSourceDataInfo(void* param);
static void* setAllSourcesCompleted(SOperatorInfo* pOperator);
static int32_t loadRemoteDataCallback(void* param, SDataBuf* pMsg, int32_t code);
@ -62,7 +62,9 @@ static int32_t getCompletedSources(const SArray* pArray);
static int32_t prepareConcurrentlyLoad(SOperatorInfo* pOperator);
static int32_t seqLoadRemoteData(SOperatorInfo* pOperator);
static int32_t prepareLoadRemoteData(SOperatorInfo* pOperator);
static int32_t handleLimitOffset(SOperatorInfo* pOperator, SLimitInfo* pLimitInfo, SSDataBlock* pBlock, bool holdDataInBuf);
static int32_t handleLimitOffset(SOperatorInfo* pOperator, SLimitInfo* pLimitInfo, SSDataBlock* pBlock,
bool holdDataInBuf);
static int32_t doExtractResultBlocks(SExchangeInfo* pExchangeInfo, SSourceDataInfo* pDataInfo);
static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeInfo* pExchangeInfo,
SExecTaskInfo* pTaskInfo) {
@ -75,7 +77,9 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn
}
while (1) {
qDebug("prepare wait for ready, %p, %s", pExchangeInfo, GET_TASKID(pTaskInfo));
tsem_wait(&pExchangeInfo->ready);
if (isTaskKilled(pTaskInfo)) {
longjmp(pTaskInfo->env, TSDB_CODE_TSC_QUERY_CANCELLED);
}
@ -103,41 +107,33 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn
if (pRsp->numOfRows == 0) {
pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED;
qDebug("%s vgId:%d, taskId:0x%" PRIx64 " execId:%d index:%d completed, rowsOfSource:%" PRIu64
", totalRows:%" PRIu64 ", try next %d/%" PRIzu,
", totalRows:%" PRIu64 ", try next %d/%" PRIzu,
GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, i, pDataInfo->totalRows,
pExchangeInfo->loadInfo.totalRows, i + 1, totalSources);
taosMemoryFreeClear(pDataInfo->pRsp);
break;
}
SRetrieveTableRsp* pRetrieveRsp = pDataInfo->pRsp;
int32_t index = 0;
char* pStart = pRetrieveRsp->data;
while (index++ < pRetrieveRsp->numOfBlocks) {
SSDataBlock* pb = createOneDataBlock(pExchangeInfo->pDummyBlock, false);
code = extractDataBlockFromFetchRsp(pb, pStart, NULL, &pStart);
if (code != 0) {
taosMemoryFreeClear(pDataInfo->pRsp);
goto _error;
}
taosArrayPush(pExchangeInfo->pResultBlockList, &pb);
code = doExtractResultBlocks(pExchangeInfo, pDataInfo);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}
SRetrieveTableRsp* pRetrieveRsp = pDataInfo->pRsp;
updateLoadRemoteInfo(pLoadInfo, pRetrieveRsp->numOfRows, pRetrieveRsp->compLen, pDataInfo->startTime, pOperator);
pDataInfo->totalRows += pRetrieveRsp->numOfRows;
if (pRsp->completed == 1) {
pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED;
qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64
" execId:%d index:%d completed, blocks:%d, numOfRows:%d, rowsOfSource:%" PRIu64 ", totalRows:%" PRIu64
", total:%.2f Kb, try next %d/%" PRIzu,
" execId:%d index:%d completed, blocks:%d, numOfRows:%d, rowsOfSource:%" PRIu64 ", totalRows:%" PRIu64
", total:%.2f Kb, try next %d/%" PRIzu,
GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, i, pRsp->numOfBlocks,
pRsp->numOfRows, pDataInfo->totalRows, pLoadInfo->totalRows, pLoadInfo->totalSize / 1024.0,
i + 1, totalSources);
pRsp->numOfRows, pDataInfo->totalRows, pLoadInfo->totalRows, pLoadInfo->totalSize / 1024.0, i + 1,
totalSources);
} else {
qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64
" execId:%d blocks:%d, numOfRows:%d, totalRows:%" PRIu64 ", total:%.2f Kb",
" execId:%d blocks:%d, numOfRows:%d, totalRows:%" PRIu64 ", total:%.2f Kb",
GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, pRsp->numOfBlocks,
pRsp->numOfRows, pLoadInfo->totalRows, pLoadInfo->totalSize / 1024.0);
}
@ -162,7 +158,7 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn
}
}
_error:
_error:
pTaskInfo->code = code;
}
@ -304,16 +300,18 @@ SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, SExchangePhysiNode
SExchangeOpStopInfo stopInfo = {QUERY_NODE_PHYSICAL_PLAN_EXCHANGE, pInfo->self};
qAppendTaskStopInfo(pTaskInfo, &stopInfo);
pInfo->seqLoadData = false;
pInfo->seqLoadData = pExNode->seqRecvData;
pInfo->pTransporter = pTransporter;
setOperatorInfo(pOperator, "ExchangeOperator", QUERY_NODE_PHYSICAL_PLAN_EXCHANGE, false, OP_NOT_OPENED, pInfo, pTaskInfo);
setOperatorInfo(pOperator, "ExchangeOperator", QUERY_NODE_PHYSICAL_PLAN_EXCHANGE, false, OP_NOT_OPENED, pInfo,
pTaskInfo);
pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pDummyBlock->pDataBlock);
pOperator->fpSet = createOperatorFpSet(prepareLoadRemoteData, doLoadRemoteData, NULL, destroyExchangeOperatorInfo, NULL);
pOperator->fpSet =
createOperatorFpSet(prepareLoadRemoteData, doLoadRemoteData, NULL, destroyExchangeOperatorInfo, NULL);
return pOperator;
_error:
_error:
if (pInfo != NULL) {
doDestroyExchangeOperatorInfo(pInfo);
}
@ -360,7 +358,7 @@ int32_t loadRemoteDataCallback(void* param, SDataBuf* pMsg, int32_t code) {
SExchangeInfo* pExchangeInfo = taosAcquireRef(exchangeObjRefPool, pWrapper->exchangeId);
if (pExchangeInfo == NULL) {
qWarn("failed to acquire exchange operator, since it may have been released");
qWarn("failed to acquire exchange operator, since it may have been released, %p", pExchangeInfo);
taosMemoryFree(pMsg->pData);
return TSDB_CODE_SUCCESS;
}
@ -379,20 +377,24 @@ int32_t loadRemoteDataCallback(void* param, SDataBuf* pMsg, int32_t code) {
pRsp->numOfBlocks = htonl(pRsp->numOfBlocks);
ASSERT(pRsp != NULL);
qDebug("%s fetch rsp received, index:%d, blocks:%d, rows:%d", pSourceDataInfo->taskId, index, pRsp->numOfBlocks,
pRsp->numOfRows);
qDebug("%s fetch rsp received, index:%d, blocks:%d, rows:%d, %p", pSourceDataInfo->taskId, index, pRsp->numOfBlocks,
pRsp->numOfRows, pExchangeInfo);
} else {
taosMemoryFree(pMsg->pData);
pSourceDataInfo->code = code;
qDebug("%s fetch rsp received, index:%d, error:%s", pSourceDataInfo->taskId, index, tstrerror(code));
qDebug("%s fetch rsp received, index:%d, error:%s, %p", pSourceDataInfo->taskId, index, tstrerror(code),
pExchangeInfo);
}
pSourceDataInfo->status = EX_SOURCE_DATA_READY;
code = tsem_post(&pExchangeInfo->ready);
if (code != TSDB_CODE_SUCCESS) {
code = TAOS_SYSTEM_ERROR(code);
qError("failed to invoke post when fetch rsp is ready, code:%s, %p", tstrerror(code), pExchangeInfo);
}
tsem_post(&pExchangeInfo->ready);
taosReleaseRef(exchangeObjRefPool, pWrapper->exchangeId);
return TSDB_CODE_SUCCESS;
return code;
}
int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInfo* pTaskInfo, int32_t sourceIndex) {
@ -444,9 +446,9 @@ int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInfo* pTas
return pTaskInfo->code;
}
qDebug("%s build fetch msg and send to vgId:%d, ep:%s, taskId:0x%" PRIx64 ", execId:%d, %d/%" PRIzu,
qDebug("%s build fetch msg and send to vgId:%d, ep:%s, taskId:0x%" PRIx64 ", execId:%d, %p, %d/%" PRIzu,
GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->addr.epSet.eps[0].fqdn, pSource->taskId,
pSource->execId, sourceIndex, totalSources);
pSource->execId, pExchangeInfo, sourceIndex, totalSources);
// send the fetch remote task result reques
SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
@ -526,7 +528,7 @@ void* setAllSourcesCompleted(SOperatorInfo* pOperator) {
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SLoadRemoteDataInfo* pLoadInfo = &pExchangeInfo->loadInfo;
size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources);
size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources);
qDebug("%s all %" PRIzu " sources are exhausted, total rows: %" PRIu64 ", %.2f Kb, elapsed:%.2f ms",
GET_TASKID(pTaskInfo), totalSources, pLoadInfo->totalRows, pLoadInfo->totalSize / 1024.0,
pLoadInfo->totalElapsed / 1000.0);
@ -581,10 +583,32 @@ int32_t prepareConcurrentlyLoad(SOperatorInfo* pOperator) {
return TSDB_CODE_SUCCESS;
}
int32_t doExtractResultBlocks(SExchangeInfo* pExchangeInfo, SSourceDataInfo* pDataInfo) {
SRetrieveTableRsp* pRetrieveRsp = pDataInfo->pRsp;
char* pStart = pRetrieveRsp->data;
int32_t index = 0;
int32_t code = 0;
while (index++ < pRetrieveRsp->numOfBlocks) {
SSDataBlock* pb = createOneDataBlock(pExchangeInfo->pDummyBlock, false);
code = extractDataBlockFromFetchRsp(pb, pStart, NULL, &pStart);
if (code != 0) {
taosMemoryFreeClear(pDataInfo->pRsp);
return code;
}
taosArrayPush(pExchangeInfo->pResultBlockList, &pb);
}
return code;
}
int32_t seqLoadRemoteData(SOperatorInfo* pOperator) {
SExchangeInfo* pExchangeInfo = pOperator->info;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
int32_t code = 0;
size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources);
int64_t startTs = taosGetTimestampUs();
@ -594,13 +618,15 @@ int32_t seqLoadRemoteData(SOperatorInfo* pOperator) {
return TSDB_CODE_SUCCESS;
}
SSourceDataInfo* pDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, pExchangeInfo->current);
pDataInfo->status = EX_SOURCE_DATA_NOT_READY;
doSendFetchDataRequest(pExchangeInfo, pTaskInfo, pExchangeInfo->current);
tsem_wait(&pExchangeInfo->ready);
if (isTaskKilled(pTaskInfo)) {
longjmp(pTaskInfo->env, TSDB_CODE_TSC_QUERY_CANCELLED);
}
SSourceDataInfo* pDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, pExchangeInfo->current);
SDownstreamSourceNode* pSource = taosArrayGet(pExchangeInfo->pSources, pExchangeInfo->current);
if (pDataInfo->code != TSDB_CODE_SUCCESS) {
@ -614,7 +640,7 @@ int32_t seqLoadRemoteData(SOperatorInfo* pOperator) {
SLoadRemoteDataInfo* pLoadInfo = &pExchangeInfo->loadInfo;
if (pRsp->numOfRows == 0) {
qDebug("%s vgId:%d, taskID:0x%" PRIx64 " execId:%d %d of total completed, rowsOfSource:%" PRIu64
", totalRows:%" PRIu64 " try next",
", totalRows:%" PRIu64 " try next",
GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, pExchangeInfo->current + 1,
pDataInfo->totalRows, pLoadInfo->totalRows);
@ -624,14 +650,15 @@ int32_t seqLoadRemoteData(SOperatorInfo* pOperator) {
continue;
}
code = doExtractResultBlocks(pExchangeInfo, pDataInfo);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}
SRetrieveTableRsp* pRetrieveRsp = pDataInfo->pRsp;
char* pStart = pRetrieveRsp->data;
int32_t code = extractDataBlockFromFetchRsp(NULL, pStart, NULL, &pStart);
if (pRsp->completed == 1) {
qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64 " execId:%d numOfRows:%d, rowsOfSource:%" PRIu64
", totalRows:%" PRIu64 ", totalBytes:%" PRIu64 " try next %d/%" PRIzu,
", totalRows:%" PRIu64 ", totalBytes:%" PRIu64 " try next %d/%" PRIzu,
GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, pRetrieveRsp->numOfRows,
pDataInfo->totalRows, pLoadInfo->totalRows, pLoadInfo->totalSize, pExchangeInfo->current + 1,
totalSources);
@ -640,7 +667,7 @@ int32_t seqLoadRemoteData(SOperatorInfo* pOperator) {
pExchangeInfo->current += 1;
} else {
qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64 " execId:%d numOfRows:%d, totalRows:%" PRIu64
", totalBytes:%" PRIu64,
", totalBytes:%" PRIu64,
GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, pRetrieveRsp->numOfRows,
pLoadInfo->totalRows, pLoadInfo->totalSize);
}
@ -651,6 +678,10 @@ int32_t seqLoadRemoteData(SOperatorInfo* pOperator) {
taosMemoryFreeClear(pDataInfo->pRsp);
return TSDB_CODE_SUCCESS;
}
_error:
pTaskInfo->code = code;
return code;
}
int32_t prepareLoadRemoteData(SOperatorInfo* pOperator) {

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

@ -1849,12 +1849,12 @@ FETCH_NEXT_BLOCK:
prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex);
copyDataBlock(pInfo->pDeleteDataRes, pInfo->pUpdateRes);
pInfo->pDeleteDataRes->info.type = STREAM_DELETE_DATA;
pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RANGE;
printDataBlock(pDelBlock, "stream scan delete data");
if (pInfo->tqReader) {
blockDataDestroy(pDelBlock);
}
if (pInfo->pDeleteDataRes->info.rows > 0) {
pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RANGE;
return pInfo->pDeleteDataRes;
} else {
goto FETCH_NEXT_BLOCK;

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

@ -3364,22 +3364,23 @@ void initDummyFunction(SqlFunctionCtx* pDummy, SqlFunctionCtx* pCtx, int32_t num
}
}
void initDownStream(SOperatorInfo* downstream, SStreamAggSupporter* pAggSup, int64_t waterMark, uint16_t type,
int32_t tsColIndex) {
void initDownStream(SOperatorInfo* downstream, SStreamAggSupporter* pAggSup, uint16_t type, int32_t tsColIndex,
STimeWindowAggSupp* pTwSup) {
if (downstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION) {
SStreamPartitionOperatorInfo* pScanInfo = downstream->info;
pScanInfo->tsColIndex = tsColIndex;
}
if (downstream->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
initDownStream(downstream->pDownstream[0], pAggSup, waterMark, type, tsColIndex);
initDownStream(downstream->pDownstream[0], pAggSup, type, tsColIndex, pTwSup);
return;
}
SStreamScanInfo* pScanInfo = downstream->info;
pScanInfo->windowSup = (SWindowSupporter){.pStreamAggSup = pAggSup, .gap = pAggSup->gap, .parentType = type};
if (!pScanInfo->pUpdateInfo) {
pScanInfo->pUpdateInfo = updateInfoInit(60000, TSDB_TIME_PRECISION_MILLI, waterMark);
pScanInfo->pUpdateInfo = updateInfoInit(60000, TSDB_TIME_PRECISION_MILLI, pTwSup->waterMark);
}
pScanInfo->twAggSup = *pTwSup;
}
int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, SqlFunctionCtx* pCtx, int32_t numOfOutput, int64_t gap,
@ -3654,6 +3655,11 @@ static void doStreamSessionAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSData
setSessionWinOutputInfo(pStUpdated, &winInfo);
winRows = updateSessionWindowInfo(&winInfo, startTsCols, endTsCols, groupId, rows, i, pAggSup->gap,
pAggSup->pResultRows, pStUpdated, pStDeleted);
// coverity scan error
if (!winInfo.pOutputBuf) {
T_LONG_JMP(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
}
code = doOneWindowAggImpl(&pInfo->twAggSup.timeWindowData, &winInfo, &pResult, i, winRows, rows, numOfOutput,
pOperator);
if (code != TSDB_CODE_SUCCESS || pResult == NULL) {
@ -4102,8 +4108,7 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh
createOperatorFpSet(operatorDummyOpenFn, doStreamSessionAgg, NULL, destroyStreamSessionAggOperatorInfo, NULL);
if (downstream) {
initDownStream(downstream, &pInfo->streamAggSup, pInfo->twAggSup.waterMark, pOperator->operatorType,
pInfo->primaryTsIndex);
initDownStream(downstream, &pInfo->streamAggSup, pOperator->operatorType, pInfo->primaryTsIndex, &pInfo->twAggSup);
code = appendDownstream(pOperator, &downstream, 1);
}
return pOperator;
@ -4606,8 +4611,7 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys
pInfo, pTaskInfo);
pOperator->fpSet =
createOperatorFpSet(operatorDummyOpenFn, doStreamStateAgg, NULL, destroyStreamStateOperatorInfo, NULL);
initDownStream(downstream, &pInfo->streamAggSup, pInfo->twAggSup.waterMark, pOperator->operatorType,
pInfo->primaryTsIndex);
initDownStream(downstream, &pInfo->streamAggSup, pOperator->operatorType, pInfo->primaryTsIndex, &pInfo->twAggSup);
code = appendDownstream(pOperator, &downstream, 1);
if (code != TSDB_CODE_SUCCESS) {
goto _error;

1
source/libs/nodes/src/nodesCloneFuncs.c
View File

@ -434,6 +434,7 @@ static int32_t logicExchangeCopy(const SExchangeLogicNode* pSrc, SExchangeLogicN
COPY_BASE_OBJECT_FIELD(node, logicNodeCopy);
COPY_SCALAR_FIELD(srcStartGroupId);
COPY_SCALAR_FIELD(srcEndGroupId);
COPY_SCALAR_FIELD(seqRecvData);
return TSDB_CODE_SUCCESS;
}

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

@ -1864,6 +1864,7 @@ static int32_t jsonToPhysiAggNode(const SJson* pJson, void* pObj) {
static const char* jkExchangePhysiPlanSrcStartGroupId = "SrcStartGroupId";
static const char* jkExchangePhysiPlanSrcEndGroupId = "SrcEndGroupId";
static const char* jkExchangePhysiPlanSrcEndPoints = "SrcEndPoints";
static const char* jkExchangePhysiPlanSeqRecvData = "SeqRecvData";
static int32_t physiExchangeNodeToJson(const void* pObj, SJson* pJson) {
const SExchangePhysiNode* pNode = (const SExchangePhysiNode*)pObj;
@ -1878,6 +1879,9 @@ static int32_t physiExchangeNodeToJson(const void* pObj, SJson* pJson) {
if (TSDB_CODE_SUCCESS == code) {
code = nodeListToJson(pJson, jkExchangePhysiPlanSrcEndPoints, pNode->pSrcEndPoints);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddBoolToObject(pJson, jkExchangePhysiPlanSeqRecvData, pNode->seqRecvData);
}
return code;
}
@ -1895,6 +1899,9 @@ static int32_t jsonToPhysiExchangeNode(const SJson* pJson, void* pObj) {
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeList(pJson, jkExchangePhysiPlanSrcEndPoints, &pNode->pSrcEndPoints);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBoolValue(pJson, jkExchangePhysiPlanSeqRecvData, &pNode->seqRecvData);
}
return code;
}

9
source/libs/nodes/src/nodesMsgFuncs.c
View File

@ -2428,7 +2428,8 @@ enum {
PHY_EXCHANGE_CODE_SRC_START_GROUP_ID,
PHY_EXCHANGE_CODE_SRC_END_GROUP_ID,
PHY_EXCHANGE_CODE_SINGLE_CHANNEL,
PHY_EXCHANGE_CODE_SRC_ENDPOINTS
PHY_EXCHANGE_CODE_SRC_ENDPOINTS,
PHY_EXCHANGE_CODE_SEQ_RECV_DATA
};
static int32_t physiExchangeNodeToMsg(const void* pObj, STlvEncoder* pEncoder) {
@ -2447,6 +2448,9 @@ static int32_t physiExchangeNodeToMsg(const void* pObj, STlvEncoder* pEncoder) {
if (TSDB_CODE_SUCCESS == code) {
code = tlvEncodeObj(pEncoder, PHY_EXCHANGE_CODE_SRC_ENDPOINTS, nodeListToMsg, pNode->pSrcEndPoints);
}
if (TSDB_CODE_SUCCESS == code) {
code = tlvEncodeBool(pEncoder, PHY_EXCHANGE_CODE_SEQ_RECV_DATA, pNode->seqRecvData);
}
return code;
}
@ -2473,6 +2477,9 @@ static int32_t msgToPhysiExchangeNode(STlvDecoder* pDecoder, void* pObj) {
case PHY_EXCHANGE_CODE_SRC_ENDPOINTS:
code = msgToNodeListFromTlv(pTlv, (void**)&pNode->pSrcEndPoints);
break;
case PHY_EXCHANGE_CODE_SEQ_RECV_DATA:
code = tlvDecodeBool(pTlv, &pNode->seqRecvData);
break;
default:
break;
}

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

@ -918,7 +918,12 @@ static int32_t preParseBoundColumnsClause(SInsertParseContext* pCxt, SVnodeModif
static int32_t getTableDataBlocks(SInsertParseContext* pCxt, SVnodeModifOpStmt* pStmt, STableDataBlocks** pDataBuf) {
if (pCxt->pComCxt->async) {
return insGetDataBlockFromList(pStmt->pTableBlockHashObj, &pStmt->pTableMeta->uid, sizeof(pStmt->pTableMeta->uid),
uint64_t uid = pStmt->pTableMeta->uid;
if (pStmt->usingTableProcessing) {
pStmt->pTableMeta->uid = 0;
}
return insGetDataBlockFromList(pStmt->pTableBlockHashObj, &uid, sizeof(pStmt->pTableMeta->uid),
TSDB_DEFAULT_PAYLOAD_SIZE, sizeof(SSubmitBlk),
getTableInfo(pStmt->pTableMeta).rowSize, pStmt->pTableMeta, pDataBuf, NULL,
&pStmt->createTblReq);

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

@ -3685,9 +3685,19 @@ static int32_t translateInsertProject(STranslateContext* pCxt, SInsertStmt* pIns
return addOrderByPrimaryKeyToQuery(pCxt, pPrimaryKeyExpr, pInsert->pQuery);
}
static int32_t translateInsertTable(STranslateContext* pCxt, SNode* pTable) {
int32_t code = translateFrom(pCxt, pTable);
if (TSDB_CODE_SUCCESS == code && TSDB_CHILD_TABLE != ((SRealTableNode*)pTable)->pMeta->tableType &&
TSDB_NORMAL_TABLE != ((SRealTableNode*)pTable)->pMeta->tableType) {
code = generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_SYNTAX_ERROR,
"insert data into super table is not supported");
}
return code;
}
static int32_t translateInsert(STranslateContext* pCxt, SInsertStmt* pInsert) {
pCxt->pCurrStmt = (SNode*)pInsert;
int32_t code = translateFrom(pCxt, pInsert->pTable);
int32_t code = translateInsertTable(pCxt, pInsert->pTable);
if (TSDB_CODE_SUCCESS == code) {
code = translateInsertCols(pCxt, pInsert);
}
@ -7089,9 +7099,10 @@ static int32_t rewriteDropTable(STranslateContext* pCxt, SQuery* pQuery) {
static int32_t buildUpdateTagValReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, STableMeta* pTableMeta,
SVAlterTbReq* pReq) {
SSchema* pSchema = getColSchema(pTableMeta, pStmt->colName);
SSchema* pSchema = getTagSchema(pTableMeta, pStmt->colName);
if (NULL == pSchema) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_ALTER_TABLE);
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_ALTER_TABLE, "Invalid tag name: %s",
pStmt->colName);
}
pReq->tagName = strdup(pStmt->colName);

98
source/libs/planner/src/planLogicCreater.c
View File

@ -36,6 +36,7 @@ static int32_t createQueryLogicNode(SLogicPlanContext* pCxt, SNode* pStmt, SLogi
typedef struct SRewriteExprCxt {
int32_t errCode;
SNodeList* pExprs;
bool* pOutputs;
} SRewriteExprCxt;
static void setColumnInfo(SFunctionNode* pFunc, SColumnNode* pCol) {
@ -63,14 +64,30 @@ static void setColumnInfo(SFunctionNode* pFunc, SColumnNode* pCol) {
}
static EDealRes doRewriteExpr(SNode** pNode, void* pContext) {
SRewriteExprCxt* pCxt = (SRewriteExprCxt*)pContext;
switch (nodeType(*pNode)) {
case QUERY_NODE_COLUMN: {
if (NULL != pCxt->pOutputs) {
SNode* pExpr;
int32_t index = 0;
FOREACH(pExpr, pCxt->pExprs) {
if (QUERY_NODE_GROUPING_SET == nodeType(pExpr)) {
pExpr = nodesListGetNode(((SGroupingSetNode*)pExpr)->pParameterList, 0);
}
if (nodesEqualNode(pExpr, *pNode)) {
pCxt->pOutputs[index] = true;
break;
}
}
}
break;
}
case QUERY_NODE_OPERATOR:
case QUERY_NODE_LOGIC_CONDITION:
case QUERY_NODE_FUNCTION:
case QUERY_NODE_CASE_WHEN: {
SRewriteExprCxt* pCxt = (SRewriteExprCxt*)pContext;
SNode* pExpr;
int32_t index = 0;
SNode* pExpr;
int32_t index = 0;
FOREACH(pExpr, pCxt->pExprs) {
if (QUERY_NODE_GROUPING_SET == nodeType(pExpr)) {
pExpr = nodesListGetNode(((SGroupingSetNode*)pExpr)->pParameterList, 0);
@ -89,6 +106,9 @@ static EDealRes doRewriteExpr(SNode** pNode, void* pContext) {
}
nodesDestroyNode(*pNode);
*pNode = (SNode*)pCol;
if (NULL != pCxt->pOutputs) {
pCxt->pOutputs[index] = true;
}
return DEAL_RES_IGNORE_CHILD;
}
++index;
@ -121,7 +141,7 @@ static EDealRes doNameExpr(SNode* pNode, void* pContext) {
static int32_t rewriteExprForSelect(SNode* pExpr, SSelectStmt* pSelect, ESqlClause clause) {
nodesWalkExpr(pExpr, doNameExpr, NULL);
SRewriteExprCxt cxt = {.errCode = TSDB_CODE_SUCCESS, .pExprs = NULL};
SRewriteExprCxt cxt = {.errCode = TSDB_CODE_SUCCESS, .pExprs = NULL, .pOutputs = NULL};
cxt.errCode = nodesListMakeAppend(&cxt.pExprs, pExpr);
if (TSDB_CODE_SUCCESS == cxt.errCode) {
nodesRewriteSelectStmt(pSelect, clause, doRewriteExpr, &cxt);
@ -130,23 +150,50 @@ static int32_t rewriteExprForSelect(SNode* pExpr, SSelectStmt* pSelect, ESqlClau
return cxt.errCode;
}
static int32_t rewriteExprsForSelect(SNodeList* pExprs, SSelectStmt* pSelect, ESqlClause clause) {
static int32_t cloneRewriteExprs(SNodeList* pExprs, bool* pOutputs, SNodeList** pRewriteExpr) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t index = 0;
SNode* pExpr = NULL;
FOREACH(pExpr, pExprs) {
if (pOutputs[index]) {
code = nodesListMakeStrictAppend(pRewriteExpr, nodesCloneNode(pExpr));
if (TSDB_CODE_SUCCESS != code) {
NODES_DESTORY_LIST(*pRewriteExpr);
break;
}
}
}
return code;
}
static int32_t rewriteExprsForSelect(SNodeList* pExprs, SSelectStmt* pSelect, ESqlClause clause,
SNodeList** pRewriteExprs) {
nodesWalkExprs(pExprs, doNameExpr, NULL);
SRewriteExprCxt cxt = {.errCode = TSDB_CODE_SUCCESS, .pExprs = pExprs};
SRewriteExprCxt cxt = {.errCode = TSDB_CODE_SUCCESS, .pExprs = pExprs, .pOutputs = NULL};
if (NULL != pRewriteExprs) {
cxt.pOutputs = taosMemoryCalloc(LIST_LENGTH(pExprs), sizeof(bool));
if (NULL == cxt.pOutputs) {
return TSDB_CODE_OUT_OF_MEMORY;
}
}
nodesRewriteSelectStmt(pSelect, clause, doRewriteExpr, &cxt);
if (TSDB_CODE_SUCCESS == cxt.errCode && NULL != pRewriteExprs) {
cxt.errCode = cloneRewriteExprs(pExprs, cxt.pOutputs, pRewriteExprs);
}
taosMemoryFree(cxt.pOutputs);
return cxt.errCode;
}
static int32_t rewriteExpr(SNodeList* pExprs, SNode** pTarget) {
nodesWalkExprs(pExprs, doNameExpr, NULL);
SRewriteExprCxt cxt = {.errCode = TSDB_CODE_SUCCESS, .pExprs = pExprs};
SRewriteExprCxt cxt = {.errCode = TSDB_CODE_SUCCESS, .pExprs = pExprs, .pOutputs = NULL};
nodesRewriteExpr(pTarget, doRewriteExpr, &cxt);
return cxt.errCode;
}
static int32_t rewriteExprs(SNodeList* pExprs, SNodeList* pTarget) {
nodesWalkExprs(pExprs, doNameExpr, NULL);
SRewriteExprCxt cxt = {.errCode = TSDB_CODE_SUCCESS, .pExprs = pExprs};
SRewriteExprCxt cxt = {.errCode = TSDB_CODE_SUCCESS, .pExprs = pExprs, .pOutputs = NULL};
nodesRewriteExprs(pTarget, doRewriteExpr, &cxt);
return cxt.errCode;
}
@ -311,7 +358,7 @@ static int32_t createScanLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
// rewrite the expression in subsequent clauses
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExprsForSelect(pScan->pScanPseudoCols, pSelect, SQL_CLAUSE_FROM);
code = rewriteExprsForSelect(pScan->pScanPseudoCols, pSelect, SQL_CLAUSE_FROM, NULL);
}
pScan->scanType = getScanType(pCxt, pScan->pScanPseudoCols, pScan->pScanCols, pScan->tableType, pSelect->tagScan);
@ -509,23 +556,20 @@ static int32_t createAggLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect,
// rewrite the expression in subsequent clauses
if (TSDB_CODE_SUCCESS == code && NULL != pAgg->pAggFuncs) {
code = rewriteExprsForSelect(pAgg->pAggFuncs, pSelect, SQL_CLAUSE_GROUP_BY);
code = rewriteExprsForSelect(pAgg->pAggFuncs, pSelect, SQL_CLAUSE_GROUP_BY, NULL);
}
if (NULL != pSelect->pGroupByList) {
if (NULL != pAgg->pGroupKeys) {
code = nodesListStrictAppendList(pAgg->pGroupKeys, nodesCloneList(pSelect->pGroupByList));
} else {
pAgg->pGroupKeys = nodesCloneList(pSelect->pGroupByList);
if (NULL == pAgg->pGroupKeys) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
pAgg->pGroupKeys = nodesCloneList(pSelect->pGroupByList);
if (NULL == pAgg->pGroupKeys) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
}
// rewrite the expression in subsequent clauses
SNodeList* pOutputGroupKeys = NULL;
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExprsForSelect(pAgg->pGroupKeys, pSelect, SQL_CLAUSE_GROUP_BY);
code = rewriteExprsForSelect(pAgg->pGroupKeys, pSelect, SQL_CLAUSE_GROUP_BY, &pOutputGroupKeys);
}
if (TSDB_CODE_SUCCESS == code && NULL != pSelect->pHaving) {
@ -536,9 +580,11 @@ static int32_t createAggLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect,
}
// set the output
if (TSDB_CODE_SUCCESS == code && NULL != pAgg->pGroupKeys) {
code = createColumnByRewriteExprs(pAgg->pGroupKeys, &pAgg->node.pTargets);
if (TSDB_CODE_SUCCESS == code && NULL != pOutputGroupKeys) {
code = createColumnByRewriteExprs(pOutputGroupKeys, &pAgg->node.pTargets);
}
nodesDestroyList(pOutputGroupKeys);
if (TSDB_CODE_SUCCESS == code && NULL != pAgg->pAggFuncs) {
code = createColumnByRewriteExprs(pAgg->pAggFuncs, &pAgg->node.pTargets);
}
@ -574,7 +620,7 @@ static int32_t createIndefRowsFuncLogicNode(SLogicPlanContext* pCxt, SSelectStmt
// indefinite rows functions and _select_values functions
int32_t code = nodesCollectFuncs(pSelect, SQL_CLAUSE_SELECT, fmIsVectorFunc, &pIdfRowsFunc->pFuncs);
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExprsForSelect(pIdfRowsFunc->pFuncs, pSelect, SQL_CLAUSE_SELECT);
code = rewriteExprsForSelect(pIdfRowsFunc->pFuncs, pSelect, SQL_CLAUSE_SELECT, NULL);
}
// set the output
@ -612,7 +658,7 @@ static int32_t createInterpFuncLogicNode(SLogicPlanContext* pCxt, SSelectStmt* p
// interp functions and _group_key functions
int32_t code = nodesCollectFuncs(pSelect, SQL_CLAUSE_SELECT, isInterpFunc, &pInterpFunc->pFuncs);
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExprsForSelect(pInterpFunc->pFuncs, pSelect, SQL_CLAUSE_SELECT);
code = rewriteExprsForSelect(pInterpFunc->pFuncs, pSelect, SQL_CLAUSE_SELECT, NULL);
}
if (TSDB_CODE_SUCCESS == code && NULL != pSelect->pFill) {
@ -656,7 +702,7 @@ static int32_t createWindowLogicNodeFinalize(SLogicPlanContext* pCxt, SSelectStm
int32_t code = nodesCollectFuncs(pSelect, SQL_CLAUSE_WINDOW, fmIsWindowClauseFunc, &pWindow->pFuncs);
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExprsForSelect(pWindow->pFuncs, pSelect, SQL_CLAUSE_WINDOW);
code = rewriteExprsForSelect(pWindow->pFuncs, pSelect, SQL_CLAUSE_WINDOW, NULL);
}
if (TSDB_CODE_SUCCESS == code) {
@ -854,10 +900,10 @@ static int32_t createFillLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
int32_t code = partFillExprs(pSelect, &pFill->pFillExprs, &pFill->pNotFillExprs);
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExprsForSelect(pFill->pFillExprs, pSelect, SQL_CLAUSE_FILL);
code = rewriteExprsForSelect(pFill->pFillExprs, pSelect, SQL_CLAUSE_FILL, NULL);
}
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExprsForSelect(pFill->pNotFillExprs, pSelect, SQL_CLAUSE_FILL);
code = rewriteExprsForSelect(pFill->pNotFillExprs, pSelect, SQL_CLAUSE_FILL, NULL);
}
if (TSDB_CODE_SUCCESS == code) {
code = createColumnByRewriteExprs(pFill->pFillExprs, &pFill->node.pTargets);
@ -1066,7 +1112,7 @@ static int32_t createDistinctLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSe
// rewrite the expression in subsequent clauses
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExprsForSelect(pAgg->pGroupKeys, pSelect, SQL_CLAUSE_DISTINCT);
code = rewriteExprsForSelect(pAgg->pGroupKeys, pSelect, SQL_CLAUSE_DISTINCT, NULL);
}
// set the output

26
source/libs/planner/src/planOptimizer.c
View File

@ -1476,19 +1476,33 @@ static bool partTagsHasIndefRowsSelectFunc(SNodeList* pFuncs) {
return false;
}
static int32_t partTagsRewriteGroupTagsToFuncs(SNodeList* pGroupTags, int32_t start, SNodeList* pAggFuncs) {
bool hasIndefRowsSelectFunc = partTagsHasIndefRowsSelectFunc(pAggFuncs);
static bool partTagsNeedOutput(SNode* pExpr, SNodeList* pTargets) {
SNode* pOutput = NULL;
FOREACH(pOutput, pTargets) {
if (QUERY_NODE_COLUMN == nodeType(pExpr)) {
if (nodesEqualNode(pExpr, pOutput)) {
return true;
}
} else if (0 == strcmp(((SExprNode*)pExpr)->aliasName, ((SColumnNode*)pOutput)->colName)) {
return true;
}
}
return false;
}
static int32_t partTagsRewriteGroupTagsToFuncs(SNodeList* pGroupTags, int32_t start, SAggLogicNode* pAgg) {
bool hasIndefRowsSelectFunc = partTagsHasIndefRowsSelectFunc(pAgg->pAggFuncs);
int32_t code = TSDB_CODE_SUCCESS;
int32_t index = 0;
SNode* pNode = NULL;
FOREACH(pNode, pGroupTags) {
if (index++ < start) {
if (index++ < start || !partTagsNeedOutput(pNode, pAgg->node.pTargets)) {
continue;
}
if (hasIndefRowsSelectFunc) {
code = nodesListStrictAppend(pAggFuncs, partTagsCreateWrapperFunc("_select_value", pNode));
code = nodesListStrictAppend(pAgg->pAggFuncs, partTagsCreateWrapperFunc("_select_value", pNode));
} else {
code = nodesListStrictAppend(pAggFuncs, partTagsCreateWrapperFunc("_group_key", pNode));
code = nodesListStrictAppend(pAgg->pAggFuncs, partTagsCreateWrapperFunc("_group_key", pNode));
}
if (TSDB_CODE_SUCCESS != code) {
break;
@ -1541,7 +1555,7 @@ static int32_t partTagsOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLogicSub
}
NODES_DESTORY_LIST(pAgg->pGroupKeys);
if (TSDB_CODE_SUCCESS == code && start >= 0) {
code = partTagsRewriteGroupTagsToFuncs(pScan->pGroupTags, start, pAgg->pAggFuncs);
code = partTagsRewriteGroupTagsToFuncs(pScan->pGroupTags, start, pAgg);
}
}
if (TSDB_CODE_SUCCESS == code) {

1
source/libs/planner/src/planPhysiCreater.c
View File

@ -1064,6 +1064,7 @@ static int32_t doCreateExchangePhysiNode(SPhysiPlanContext* pCxt, SExchangeLogic
pExchange->srcStartGroupId = pExchangeLogicNode->srcStartGroupId;
pExchange->srcEndGroupId = pExchangeLogicNode->srcEndGroupId;
pExchange->seqRecvData = pExchangeLogicNode->seqRecvData;
*pPhyNode = (SPhysiNode*)pExchange;
return TSDB_CODE_SUCCESS;

93
source/libs/planner/src/planSpliter.c
View File

@ -292,6 +292,43 @@ static bool stbSplNeedSplitJoin(bool streamQuery, SJoinLogicNode* pJoin) {
return true;
}
static SNodeList* stbSplGetPartKeys(SLogicNode* pNode) {
if (QUERY_NODE_LOGIC_PLAN_SCAN == nodeType(pNode)) {
return ((SScanLogicNode*)pNode)->pGroupTags;
} else if (QUERY_NODE_LOGIC_PLAN_PARTITION == nodeType(pNode)) {
return ((SPartitionLogicNode*)pNode)->pPartitionKeys;
} else {
return NULL;
}
}
static bool stbSplHasPartTbname(SNodeList* pPartKeys) {
if (NULL == pPartKeys) {
return false;
}
SNode* pPartKey = NULL;
FOREACH(pPartKey, pPartKeys) {
if (QUERY_NODE_GROUPING_SET == nodeType(pPartKey)) {
pPartKey = nodesListGetNode(((SGroupingSetNode*)pPartKey)->pParameterList, 0);
}
if ((QUERY_NODE_FUNCTION == nodeType(pPartKey) && FUNCTION_TYPE_TBNAME == ((SFunctionNode*)pPartKey)->funcType) ||
(QUERY_NODE_COLUMN == nodeType(pPartKey) && COLUMN_TYPE_TBNAME == ((SColumnNode*)pPartKey)->colType)) {
return true;
}
}
return false;
}
static bool stbSplIsPartTableAgg(SAggLogicNode* pAgg) {
if (NULL != pAgg->pGroupKeys) {
return stbSplHasPartTbname(pAgg->pGroupKeys);
}
if (1 != LIST_LENGTH(pAgg->node.pChildren)) {
return false;
}
return stbSplHasPartTbname(stbSplGetPartKeys((SLogicNode*)nodesListGetNode(pAgg->node.pChildren, 0)));
}
static bool stbSplNeedSplit(bool streamQuery, SLogicNode* pNode) {
switch (nodeType(pNode)) {
case QUERY_NODE_LOGIC_PLAN_SCAN:
@ -301,7 +338,9 @@ static bool stbSplNeedSplit(bool streamQuery, SLogicNode* pNode) {
case QUERY_NODE_LOGIC_PLAN_PARTITION:
return streamQuery ? false : stbSplIsMultiTbScanChild(streamQuery, pNode);
case QUERY_NODE_LOGIC_PLAN_AGG:
return !stbSplHasGatherExecFunc(((SAggLogicNode*)pNode)->pAggFuncs) && stbSplHasMultiTbScan(streamQuery, pNode);
return (!stbSplHasGatherExecFunc(((SAggLogicNode*)pNode)->pAggFuncs) ||
stbSplIsPartTableAgg((SAggLogicNode*)pNode)) &&
stbSplHasMultiTbScan(streamQuery, pNode);
case QUERY_NODE_LOGIC_PLAN_WINDOW:
return stbSplNeedSplitWindow(streamQuery, pNode);
case QUERY_NODE_LOGIC_PLAN_SORT:
@ -676,27 +715,8 @@ static int32_t stbSplSplitState(SSplitContext* pCxt, SStableSplitInfo* pInfo) {
}
}
static SNodeList* stbSplGetPartKeys(SLogicNode* pNode) {
if (QUERY_NODE_LOGIC_PLAN_SCAN == nodeType(pNode)) {
return ((SScanLogicNode*)pNode)->pGroupTags;
} else if (QUERY_NODE_LOGIC_PLAN_PARTITION == nodeType(pNode)) {
return ((SPartitionLogicNode*)pNode)->pPartitionKeys;
} else {
return NULL;
}
}
static bool stbSplIsPartTbanme(SNodeList* pPartKeys) {
if (NULL == pPartKeys || 1 != LIST_LENGTH(pPartKeys)) {
return false;
}
SNode* pPartKey = nodesListGetNode(pPartKeys, 0);
return (QUERY_NODE_FUNCTION == nodeType(pPartKey) && FUNCTION_TYPE_TBNAME == ((SFunctionNode*)pPartKey)->funcType) ||
(QUERY_NODE_COLUMN == nodeType(pPartKey) && COLUMN_TYPE_TBNAME == ((SColumnNode*)pPartKey)->colType);
}
static bool stbSplIsPartTableWinodw(SWindowLogicNode* pWindow) {
return stbSplIsPartTbanme(stbSplGetPartKeys((SLogicNode*)nodesListGetNode(pWindow->node.pChildren, 0)));
return stbSplHasPartTbname(stbSplGetPartKeys((SLogicNode*)nodesListGetNode(pWindow->node.pChildren, 0)));
}
static int32_t stbSplSplitWindowForCrossTable(SSplitContext* pCxt, SStableSplitInfo* pInfo) {
@ -713,6 +733,17 @@ static int32_t stbSplSplitWindowForCrossTable(SSplitContext* pCxt, SStableSplitI
return TSDB_CODE_PLAN_INTERNAL_ERROR;
}
static bool stbSplNeedSeqRecvData(SLogicNode* pNode) {
if (NULL == pNode) {
return false;
}
if (NULL != pNode->pLimit || NULL != pNode->pSlimit) {
return true;
}
return stbSplNeedSeqRecvData(pNode->pParent);
}
static int32_t stbSplSplitWindowForPartTable(SSplitContext* pCxt, SStableSplitInfo* pInfo) {
if (pCxt->pPlanCxt->streamQuery) {
SPLIT_FLAG_SET_MASK(pInfo->pSubplan->splitFlag, SPLIT_FLAG_STABLE_SPLIT);
@ -728,6 +759,7 @@ static int32_t stbSplSplitWindowForPartTable(SSplitContext* pCxt, SStableSplitIn
code = replaceLogicNode(pInfo->pSubplan, pInfo->pSplitNode, (SLogicNode*)pExchange);
}
if (TSDB_CODE_SUCCESS == code) {
pExchange->seqRecvData = stbSplNeedSeqRecvData((SLogicNode*)pExchange);
code = nodesListMakeStrictAppend(&pInfo->pSubplan->pChildren,
(SNode*)splCreateScanSubplan(pCxt, pInfo->pSplitNode, SPLIT_FLAG_STABLE_SPLIT));
}
@ -797,7 +829,17 @@ static int32_t stbSplCreatePartAggNode(SAggLogicNode* pMergeAgg, SLogicNode** pO
return code;
}
static int32_t stbSplSplitAggNode(SSplitContext* pCxt, SStableSplitInfo* pInfo) {
static int32_t stbSplSplitAggNodeForPartTable(SSplitContext* pCxt, SStableSplitInfo* pInfo) {
int32_t code = splCreateExchangeNodeForSubplan(pCxt, pInfo->pSubplan, pInfo->pSplitNode, SUBPLAN_TYPE_MERGE);
if (TSDB_CODE_SUCCESS == code) {
code = nodesListMakeStrictAppend(&pInfo->pSubplan->pChildren,
(SNode*)splCreateScanSubplan(pCxt, pInfo->pSplitNode, SPLIT_FLAG_STABLE_SPLIT));
}
++(pCxt->groupId);
return code;
}
static int32_t stbSplSplitAggNodeForCrossTable(SSplitContext* pCxt, SStableSplitInfo* pInfo) {
SLogicNode* pPartAgg = NULL;
int32_t code = stbSplCreatePartAggNode((SAggLogicNode*)pInfo->pSplitNode, &pPartAgg);
if (TSDB_CODE_SUCCESS == code) {
@ -812,6 +854,13 @@ static int32_t stbSplSplitAggNode(SSplitContext* pCxt, SStableSplitInfo* pInfo)
return code;
}
static int32_t stbSplSplitAggNode(SSplitContext* pCxt, SStableSplitInfo* pInfo) {
if (stbSplIsPartTableAgg((SAggLogicNode*)pInfo->pSplitNode)) {
return stbSplSplitAggNodeForPartTable(pCxt, pInfo);
}
return stbSplSplitAggNodeForCrossTable(pCxt, pInfo);
}
static SNode* stbSplCreateColumnNode(SExprNode* pExpr) {
SColumnNode* pCol = (SColumnNode*)nodesMakeNode(QUERY_NODE_COLUMN);
if (NULL == pCol) {

2
source/libs/sync/inc/syncInt.h
View File

@ -232,7 +232,7 @@ int32_t syncNodeSendMsgById(const SRaftId* destRaftId, SSyncNode* pSyncNode, S
int32_t syncNodeSendMsgByInfo(const SNodeInfo* nodeInfo, SSyncNode* pSyncNode, SRpcMsg* pMsg);
SyncIndex syncMinMatchIndex(SSyncNode* pSyncNode);
int32_t syncCacheEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry, LRUHandle** h);
bool syncNodeHeartbeatTimeout(SSyncNode* pSyncNode);
bool syncNodeHeartbeatReplyTimeout(SSyncNode* pSyncNode);
// raft state change --------------
void syncNodeUpdateTerm(SSyncNode* pSyncNode, SyncTerm term);

1
source/libs/sync/inc/syncMessage.h
View File

@ -124,6 +124,7 @@ typedef struct SyncHeartbeat {
SyncIndex commitIndex;
SyncTerm privateTerm;
SyncTerm minMatchIndex;
int64_t timeStamp;
} SyncHeartbeat;
typedef struct SyncHeartbeatReply {

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

@ -640,7 +640,7 @@ int32_t syncNodePropose(SSyncNode* pSyncNode, SRpcMsg* pMsg, bool isWeak) {
}
// heartbeat timeout
if (syncNodeHeartbeatTimeout(pSyncNode)) {
if (syncNodeHeartbeatReplyTimeout(pSyncNode)) {
terrno = TSDB_CODE_SYN_PROPOSE_NOT_READY;
sNError(pSyncNode, "failed to sync propose since hearbeat timeout, type:%s, last:%" PRId64 ", cmt:%" PRId64,
TMSG_INFO(pMsg->msgType), syncNodeGetLastIndex(pSyncNode), pSyncNode->commitIndex);
@ -2039,6 +2039,7 @@ static void syncNodeEqPeerHeartbeatTimer(void* param, void* tmrId) {
pSyncMsg->commitIndex = pSyncNode->commitIndex;
pSyncMsg->minMatchIndex = syncMinMatchIndex(pSyncNode);
pSyncMsg->privateTerm = 0;
pSyncMsg->timeStamp = taosGetTimestampMs();
// send msg
syncNodeSendHeartbeat(pSyncNode, &pSyncMsg->destId, &rpcMsg);
@ -2094,7 +2095,7 @@ int32_t syncCacheEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry, LRUHand
return code;
}
bool syncNodeHeartbeatTimeout(SSyncNode* pSyncNode) {
bool syncNodeHeartbeatReplyTimeout(SSyncNode* pSyncNode) {
if (pSyncNode->replicaNum == 1) {
return false;
}
@ -2148,7 +2149,11 @@ static int32_t syncNodeAppendNoop(SSyncNode* ths) {
int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
SyncHeartbeat* pMsg = pRpcMsg->pCont;
syncLogRecvHeartbeat(ths, pMsg, "");
int64_t tsMs = taosGetTimestampMs();
char buf[128];
snprintf(buf, sizeof(buf), "recv local time:%" PRId64, tsMs);
syncLogRecvHeartbeat(ths, pMsg, buf);
SRpcMsg rpcMsg = {0};
(void)syncBuildHeartbeatReply(&rpcMsg, ths->vgId);
@ -2161,6 +2166,8 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
pMsgReply->timeStamp = taosGetTimestampMs();
if (pMsg->term == ths->pRaftStore->currentTerm && ths->state != TAOS_SYNC_STATE_LEADER) {
syncIndexMgrSetRecvTime(ths->pNextIndex, &(pMsg->srcId), tsMs);
syncNodeResetElectTimer(ths);
ths->minMatchIndex = pMsg->minMatchIndex;
@ -2220,10 +2227,14 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
int32_t syncNodeOnHeartbeatReply(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
SyncHeartbeatReply* pMsg = pRpcMsg->pCont;
syncLogRecvHeartbeatReply(ths, pMsg, "");
int64_t tsMs = taosGetTimestampMs();
char buf[128];
snprintf(buf, sizeof(buf), "recv local time:%" PRId64, tsMs);
syncLogRecvHeartbeatReply(ths, pMsg, buf);
// update last reply time, make decision whether the other node is alive or not
syncIndexMgrSetRecvTime(ths->pMatchIndex, &pMsg->srcId, pMsg->timeStamp);
syncIndexMgrSetRecvTime(ths->pMatchIndex, &pMsg->srcId, tsMs);
return 0;
}
@ -2498,6 +2509,8 @@ int32_t syncNodeDoCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endInde
SRpcMsg rpcMsg = {0};
syncEntry2OriginalRpc(pEntry, &rpcMsg);
sTrace("do commit index:%" PRId64 ", type:%s", i, TMSG_INFO(pEntry->msgType));
// user commit
if ((ths->pFsm->FpCommitCb != NULL) && syncUtilUserCommit(pEntry->originalRpcType)) {
bool internalExecute = true;
@ -2505,7 +2518,8 @@ int32_t syncNodeDoCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endInde
internalExecute = false;
}
sNTrace(ths, "commit index:%" PRId64 ", internal:%d", i, internalExecute);
sNTrace(ths, "user commit index:%" PRId64 ", internal:%d, type:%s", i, internalExecute,
TMSG_INFO(pEntry->msgType));
// execute fsm in apply thread, or execute outside syncPropose
if (internalExecute) {

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

@ -213,9 +213,11 @@ int32_t syncNodeSendHeartbeat(SSyncNode* pSyncNode, const SRaftId* destId, SRpcM
}
int32_t syncNodeHeartbeatPeers(SSyncNode* pSyncNode) {
int64_t ts = taosGetTimestampMs();
for (int32_t i = 0; i < pSyncNode->peersNum; ++i) {
SRpcMsg rpcMsg = {0};
if (syncBuildHeartbeat(&rpcMsg, pSyncNode->vgId) != 0) {
sError("vgId:%d, build sync-heartbeat error", pSyncNode->vgId);
continue;
}
@ -226,6 +228,7 @@ int32_t syncNodeHeartbeatPeers(SSyncNode* pSyncNode) {
pSyncMsg->commitIndex = pSyncNode->commitIndex;
pSyncMsg->minMatchIndex = syncMinMatchIndex(pSyncNode);
pSyncMsg->privateTerm = 0;
pSyncMsg->timeStamp = ts;
// send msg
syncNodeSendHeartbeat(pSyncNode, &pSyncMsg->destId, &rpcMsg);

55
source/libs/sync/src/syncUtil.c
View File

@ -15,6 +15,7 @@
#define _DEFAULT_SOURCE
#include "syncUtil.h"
#include "syncIndexMgr.h"
#include "syncMessage.h"
#include "syncRaftCfg.h"
#include "syncRaftStore.h"
@ -175,6 +176,36 @@ void syncCfg2SimpleStr(const SSyncCfg* pCfg, char* buf, int32_t bufLen) {
}
}
// for leader
static void syncHearbeatReplyTime2Str(SSyncNode* pSyncNode, char* buf, int32_t bufLen) {
int32_t len = 5;
for (int32_t i = 0; i < pSyncNode->replicaNum; ++i) {
int64_t tsMs = syncIndexMgrGetRecvTime(pSyncNode->pMatchIndex, &(pSyncNode->replicasId[i]));
if (i < pSyncNode->replicaNum - 1) {
len += snprintf(buf + len, bufLen - len, "%d:%" PRId64 ",", i, tsMs);
} else {
len += snprintf(buf + len, bufLen - len, "%d:%" PRId64 "}", i, tsMs);
}
}
}
// for follower
static void syncHearbeatTime2Str(SSyncNode* pSyncNode, char* buf, int32_t bufLen) {
int32_t len = 4;
for (int32_t i = 0; i < pSyncNode->replicaNum; ++i) {
int64_t tsMs = syncIndexMgrGetRecvTime(pSyncNode->pNextIndex, &(pSyncNode->replicasId[i]));
if (i < pSyncNode->replicaNum - 1) {
len += snprintf(buf + len, bufLen - len, "%d:%" PRId64 ",", i, tsMs);
} else {
len += snprintf(buf + len, bufLen - len, "%d:%" PRId64 "}", i, tsMs);
}
}
}
static void syncPeerState2Str(SSyncNode* pSyncNode, char* buf, int32_t bufLen) {
int32_t len = 1;
@ -221,6 +252,12 @@ void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNo
char peerStr[1024] = "{";
syncPeerState2Str(pNode, peerStr, sizeof(peerStr));
char hbrTimeStr[256] = "hbr:{";
syncHearbeatReplyTime2Str(pNode, hbrTimeStr, sizeof(hbrTimeStr));
char hbTimeStr[256] = "hb:{";
syncHearbeatTime2Str(pNode, hbTimeStr, sizeof(hbTimeStr));
int32_t quorum = syncNodeDynamicQuorum(pNode);
char eventLog[512]; // {0};
@ -243,12 +280,13 @@ void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNo
"%s"
", tm:%" PRIu64 ", cmt:%" PRId64 ", fst:%" PRId64 ", lst:%" PRId64 ", min:%" PRId64 ", snap:%" PRId64
", snap-tm:%" PRIu64 ", sby:%d, aq:%d, snaping:%" PRId64 ", r-num:%d, lcfg:%" PRId64
", chging:%d, rsto:%d, dquorum:%d, elt:%" PRId64 ", hb:%" PRId64 ", %s, %s",
", chging:%d, rsto:%d, dquorum:%d, elt:%" PRId64 ", hb:%" PRId64 ", %s, %s, %s, %s",
pNode->vgId, syncStr(pNode->state), eventLog, currentTerm, pNode->commitIndex, logBeginIndex,
logLastIndex, pNode->minMatchIndex, snapshot.lastApplyIndex, snapshot.lastApplyTerm,
pNode->pRaftCfg->isStandBy, aqItems, pNode->snapshottingIndex, pNode->replicaNum,
pNode->pRaftCfg->lastConfigIndex, pNode->changing, pNode->restoreFinish, quorum,
pNode->electTimerLogicClock, pNode->heartbeatTimerLogicClockUser, peerStr, cfgStr);
pNode->electTimerLogicClock, pNode->heartbeatTimerLogicClockUser, peerStr, cfgStr, hbTimeStr,
hbrTimeStr);
}
}
@ -395,9 +433,8 @@ void syncLogSendHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, const
syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port);
sNTrace(pSyncNode,
"send sync-heartbeat to %s:%d {term:%" PRId64 ", cmt:%" PRId64 ", min-match:%" PRId64 ", pterm:%" PRId64
"}, %s",
host, port, pMsg->term, pMsg->commitIndex, pMsg->minMatchIndex, pMsg->privateTerm, s);
"send sync-heartbeat to %s:%d {term:%" PRId64 ", cmt:%" PRId64 ", min-match:%" PRId64 ", ts:%" PRId64 "}, %s",
host, port, pMsg->term, pMsg->commitIndex, pMsg->minMatchIndex, pMsg->timeStamp, s);
}
void syncLogRecvHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, const char* s) {
@ -406,9 +443,9 @@ void syncLogRecvHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, const
syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
sNTrace(pSyncNode,
"recv sync-heartbeat from %s:%d {term:%" PRId64 ", cmt:%" PRId64 ", min-match:%" PRId64 ", pterm:%" PRId64
"recv sync-heartbeat from %s:%d {term:%" PRId64 ", cmt:%" PRId64 ", min-match:%" PRId64 ", ts:%" PRId64
"}, %s",
host, port, pMsg->term, pMsg->commitIndex, pMsg->minMatchIndex, pMsg->privateTerm, s);
host, port, pMsg->term, pMsg->commitIndex, pMsg->minMatchIndex, pMsg->timeStamp, s);
}
void syncLogSendHeartbeatReply(SSyncNode* pSyncNode, const SyncHeartbeatReply* pMsg, const char* s) {
@ -416,8 +453,8 @@ void syncLogSendHeartbeatReply(SSyncNode* pSyncNode, const SyncHeartbeatReply* p
uint16_t port;
syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port);
sNTrace(pSyncNode, "send sync-heartbeat-reply from %s:%d {term:%" PRId64 ", pterm:%" PRId64 "}, %s", host, port,
pMsg->term, pMsg->privateTerm, s);
sNTrace(pSyncNode, "send sync-heartbeat-reply from %s:%d {term:%" PRId64 ", ts:%" PRId64 "}, %s", host, port,
pMsg->term, pMsg->timeStamp, s);
}
void syncLogRecvHeartbeatReply(SSyncNode* pSyncNode, const SyncHeartbeatReply* pMsg, const char* s) {

6
source/libs/tdb/src/db/tdbBtree.c
View File

@ -741,14 +741,12 @@ static int tdbBtreeBalanceNonRoot(SBTree *pBt, SPage *pParent, int idx, TXN *pTx
tdbPageCreate(pOlds[0]->pageSize, &pOldsCopy[i], tdbDefaultMalloc, NULL);
tdbBtreeInitPage(pOldsCopy[i], &iarg, 0);
tdbPageCopy(pOlds[i], pOldsCopy[i], 0);
}
for (iNew = 0; iNew < nNews; ++iNew) {
tdbBtreeInitPage(pNews[iNew], &iarg, 0);
pOlds[i]->nOverflow = 0;
}
iNew = 0;
nNewCells = 0;
tdbBtreeInitPage(pNews[iNew], &iarg, 0);
for (int iOld = 0; iOld < nOlds; iOld++) {
SPage *pPage;

32
source/libs/tdb/src/db/tdbPager.c
View File

@ -28,12 +28,12 @@ typedef struct {
TDB_STATIC_ASSERT(sizeof(SFileHdr) == 128, "Size of file header is not correct");
struct hashset_st {
size_t nbits;
size_t mask;
size_t capacity;
size_t nbits;
size_t mask;
size_t capacity;
size_t *items;
size_t nitems;
double load_factor;
size_t nitems;
double load_factor;
};
static const unsigned int prime = 39;
@ -68,11 +68,11 @@ void hashset_destroy(hashset_t set) {
}
int hashset_add_member(hashset_t set, void *item) {
size_t value = (size_t) item;
size_t value = (size_t)item;
size_t h;
if (value == 0) {
return -1;
return -1;
}
for (h = set->mask & (prime * value); set->items[h] != 0; h = set->mask & (h + prime2)) {
@ -103,7 +103,7 @@ int hashset_add(hashset_t set, void *item) {
set->nitems = 0;
for (size_t i = 0; i < old_capacity; ++i) {
hashset_add_member(set, (void*)old_items[i]);
hashset_add_member(set, (void *)old_items[i]);
}
tdbOsFree(old_items);
}
@ -112,7 +112,7 @@ int hashset_add(hashset_t set, void *item) {
}
int hashset_remove(hashset_t set, void *item) {
size_t value = (size_t) item;
size_t value = (size_t)item;
for (size_t h = set->mask & (prime * value); set->items[h] != 0; h = set->mask & (h + prime2)) {
if (set->items[h] == value) {
@ -126,7 +126,7 @@ int hashset_remove(hashset_t set, void *item) {
}
int hashset_contains(hashset_t set, void *item) {
size_t value = (size_t) item;
size_t value = (size_t)item;
for (size_t h = set->mask & (prime * value); set->items[h] != 0; h = set->mask & (h + prime2)) {
if (set->items[h] == value) {
@ -319,7 +319,8 @@ int tdbPagerWrite(SPager *pPager, SPage *pPage) {
tRBTreePut(&pPager->rbt, (SRBTreeNode *)pPage);
// Write page to journal if neccessary
if (TDB_PAGE_PGNO(pPage) <= pPager->dbOrigSize && (pPager->jPageSet == NULL || !hashset_contains(pPager->jPageSet, (void*)((long)TDB_PAGE_PGNO(pPage))))) {
if (TDB_PAGE_PGNO(pPage) <= pPager->dbOrigSize &&
(pPager->jPageSet == NULL || !hashset_contains(pPager->jPageSet, (void *)((long)TDB_PAGE_PGNO(pPage))))) {
ret = tdbPagerWritePageToJournal(pPager, pPage);
if (ret < 0) {
tdbError("failed to write page to journal since %s", tstrerror(terrno));
@ -327,7 +328,7 @@ int tdbPagerWrite(SPager *pPager, SPage *pPage) {
}
if (pPager->jPageSet) {
hashset_add(pPager->jPageSet, (void*)((long)TDB_PAGE_PGNO(pPage)));
hashset_add(pPager->jPageSet, (void *)((long)TDB_PAGE_PGNO(pPage)));
}
}
@ -372,6 +373,7 @@ int tdbPagerCommit(SPager *pPager, TXN *pTxn) {
SRBTreeNode *pNode = NULL;
while ((pNode = tRBTreeIterNext(&iter)) != NULL) {
pPage = (SPage *)pNode;
ASSERT(pPage->nOverflow == 0);
ret = tdbPagerWritePageToDB(pPager, pPage);
if (ret < 0) {
tdbError("failed to write page to db since %s", tstrerror(terrno));
@ -391,7 +393,7 @@ int tdbPagerCommit(SPager *pPager, TXN *pTxn) {
tRBTreeDrop(&pPager->rbt, (SRBTreeNode *)pPage);
if (pPager->jPageSet) {
hashset_remove(pPager->jPageSet, (void*)((long)TDB_PAGE_PGNO(pPage)));
hashset_remove(pPager->jPageSet, (void *)((long)TDB_PAGE_PGNO(pPage)));
}
tdbPCacheRelease(pPager->pCache, pPage, pTxn);
}
@ -503,7 +505,7 @@ int tdbPagerAbort(SPager *pPager, TXN *pTxn) {
return -1;
}
u8 *pageBuf = tdbOsCalloc(1, pPager->pageSize);
u8 *pageBuf = tdbOsCalloc(1, pPager->pageSize);
if (pageBuf == NULL) {
return -1;
}
@ -560,7 +562,7 @@ int tdbPagerAbort(SPager *pPager, TXN *pTxn) {
pPage->isDirty = 0;
tRBTreeDrop(&pPager->rbt, (SRBTreeNode *)pPage);
hashset_remove(pPager->jPageSet, (void*)((long)TDB_PAGE_PGNO(pPage)));
hashset_remove(pPager->jPageSet, (void *)((long)TDB_PAGE_PGNO(pPage)));
tdbPCacheRelease(pPager->pCache, pPage, pTxn);
}

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

@ -794,15 +794,18 @@ void cliSend(SCliConn* pConn) {
int msgLen = transMsgLenFromCont(pMsg->contLen);
STransMsgHead* pHead = transHeadFromCont(pMsg->pCont);
pHead->ahandle = pCtx != NULL ? (uint64_t)pCtx->ahandle : 0;
pHead->noResp = REQUEST_NO_RESP(pMsg) ? 1 : 0;
pHead->persist = REQUEST_PERSIS_HANDLE(pMsg) ? 1 : 0;
pHead->msgType = pMsg->msgType;
pHead->msgLen = (int32_t)htonl((uint32_t)msgLen);
pHead->release = REQUEST_RELEASE_HANDLE(pCliMsg) ? 1 : 0;
memcpy(pHead->user, pTransInst->user, strlen(pTransInst->user));
pHead->traceId = pMsg->info.traceId;
pHead->magicNum = htonl(TRANS_MAGIC_NUM);
if (pHead->comp == 0) {
pHead->ahandle = pCtx != NULL ? (uint64_t)pCtx->ahandle : 0;
pHead->noResp = REQUEST_NO_RESP(pMsg) ? 1 : 0;
pHead->persist = REQUEST_PERSIS_HANDLE(pMsg) ? 1 : 0;
pHead->msgType = pMsg->msgType;
pHead->msgLen = (int32_t)htonl((uint32_t)msgLen);
pHead->release = REQUEST_RELEASE_HANDLE(pCliMsg) ? 1 : 0;
memcpy(pHead->user, pTransInst->user, strlen(pTransInst->user));
pHead->traceId = pMsg->info.traceId;
pHead->magicNum = htonl(TRANS_MAGIC_NUM);
}
if (pHead->persist == 1) {
CONN_SET_PERSIST_BY_APP(pConn);
}

9
source/libs/transport/src/transComm.c
View File

@ -60,21 +60,20 @@ int32_t transDecompressMsg(char** msg, int32_t len) {
STransMsgHead* pHead = (STransMsgHead*)(*msg);
if (pHead->comp == 0) return 0;
char* pCont = transContFromHead(pHead);
char* pCont = transContFromHead(pHead);
STransCompMsg* pComp = (STransCompMsg*)pCont;
int32_t oriLen = htonl(pComp->contLen);
char* buf = taosMemoryCalloc(1, oriLen + sizeof(STransMsgHead));
STransMsgHead* pNewHead = (STransMsgHead*)buf;
int32_t decompLen = LZ4_decompress_safe(pCont + sizeof(STransCompMsg), pNewHead->content,
len - sizeof(STransMsgHead) - sizeof(STransCompMsg), oriLen);
int32_t decompLen = LZ4_decompress_safe(pCont + sizeof(STransCompMsg), pNewHead->content,
len - sizeof(STransMsgHead) - sizeof(STransCompMsg), oriLen);
memcpy((char*)pNewHead, (char*)pHead, sizeof(STransMsgHead));
pNewHead->msgLen = htonl(oriLen + sizeof(STransMsgHead));
taosMemoryFree(pHead);
*msg = buf;
if (decompLen != oriLen) {
return -1;

32
tests/parallel_test/cases.task
View File

@ -278,7 +278,7 @@
,,y,script,./test.sh -f tsim/stable/values.sim
,,y,script,./test.sh -f tsim/stable/vnode3.sim
,,y,script,./test.sh -f tsim/stable/metrics_idx.sim
,,,script,./test.sh -f tsim/sma/drop_sma.sim
,,n,script,./test.sh -f tsim/sma/drop_sma.sim
,,y,script,./test.sh -f tsim/sma/sma_leak.sim
,,y,script,./test.sh -f tsim/sma/tsmaCreateInsertQuery.sim
,,y,script,./test.sh -f tsim/sma/rsmaCreateInsertQuery.sim
@ -417,18 +417,18 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/user_control.py
,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/fsync.py
,,n,system-test,python3 ./test.py -f 0-others/compatibility.py
#,,,system-test,python3 ./test.py -f 1-insert/alter_database.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/influxdb_line_taosc_insert.py
,,,system-test,python3 ./test.py -f 1-insert/alter_database.py
,,,system-test,python3 ./test.py -f 1-insert/influxdb_line_taosc_insert.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/opentsdb_telnet_line_taosc_insert.py
,,,system-test,python3 ./test.py -f 1-insert/opentsdb_json_taosc_insert.py
,,,system-test,python3 ./test.py -f 1-insert/test_stmt_muti_insert_query.py
,,,system-test,python3 ./test.py -f 1-insert/test_stmt_set_tbname_tag.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/alter_stable.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/alter_table.py
#,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/alter_table.py
,,n,system-test,python3 ./test.py -f 1-insert/boundary.py
,,n,system-test,python3 ./test.py -f 1-insert/insertWithMoreVgroup.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/table_comment.py
,,,system-test,python3 ./test.py -f 1-insert/time_range_wise.py
,,n,system-test,python3 ./test.py -f 1-insert/time_range_wise.py
,,,system-test,python3 ./test.py -f 1-insert/block_wise.py
,,,system-test,python3 ./test.py -f 1-insert/create_retentions.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/mutil_stage.py
@ -436,7 +436,7 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/table_param_ttl.py -R
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/update_data_muti_rows.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/db_tb_name_check.py
,,,system-test,python3 ./test.py -f 1-insert/database_pre_suf.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/database_pre_suf.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/InsertFuturets.py
,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/show.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/abs.py
@ -561,8 +561,8 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/sample.py -R
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/sin.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/sin.py -R
,,,system-test,python3 ./test.py -f 2-query/smaTest.py
,,,system-test,python3 ./test.py -f 2-query/smaTest.py -R
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/smaTest.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/smaTest.py -R
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/sml.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/sml.py -R
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/spread.py
@ -595,8 +595,8 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/Today.py -R
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/top.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/top.py -R
,,,system-test,python3 ./test.py -f 2-query/tsbsQuery.py
,,,system-test,python3 ./test.py -f 2-query/tsbsQuery.py -R
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/tsbsQuery.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/tsbsQuery.py -R
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/ttl_comment.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/ttl_comment.py -R
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/twa.py
@ -617,6 +617,8 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/delete_childtable.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/delete_normaltable.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/keep_expired.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/drop.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/drop.py -N 3 -M 3 -i False -n 3
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/join2.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/union1.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/concat2.py
@ -630,7 +632,7 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/elapsed.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/csum.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/function_diff.py
,,,system-test,python3 ./test.py -f 2-query/queryQnode.py
,,n,system-test,python3 ./test.py -f 2-query/queryQnode.py
,,,system-test,python3 ./test.py -f 6-cluster/5dnode1mnode.py
,,,system-test,python3 ./test.py -f 6-cluster/5dnode2mnode.py -N 5
,,,system-test,python3 ./test.py -f 6-cluster/5dnode3mnodeStop.py -N 5 -M 3
@ -667,7 +669,7 @@
,,,system-test,python3 test.py -f 6-cluster/vnode/4dnode1mnode_basic_replica3_insertdatas_querys.py -N 4 -M 1
,,,system-test,python3 test.py -f 6-cluster/vnode/4dnode1mnode_basic_replica3_vgroups.py -N 4 -M 1
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/create_wrong_topic.py
,,,system-test,python3 ./test.py -f 7-tmq/dropDbR3ConflictTransaction.py -N 3
,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/dropDbR3ConflictTransaction.py -N 3
,,,system-test,python3 ./test.py -f 7-tmq/basic5.py
,,,system-test,python3 ./test.py -f 7-tmq/subscribeDb.py
,,,system-test,python3 ./test.py -f 7-tmq/subscribeDb0.py
@ -723,7 +725,7 @@
,,,system-test,python3 ./test.py -f 7-tmq/dataFromTsdbNWal-multiCtb.py
,,,system-test,python3 ./test.py -f 7-tmq/tmq_taosx.py
,,,system-test,python3 ./test.py -f 7-tmq/stbTagFilter-multiCtb.py
,,,system-test,python3 ./test.py -f 99-TDcase/TD-19201.py
,,y,system-test,./pytest.sh python3 ./test.py -f 99-TDcase/TD-19201.py
,,,system-test,python3 ./test.py -f 7-tmq/tmqSubscribeStb-r3.py -N 5
,,,system-test,python3 ./test.py -f 7-tmq/tmq3mnodeSwitch.py -N 6 -M 3
,,,system-test,python3 ./test.py -f 7-tmq/tmq3mnodeSwitch.py -N 6 -M 3 -n 3
@ -818,7 +820,7 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/count_partition.py -Q 2
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/max_partition.py -Q 2
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/last_row.py -Q 2
,,,system-test,python3 ./test.py -f 2-query/tsbsQuery.py -Q 2
,,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/between.py -Q 3
@ -911,7 +913,7 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/count_partition.py -Q 3
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/max_partition.py -Q 3
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/last_row.py -Q 3
,,,system-test,python3 ./test.py -f 2-query/tsbsQuery.py -Q 3
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/tsbsQuery.py -Q 3
,,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

14
tests/pytest/util/dnodes.py
View File

@ -312,8 +312,14 @@ class TDDnode:
cmd = "mintty -h never %s -c %s" % (
binPath, self.cfgDir)
else:
cmd = "nohup %s -c %s > /dev/null 2>&1 & " % (
binPath, self.cfgDir)
if self.asan:
asanDir = "%s/sim/asan/dnode%d.asan" % (
self.path, self.index)
cmd = "nohup %s -c %s > /dev/null 2> %s & " % (
binPath, self.cfgDir, asanDir)
else:
cmd = "nohup %s -c %s > /dev/null 2>&1 & " % (
binPath, self.cfgDir)
else:
valgrindCmdline = "valgrind --log-file=\"%s/../log/valgrind.log\" --tool=memcheck --leak-check=full --show-reachable=no --track-origins=yes --show-leak-kinds=all -v --workaround-gcc296-bugs=yes"%self.cfgDir
@ -748,7 +754,7 @@ class TDDnodes:
tdLog.exit("index:%d should on a scale of [1, 10]" % (index))
def StopAllSigint(self):
tdLog.info("stop all dnodes sigint")
tdLog.info("stop all dnodes sigint, asan:%d" % self.asan)
if self.asan:
tdLog.info("execute script: %s" % self.stopDnodesSigintPath)
os.system(self.stopDnodesSigintPath)
@ -756,7 +762,7 @@ class TDDnodes:
return
def stopAll(self):
tdLog.info("stop all dnodes")
tdLog.info("stop all dnodes, asan:%d" % self.asan)
if self.asan:
tdLog.info("execute script: %s" % self.stopDnodesPath)
os.system(self.stopDnodesPath)

355
tests/script/api/batchprepare.c
View File

@ -115,6 +115,7 @@ int insertMBMETest3(TAOS_STMT *stmt, TAOS *taos);
int insertMBMETest4(TAOS_STMT *stmt, TAOS *taos);
int insertMPMETest1(TAOS_STMT *stmt, TAOS *taos);
int insertAUTOTest1(TAOS_STMT *stmt, TAOS *taos);
int insertAUTOTest2(TAOS_STMT *stmt, TAOS *taos);
int queryColumnTest(TAOS_STMT *stmt, TAOS *taos);
int queryMiscTest(TAOS_STMT *stmt, TAOS *taos);
@ -128,7 +129,7 @@ typedef struct {
int32_t colNum;
int32_t *colList; // full table column list
int32_t testType;
bool autoCreateTbl;
int32_t autoCreateTbl;
bool fullCol;
int32_t (*runFn)(TAOS_STMT*, TAOS*);
int32_t tblNum;
@ -142,45 +143,46 @@ typedef struct {
} CaseCfg;
CaseCfg gCase[] = {
{"insert:MBSE0-FULL", tListLen(shortColList), shortColList, TTYPE_INSERT, false, true, insertMBSETest1, 1, 10, 10, 0, 0, 0, 1, -1},
{"insert:MBSE0-FULL", tListLen(shortColList), shortColList, TTYPE_INSERT, false, true, insertMBSETest1, 10, 100, 10, 0, 0, 0, 1, -1},
{"insert:MBSE0-FULL", tListLen(shortColList), shortColList, TTYPE_INSERT, 0, true, insertMBSETest1, 1, 10, 10, 0, 0, 0, 1, -1},
{"insert:MBSE0-FULL", tListLen(shortColList), shortColList, TTYPE_INSERT, 0, true, insertMBSETest1, 10, 100, 10, 0, 0, 0, 1, -1},
{"insert:MBSE1-FULL", tListLen(fullColList), fullColList, TTYPE_INSERT, false, true, insertMBSETest1, 10, 10, 2, 0, 0, 0, 1, -1},
{"insert:MBSE1-C012", tListLen(fullColList), fullColList, TTYPE_INSERT, false, false, insertMBSETest1, 10, 10, 2, 12, 0, 0, 1, -1},
{"insert:MBSE1-C002", tListLen(fullColList), fullColList, TTYPE_INSERT, false, false, insertMBSETest1, 10, 10, 2, 2, 0, 0, 1, -1},
{"insert:MBSE1-FULL", tListLen(fullColList), fullColList, TTYPE_INSERT, 0, true, insertMBSETest1, 10, 10, 2, 0, 0, 0, 1, -1},
{"insert:MBSE1-C012", tListLen(fullColList), fullColList, TTYPE_INSERT, 0, false, insertMBSETest1, 10, 10, 2, 12, 0, 0, 1, -1},
{"insert:MBSE1-C002", tListLen(fullColList), fullColList, TTYPE_INSERT, 0, false, insertMBSETest1, 10, 10, 2, 2, 0, 0, 1, -1},
{"insert:MBSE2-FULL", tListLen(fullColList), fullColList, TTYPE_INSERT, false, true, insertMBSETest2, 10, 10, 2, 0, 0, 0, 1, -1},
{"insert:MBSE2-C012", tListLen(fullColList), fullColList, TTYPE_INSERT, false, false, insertMBSETest2, 10, 10, 2, 12, 0, 0, 1, -1},
{"insert:MBSE2-C002", tListLen(fullColList), fullColList, TTYPE_INSERT, false, false, insertMBSETest2, 10, 10, 2, 2, 0, 0, 1, -1},
{"insert:MBSE2-FULL", tListLen(fullColList), fullColList, TTYPE_INSERT, 0, true, insertMBSETest2, 10, 10, 2, 0, 0, 0, 1, -1},
{"insert:MBSE2-C012", tListLen(fullColList), fullColList, TTYPE_INSERT, 0, false, insertMBSETest2, 10, 10, 2, 12, 0, 0, 1, -1},
{"insert:MBSE2-C002", tListLen(fullColList), fullColList, TTYPE_INSERT, 0, false, insertMBSETest2, 10, 10, 2, 2, 0, 0, 1, -1},
{"insert:MBME1-FULL", tListLen(fullColList), fullColList, TTYPE_INSERT, false, true, insertMBMETest1, 10, 10, 2, 0, 0, 0, 1, -1},
{"insert:MBME1-C012", tListLen(fullColList), fullColList, TTYPE_INSERT, false, false, insertMBMETest1, 10, 10, 2, 12, 0, 0, 1, -1},
{"insert:MBME1-C002", tListLen(fullColList), fullColList, TTYPE_INSERT, false, false, insertMBMETest1, 10, 10, 2, 2, 0, 0, 1, -1},
{"insert:MBME1-FULL", tListLen(fullColList), fullColList, TTYPE_INSERT, 0, true, insertMBMETest1, 10, 10, 2, 0, 0, 0, 1, -1},
{"insert:MBME1-C012", tListLen(fullColList), fullColList, TTYPE_INSERT, 0, false, insertMBMETest1, 10, 10, 2, 12, 0, 0, 1, -1},
{"insert:MBME1-C002", tListLen(fullColList), fullColList, TTYPE_INSERT, 0, false, insertMBMETest1, 10, 10, 2, 2, 0, 0, 1, -1},
// 11
{"insert:MBME2-FULL", tListLen(fullColList), fullColList, TTYPE_INSERT, false, true, insertMBMETest2, 10, 10, 2, 0, 0, 0, 1, -1},
{"insert:MBME2-C012", tListLen(fullColList), fullColList, TTYPE_INSERT, false, false, insertMBMETest2, 10, 10, 2, 12, 0, 0, 1, -1},
{"insert:MBME2-C002", tListLen(fullColList), fullColList, TTYPE_INSERT, false, false, insertMBMETest2, 10, 10, 2, 2, 0, 0, 1, -1},
{"insert:MBME2-FULL", tListLen(fullColList), fullColList, TTYPE_INSERT, 0, true, insertMBMETest2, 10, 10, 2, 0, 0, 0, 1, -1},
{"insert:MBME2-C012", tListLen(fullColList), fullColList, TTYPE_INSERT, 0, false, insertMBMETest2, 10, 10, 2, 12, 0, 0, 1, -1},
{"insert:MBME2-C002", tListLen(fullColList), fullColList, TTYPE_INSERT, 0, false, insertMBMETest2, 10, 10, 2, 2, 0, 0, 1, -1},
{"insert:MBME3-FULL", tListLen(fullColList), fullColList, TTYPE_INSERT, false, true, insertMBMETest3, 10, 10, 2, 0, 0, 0, 1, -1},
{"insert:MBME3-C012", tListLen(fullColList), fullColList, TTYPE_INSERT, false, false, insertMBMETest3, 10, 10, 2, 12, 0, 0, 1, -1},
{"insert:MBME3-C002", tListLen(fullColList), fullColList, TTYPE_INSERT, false, false, insertMBMETest3, 10, 10, 2, 2, 0, 0, 1, -1},
{"insert:MBME3-FULL", tListLen(fullColList), fullColList, TTYPE_INSERT, 0, true, insertMBMETest3, 10, 10, 2, 0, 0, 0, 1, -1},
{"insert:MBME3-C012", tListLen(fullColList), fullColList, TTYPE_INSERT, 0, false, insertMBMETest3, 10, 10, 2, 12, 0, 0, 1, -1},
{"insert:MBME3-C002", tListLen(fullColList), fullColList, TTYPE_INSERT, 0, false, insertMBMETest3, 10, 10, 2, 2, 0, 0, 1, -1},
{"insert:MBME4-FULL", tListLen(fullColList), fullColList, TTYPE_INSERT, false, true, insertMBMETest4, 10, 10, 2, 0, 0, 0, 1, -1},
{"insert:MBME4-C012", tListLen(fullColList), fullColList, TTYPE_INSERT, false, false, insertMBMETest4, 10, 10, 2, 12, 0, 0, 1, -1},
{"insert:MBME4-C002", tListLen(fullColList), fullColList, TTYPE_INSERT, false, false, insertMBMETest4, 10, 10, 2, 2, 0, 0, 1, -1},
{"insert:MBME4-FULL", tListLen(fullColList), fullColList, TTYPE_INSERT, 0, true, insertMBMETest4, 10, 10, 2, 0, 0, 0, 1, -1},
{"insert:MBME4-C012", tListLen(fullColList), fullColList, TTYPE_INSERT, 0, false, insertMBMETest4, 10, 10, 2, 12, 0, 0, 1, -1},
{"insert:MBME4-C002", tListLen(fullColList), fullColList, TTYPE_INSERT, 0, false, insertMBMETest4, 10, 10, 2, 2, 0, 0, 1, -1},
{"insert:MPME1-FULL", tListLen(fullColList), fullColList, TTYPE_INSERT, false, true, insertMPMETest1, 10, 10, 2, 0, 0, 0, 1, -1},
{"insert:MPME1-C012", tListLen(fullColList), fullColList, TTYPE_INSERT, false, false, insertMPMETest1, 10, 10, 2, 12, 0, 0, 1, -1},
{"insert:MPME1-FULL", tListLen(fullColList), fullColList, TTYPE_INSERT, 0, true, insertMPMETest1, 10, 10, 2, 0, 0, 0, 1, -1},
{"insert:MPME1-C012", tListLen(fullColList), fullColList, TTYPE_INSERT, 0, false, insertMPMETest1, 10, 10, 2, 12, 0, 0, 1, -1},
// 22
{"insert:AUTO1-FULL", tListLen(fullColList), fullColList, TTYPE_INSERT, true, true, insertAUTOTest1, 10, 10, 2, 0, 0, 0, 1, -1},
{"insert:AUTO1-FULL", tListLen(fullColList), fullColList, TTYPE_INSERT, 1, true, insertAUTOTest1, 10, 10, 2, 0, 0, 0, 1, -1},
{"insert:AUTO1-TBEXISTS", tListLen(fullColList), fullColList, TTYPE_INSERT, 3, true, insertAUTOTest2, 10, 10, 2, 0, 0, 0, 1, -1},
{"query:SUBT-COLUMN", tListLen(fullColList), fullColList, TTYPE_QUERY, false, false, queryColumnTest, 10, 10, 1, 3, 0, 0, 1, 2},
{"query:SUBT-MISC", tListLen(fullColList), fullColList, TTYPE_QUERY, false, false, queryMiscTest, 10, 10, 1, 3, 0, 0, 1, 2},
{"query:SUBT-COLUMN", tListLen(fullColList), fullColList, TTYPE_QUERY, 0, false, queryColumnTest, 10, 10, 1, 3, 0, 0, 1, 2},
{"query:SUBT-MISC", tListLen(fullColList), fullColList, TTYPE_QUERY, 0, false, queryMiscTest, 10, 10, 1, 3, 0, 0, 1, 2},
// {"query:SUBT-COLUMN", tListLen(fullColList), fullColList, TTYPE_QUERY, false, false, queryColumnTest, 1, 10, 1, 1, 0, 0, 1, 2},
// {"query:SUBT-MISC", tListLen(fullColList), fullColList, TTYPE_QUERY, false, false, queryMiscTest, 2, 10, 1, 1, 0, 0, 1, 2},
// {"query:SUBT-COLUMN", tListLen(fullColList), fullColList, TTYPE_QUERY, 0, false, queryColumnTest, 1, 10, 1, 1, 0, 0, 1, 2},
// {"query:SUBT-MISC", tListLen(fullColList), fullColList, TTYPE_QUERY, 0, false, queryMiscTest, 2, 10, 1, 1, 0, 0, 1, 2},
};
@ -221,7 +223,7 @@ typedef struct {
CaseCtrl gCaseCtrl = {
.precision = TIME_PRECISION_MICRO,
.bindNullNum = 0,
.printCreateTblSql = false,
.printCreateTblSql = true,
.printQuerySql = true,
.printStmtSql = true,
.printVerbose = false,
@ -230,7 +232,7 @@ CaseCtrl gCaseCtrl = {
.numericParam = false,
.rowNum = 0,
.bindColNum = 0,
.bindTagNum = 0,
.bindTagNum = 14,
.bindRowNum = 0,
.bindColTypeNum = 0,
.bindColTypeList = NULL,
@ -242,8 +244,8 @@ CaseCtrl gCaseCtrl = {
.funcIdxList = NULL,
.checkParamNum = false,
.runTimes = 0,
.caseIdx = -1,
.caseNum = -1,
.caseIdx = 23,
.caseNum = 1,
.caseRunIdx = -1,
.caseRunNum = -1,
};
@ -1946,6 +1948,73 @@ int insertAUTOTest1(TAOS_STMT *stmt, TAOS *taos) {
}
/* [prepare [settbnametag [bind add exec]]] */
int insertAUTOTest2(TAOS_STMT *stmt, TAOS *taos) {
int32_t loop = 0;
while (gCurCase->bindTagNum > 0 && gCurCase->bindColNum > 0) {
BindData data = {0};
prepareInsertData(&data);
int code = taos_stmt_prepare(stmt, data.sql, 0);
if (code != 0){
printf("!!!failed to execute taos_stmt_prepare. error:%s\n", taos_stmt_errstr(stmt));
exit(1);
}
bpCheckIsInsert(stmt, 1);
int32_t bindTimes = gCurCase->rowNum/gCurCase->bindRowNum;
for (int32_t b = 0; b <bindTimes; ++b) {
for (int32_t t = 0; t< gCurCase->tblNum; ++t) {
if (gCurCase->tblNum > 1) {
char buf[32];
sprintf(buf, "t%d", t);
code = bpSetTableNameTags(&data, t, buf, stmt);
if (code != 0){
printf("!!!taos_stmt_set_tbname_tags error:%s\n", taos_stmt_errstr(stmt));
exit(1);
}
}
if (gCaseCtrl.checkParamNum) {
bpCheckParamNum(stmt);
}
if (bpBindParam(stmt, data.pBind + t*bindTimes*gCurCase->bindColNum + b*gCurCase->bindColNum)) {
exit(1);
}
if (taos_stmt_add_batch(stmt)) {
printf("!!!taos_stmt_add_batch error:%s\n", taos_stmt_errstr(stmt));
exit(1);
}
if (taos_stmt_execute(stmt) != 0) {
printf("!!!taos_stmt_execute error:%s\n", taos_stmt_errstr(stmt));
exit(1);
}
}
}
bpCheckIsInsert(stmt, 1);
destroyData(&data);
gCurCase->bindColNum -= 2;
gCurCase->bindTagNum -= 2;
gCurCase->fullCol = false;
loop++;
}
bpCheckAffectedRows(stmt, loop);
gExecLoopTimes = loop;
return 0;
}
/* select * from table */
int queryColumnTest(TAOS_STMT *stmt, TAOS *taos) {
BindData data = {0};
@ -2243,70 +2312,76 @@ int sql_s_perf1(TAOS *taos) {
return 0;
}
void generateCreateTableSQL(char *buf, int32_t tblIdx, int32_t colNum, int32_t *colList, bool stable) {
void generateCreateTableSQL(char *buf, int32_t tblIdx, int32_t colNum, int32_t *colList, int32_t tableType) {
int32_t blen = 0;
blen = sprintf(buf, "create table %s%d ", (stable ? bpStbPrefix : bpTbPrefix), tblIdx);
blen = sprintf(buf, "create table %s%d ", (1 == tableType ? bpStbPrefix : bpTbPrefix), tblIdx);
blen += sprintf(buf + blen, " (");
for (int c = 0; c < colNum; ++c) {
if (c > 0) {
blen += sprintf(buf + blen, ",");
}
switch (colList[c]) {
case TSDB_DATA_TYPE_BOOL:
blen += sprintf(buf + blen, "booldata bool");
break;
case TSDB_DATA_TYPE_TINYINT:
blen += sprintf(buf + blen, "tinydata tinyint");
break;
case TSDB_DATA_TYPE_SMALLINT:
blen += sprintf(buf + blen, "smalldata smallint");
break;
case TSDB_DATA_TYPE_INT:
blen += sprintf(buf + blen, "intdata int");
break;
case TSDB_DATA_TYPE_BIGINT:
blen += sprintf(buf + blen, "bigdata bigint");
break;
case TSDB_DATA_TYPE_FLOAT:
blen += sprintf(buf + blen, "floatdata float");
break;
case TSDB_DATA_TYPE_DOUBLE:
blen += sprintf(buf + blen, "doubledata double");
break;
case TSDB_DATA_TYPE_VARCHAR:
blen += sprintf(buf + blen, "binarydata binary(%d)", gVarCharSize);
break;
case TSDB_DATA_TYPE_TIMESTAMP:
blen += sprintf(buf + blen, "ts timestamp");
break;
case TSDB_DATA_TYPE_NCHAR:
blen += sprintf(buf + blen, "nchardata nchar(%d)", gVarCharSize);
break;
case TSDB_DATA_TYPE_UTINYINT:
blen += sprintf(buf + blen, "utinydata tinyint unsigned");
break;
case TSDB_DATA_TYPE_USMALLINT:
blen += sprintf(buf + blen, "usmalldata smallint unsigned");
break;
case TSDB_DATA_TYPE_UINT:
blen += sprintf(buf + blen, "uintdata int unsigned");
break;
case TSDB_DATA_TYPE_UBIGINT:
blen += sprintf(buf + blen, "ubigdata bigint unsigned");
break;
default:
printf("invalid col type:%d", colList[c]);
exit(1);
}
if (tableType == 3) {
blen += sprintf(buf + blen, "using %s%d", bpStbPrefix, bpDefaultStbId);
}
blen += sprintf(buf + blen, ")");
if (tableType == 0 || tableType == 1) {
blen += sprintf(buf + blen, " (");
if (stable) {
blen += sprintf(buf + blen, "tags (");
for (int c = 0; c < colNum; ++c) {
if (c > 0) {
blen += sprintf(buf + blen, ",");
}
switch (colList[c]) {
case TSDB_DATA_TYPE_BOOL:
blen += sprintf(buf + blen, "booldata bool");
break;
case TSDB_DATA_TYPE_TINYINT:
blen += sprintf(buf + blen, "tinydata tinyint");
break;
case TSDB_DATA_TYPE_SMALLINT:
blen += sprintf(buf + blen, "smalldata smallint");
break;
case TSDB_DATA_TYPE_INT:
blen += sprintf(buf + blen, "intdata int");
break;
case TSDB_DATA_TYPE_BIGINT:
blen += sprintf(buf + blen, "bigdata bigint");
break;
case TSDB_DATA_TYPE_FLOAT:
blen += sprintf(buf + blen, "floatdata float");
break;
case TSDB_DATA_TYPE_DOUBLE:
blen += sprintf(buf + blen, "doubledata double");
break;
case TSDB_DATA_TYPE_VARCHAR:
blen += sprintf(buf + blen, "binarydata binary(%d)", gVarCharSize);
break;
case TSDB_DATA_TYPE_TIMESTAMP:
blen += sprintf(buf + blen, "ts timestamp");
break;
case TSDB_DATA_TYPE_NCHAR:
blen += sprintf(buf + blen, "nchardata nchar(%d)", gVarCharSize);
break;
case TSDB_DATA_TYPE_UTINYINT:
blen += sprintf(buf + blen, "utinydata tinyint unsigned");
break;
case TSDB_DATA_TYPE_USMALLINT:
blen += sprintf(buf + blen, "usmalldata smallint unsigned");
break;
case TSDB_DATA_TYPE_UINT:
blen += sprintf(buf + blen, "uintdata int unsigned");
break;
case TSDB_DATA_TYPE_UBIGINT:
blen += sprintf(buf + blen, "ubigdata bigint unsigned");
break;
default:
printf("invalid col type:%d", colList[c]);
exit(1);
}
}
blen += sprintf(buf + blen, ")");
}
if (1 == tableType) {
blen += sprintf(buf + blen, " tags (");
for (int c = 0; c < colNum; ++c) {
if (c > 0) {
blen += sprintf(buf + blen, ",");
@ -2363,6 +2438,64 @@ void generateCreateTableSQL(char *buf, int32_t tblIdx, int32_t colNum, int32_t *
blen += sprintf(buf + blen, ")");
}
if (3 == tableType) {
blen += sprintf(buf + blen, " tags (");
for (int c = 0; c < colNum; ++c) {
if (c > 0) {
blen += sprintf(buf + blen, ",");
}
switch (colList[c]) {
case TSDB_DATA_TYPE_BOOL:
blen += sprintf(buf + blen, "%s", rand() % 2 ? "true": "false");
break;
case TSDB_DATA_TYPE_TINYINT:
blen += sprintf(buf + blen, "%d", rand() % 128);
break;
case TSDB_DATA_TYPE_SMALLINT:
blen += sprintf(buf + blen, "%d", rand() % 128);
break;
case TSDB_DATA_TYPE_INT:
blen += sprintf(buf + blen, "%d", rand() % 128);
break;
case TSDB_DATA_TYPE_BIGINT:
blen += sprintf(buf + blen, "%d", rand() % 128);
break;
case TSDB_DATA_TYPE_FLOAT:
blen += sprintf(buf + blen, "%f", rand() % 128);
break;
case TSDB_DATA_TYPE_DOUBLE:
blen += sprintf(buf + blen, "%f", rand() % 128);
break;
case TSDB_DATA_TYPE_VARCHAR:
blen += sprintf(buf + blen, "'var%d'", rand() % 128);
break;
case TSDB_DATA_TYPE_TIMESTAMP:
blen += sprintf(buf + blen, "%lld", bpTs);
break;
case TSDB_DATA_TYPE_NCHAR:
blen += sprintf(buf + blen, "'nch%d'", rand() % 128);
break;
case TSDB_DATA_TYPE_UTINYINT:
blen += sprintf(buf + blen, "%d", rand() % 128);
break;
case TSDB_DATA_TYPE_USMALLINT:
blen += sprintf(buf + blen, "%d", rand() % 128);
break;
case TSDB_DATA_TYPE_UINT:
blen += sprintf(buf + blen, "%d", rand() % 128);
break;
case TSDB_DATA_TYPE_UBIGINT:
blen += sprintf(buf + blen, "%d", rand() % 128);
break;
default:
printf("invalid col type:%d", colList[c]);
exit(1);
}
}
blen += sprintf(buf + blen, ")");
}
if (gCaseCtrl.printCreateTblSql) {
printf("\tCreate Table SQL:%s\n", buf);
}
@ -2421,11 +2554,11 @@ void prepare(TAOS *taos, int32_t colNum, int32_t *colList, int prepareStb) {
result = taos_query(taos, "use demo");
taos_free_result(result);
if (!prepareStb) {
if (0 == prepareStb) {
// create table
for (int i = 0 ; i < 10; i++) {
char buf[1024];
generateCreateTableSQL(buf, i, colNum, colList, false);
generateCreateTableSQL(buf, i, colNum, colList, 0);
result = taos_query(taos, buf);
code = taos_errno(result);
if (code != 0) {
@ -2436,17 +2569,35 @@ void prepare(TAOS *taos, int32_t colNum, int32_t *colList, int prepareStb) {
taos_free_result(result);
}
} else {
char buf[1024];
generateCreateTableSQL(buf, bpDefaultStbId, colNum, colList, true);
if (1 == prepareStb || 3 == prepareStb) {
char buf[1024];
generateCreateTableSQL(buf, bpDefaultStbId, colNum, colList, 1);
result = taos_query(taos, buf);
code = taos_errno(result);
if (code != 0) {
printf("!!!failed to create table, reason:%s\n", taos_errstr(result));
result = taos_query(taos, buf);
code = taos_errno(result);
if (code != 0) {
printf("!!!failed to create table, reason:%s\n", taos_errstr(result));
taos_free_result(result);
exit(1);
}
taos_free_result(result);
exit(1);
}
taos_free_result(result);
if (3 == prepareStb) {
for (int i = 0 ; i < 10; i++) {
char buf[1024];
generateCreateTableSQL(buf, i, colNum, colList, 3);
result = taos_query(taos, buf);
code = taos_errno(result);
if (code != 0) {
printf("!!!failed to create table, reason:%s\n", taos_errstr(result));
taos_free_result(result);
exit(1);
}
taos_free_result(result);
}
}
}
}
@ -2486,7 +2637,7 @@ int32_t runCase(TAOS *taos, int32_t caseIdx, int32_t caseRunIdx, bool silent) {
if (gCaseCtrl.autoCreateTbl) {
if (gCurCase->testType == TTYPE_INSERT && gCurCase->tblNum > 1) {
gCurCase->autoCreateTbl = true;
gCurCase->autoCreateTbl = 1;
if (gCurCase->bindTagNum <= 0) {
gCurCase->bindTagNum = gCurCase->colNum;
}

3
tests/script/sh/checkAsan.sh
View File

@ -37,8 +37,9 @@ python_error=`cat ${LOG_DIR}/*.info | grep -w "stack" | wc -l`
# TD-20569
# /root/TDengine/source/libs/function/src/builtinsimpl.c:856:29: runtime error: signed integer overflow: 9223372036854775806 + 9223372036854775805 cannot be represented in type 'long int'
# /root/TDengine/source/libs/scalar/src/sclvector.c:1075:66: runtime error: signed integer overflow: 9223372034707292160 + 1668838476672 cannot be represented in type 'long int'
# /root/TDengine/source/common/src/tdataformat.c:1876:7: runtime error: signed integer overflow: 8252423483843671206 + 2406154664059062870 cannot be represented in type 'long int'
runtime_error=`cat ${LOG_DIR}/*.asan | grep "runtime error" | grep -v "trees.c:873" | grep -v "sclfunc.c.*outside the range of representable values of type"| grep -v "builtinsimpl.c.*signed integer overflow"| grep -v "sclvector.c.*signed integer overflow" | grep -v "tdataformat.c.*signed integer overflow" |wc -l`
runtime_error=`cat ${LOG_DIR}/*.asan | grep "runtime error" | grep -v "trees.c:873" | grep -v "sclfunc.c.*outside the range of representable values of type"| grep -v "signed integer overflow" | wc -l`
echo -e "\033[44;32;1m"asan error_num: $error_num"\033[0m"
echo -e "\033[44;32;1m"asan memory_leak: $memory_leak"\033[0m"

60
tests/script/tsim/stream/state0.sim
View File

@ -731,5 +731,65 @@ if $data32 != 1 then
goto loop9
endi
sql drop stream if exists streams5;
sql drop database if exists test5;
sql create database test5;
sql use test5;
sql create table tb (ts timestamp, a int);
sql insert into tb values (now + 1m , 1 );
sql create table b (c timestamp, d int, e int , f int, g double);
sql create stream streams0 trigger at_once into streamt as select _wstart c1, count(*) c2, max(a) c3 from tb state_window(a);
sql insert into b values(1648791213000,NULL,NULL,NULL,NULL);
sql select * from streamt order by c1, c2, c3;
print data00:$data00
print data01:$data01
sql insert into b values(1648791213000,NULL,NULL,NULL,NULL);
sql select * from streamt order by c1, c2, c3;
print data00:$data00
print data01:$data01
sql insert into b values(1648791213001,1,2,2,2.0);
sql insert into b values(1648791213002,1,3,3,3.0);
sql insert into tb values(1648791213003,1);
sql select * from streamt;
print data00:$data00
print data01:$data01
sql delete from b where c >= 1648791213001 and c <= 1648791213002;
sql insert into b values(1648791223003,2,2,3,1.0); insert into b values(1648791223002,2,3,3,3.0);
sql insert into tb values (now + 1m , 1 );
sql select * from streamt;
print data00:$data00
print data01:$data01
sql insert into b(c,d) values (now + 6m , 6 );
sql delete from b where c >= 1648791213001 and c <= 1648791233005;;
$loop_count = 0
loop10:
sleep 200
sql select c2 from streamt;
$loop_count = $loop_count + 1
if $loop_count == 20 then
return -1
endi
if $rows != 1 then
print =====rows=$rows
goto loop10
endi
if $data00 != 2 then
print =====data00=$data00
goto loop10
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT

4
tests/system-test/1-insert/alter_database.py
View File

@ -18,7 +18,7 @@ class TDTestCase:
tdSql.init(conn.cursor(), logSql)
self.buffer_boundary = [3, 4097, 8193, 12289, 16384]
self.buffer_error = [self.buffer_boundary[0] -
1, self.buffer_boundary[-1]+1, 12289, 256]
1, self.buffer_boundary[-1]+1, 256]
# pages_boundary >= 64
self.pages_boundary = [64, 128, 512]
self.pages_error = [self.pages_boundary[0]-1]
@ -53,7 +53,7 @@ class TDTestCase:
tdSql.execute('drop database db')
def run(self):
tdSql.error('create database db1 vgroups 10 buffer 12289')
self.alter_buffer()
self.alter_pages()

7
tests/system-test/1-insert/alter_table.py
View File

@ -87,7 +87,6 @@ class TDTestCase:
}
def alter_check_ntb(self):
tdSql.prepare()
tdSql.execute(self.setsql.set_create_normaltable_sql(self.ntbname,self.column_dict))
for i in self.values_list:
@ -107,6 +106,7 @@ class TDTestCase:
v = f'binary({self.binary_length+1})'
v_error = f'binary({self.binary_length-1})'
tdSql.error(f'alter table {self.ntbname} modify column {key} {v_error}')
tdSql.error(f'alter table {self.ntbname} set tag {key} = "abcd1"')
tdSql.execute(f'alter table {self.ntbname} modify column {key} {v}')
tdSql.query(f'describe {self.ntbname}')
result = tdCom.getOneRow(1,'VARCHAR')
@ -115,6 +115,7 @@ class TDTestCase:
v = f'nchar({self.binary_length+1})'
v_error = f'nchar({self.binary_length-1})'
tdSql.error(f'alter table {self.ntbname} modify column {key} {v_error}')
tdSql.error(f'alter table {self.ntbname} set tag {key} = "abcd1"')
tdSql.execute(f'alter table {self.ntbname} modify column {key} {v}')
tdSql.query(f'describe {self.ntbname}')
result = tdCom.getOneRow(1,'NCHAR')
@ -122,6 +123,7 @@ class TDTestCase:
else:
for v in self.column_dict.values():
tdSql.error(f'alter table {self.ntbname} modify column {key} {v}')
tdSql.error(f'alter table {self.ntbname} set tag {key} = "abcd1"')
for key,values in self.column_dict.items():
rename_str = f'{tdCom.getLongName(constant.COL_NAME_LENGTH_MAX,"letters")}'
tdSql.execute(f'alter table {self.ntbname} rename column {key} {rename_str}')
@ -285,6 +287,7 @@ class TDTestCase:
v = f'binary({self.binary_length+1})'
v_error = f'binary({self.binary_length-1})'
tdSql.error(f'alter table {self.stbname} modify column {key} {v_error}')
tdSql.error(f'alter table {self.stbname} set tag {key} = "abcd1"')
tdSql.execute(f'alter table {self.stbname} modify column {key} {v}')
tdSql.query(f'describe {self.stbname}')
result = tdCom.getOneRow(1,'VARCHAR')
@ -297,6 +300,7 @@ class TDTestCase:
v = f'nchar({self.binary_length+1})'
v_error = f'nchar({self.binary_length-1})'
tdSql.error(f'alter table {self.stbname} modify column {key} {v_error}')
tdSql.error(f'alter table {self.stbname} set tag {key} = "abcd1"')
tdSql.execute(f'alter table {self.stbname} modify column {key} {v}')
tdSql.query(f'describe {self.stbname}')
result = tdCom.getOneRow(1,'NCHAR')
@ -308,6 +312,7 @@ class TDTestCase:
else:
for v in self.column_dict.values():
tdSql.error(f'alter table {self.stbname} modify column {key} {v}')
tdSql.error(f'alter table {self.stbname} set tag {key} = "abcd1"')
for key,values in self.column_dict.items():
rename_str = f'{tdCom.getLongName(constant.COL_NAME_LENGTH_MAX,"letters")}'
tdSql.error(f'alter table {self.stbname} rename column {key} {rename_str}')

161
tests/system-test/1-insert/drop.py Normal file
View File

@ -0,0 +1,161 @@
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import math
from util.log import *
from util.cases import *
from util.sql import *
from util.common import *
from util.sqlset import *
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.setsql = TDSetSql()
self.dbname = 'db'
self.ntbname = f"{self.dbname}.ntb"
self.rowNum = 10
self.tbnum = 20
self.ts = 1537146000000
self.binary_str = 'taosdata'
self.nchar_str = '涛思数据'
self.column_dict = {
'ts' : 'timestamp',
'col1': 'tinyint',
'col2': 'smallint',
'col3': 'int',
'col4': 'bigint',
'col5': 'tinyint unsigned',
'col6': 'smallint unsigned',
'col7': 'int unsigned',
'col8': 'bigint unsigned',
'col9': 'float',
'col10': 'double',
'col11': 'bool',
'col12': 'binary(20)',
'col13': 'nchar(20)'
}
def insert_data(self,column_dict,tbname,row_num):
insert_sql = self.setsql.set_insertsql(column_dict,tbname,self.binary_str,self.nchar_str)
for i in range(row_num):
insert_list = []
self.setsql.insert_values(column_dict,i,insert_sql,insert_list,self.ts)
def drop_ntb_check(self):
tdSql.execute(f'create database if not exists {self.dbname} replica {self.replicaVar}')
tdSql.execute(f'use {self.dbname}')
tdSql.execute(self.setsql.set_create_normaltable_sql(self.ntbname,self.column_dict))
self.insert_data(self.column_dict,self.ntbname,self.rowNum)
for k,v in self.column_dict.items():
if v.lower() == "timestamp":
tdSql.query(f'select * from {self.ntbname} where {k} = {self.ts}')
tdSql.checkRows(1)
tdSql.execute(f'drop table {self.ntbname}')
tdSql.execute(f'flush database {self.dbname}')
tdSql.execute(self.setsql.set_create_normaltable_sql(self.ntbname,self.column_dict))
self.insert_data(self.column_dict,self.ntbname,self.rowNum)
for k,v in self.column_dict.items():
if v.lower() == "timestamp":
tdSql.query(f'select * from {self.ntbname} where {k} = {self.ts}')
tdSql.checkRows(1)
tdSql.execute(f'drop database {self.dbname}')
def drop_stb_ctb_check(self):
stbname = f'{self.dbname}.{tdCom.getLongName(5,"letters")}'
tag_dict = {
't0':'int'
}
tag_values = [
f'1'
]
tdSql.execute(f"create database if not exists {self.dbname} replica {self.replicaVar}")
tdSql.execute(f'use {self.dbname}')
tdSql.execute(self.setsql.set_create_stable_sql(stbname,self.column_dict,tag_dict))
for i in range(self.tbnum):
tdSql.execute(f"create table {stbname}_{i} using {stbname} tags({tag_values[0]})")
self.insert_data(self.column_dict,f'{stbname}_{i}',self.rowNum)
for k,v in self.column_dict.items():
for i in range(self.tbnum):
if v.lower() == "timestamp":
tdSql.query(f'select * from {stbname}_{i} where {k} = {self.ts}')
tdSql.checkRows(1)
tdSql.execute(f'drop table {stbname}_{i}')
tdSql.execute(f'flush database {self.dbname}')
for i in range(self.tbnum):
tdSql.execute(f"create table {stbname}_{i} using {stbname} tags({tag_values[0]})")
self.insert_data(self.column_dict,f'{stbname}_{i}',self.rowNum)
for k,v in self.column_dict.items():
for i in range(self.tbnum):
if v.lower() == "timestamp":
tdSql.query(f'select * from {stbname}_{i} where {k} = {self.ts}')
tdSql.checkRows(1)
if v.lower() == "timestamp":
tdSql.query(f'select * from {stbname} where {k} = {self.ts}')
tdSql.checkRows(self.tbnum)
tdSql.execute(f'drop table {stbname}')
tdSql.execute(f'flush database {self.dbname}')
tdSql.execute(self.setsql.set_create_stable_sql(stbname,self.column_dict,tag_dict))
for i in range(self.tbnum):
tdSql.execute(f"create table {stbname}_{i} using {stbname} tags({tag_values[0]})")
self.insert_data(self.column_dict,f'{stbname}_{i}',self.rowNum)
for k,v in self.column_dict.items():
if v.lower() == "timestamp":
tdSql.query(f'select * from {stbname} where {k} = {self.ts}')
tdSql.checkRows(self.tbnum)
tdSql.execute(f'drop database {self.dbname}')
def drop_topic_check(self):
tdSql.execute(f'create database {self.dbname} replica {self.replicaVar}')
tdSql.execute(f'use {self.dbname}')
stbname = tdCom.getLongName(5,"letters")
topic_name = tdCom.getLongName(5,"letters")
tdSql.execute(f'create table {stbname} (ts timestamp,c0 int) tags(t0 int)')
tdSql.execute(f'create topic {topic_name} as select * from {self.dbname}.{stbname}')
tdSql.query(f'select * from information_schema.ins_topics where topic_name = "{topic_name}"')
tdSql.checkEqual(tdSql.queryResult[0][3],f'create topic {topic_name} as select * from {self.dbname}.{stbname}')
tdSql.execute(f'drop topic {topic_name}')
tdSql.execute(f'create topic {topic_name} as select c0 from {self.dbname}.{stbname}')
tdSql.query(f'select * from information_schema.ins_topics where topic_name = "{topic_name}"')
tdSql.checkEqual(tdSql.queryResult[0][3],f'create topic {topic_name} as select c0 from {self.dbname}.{stbname}')
tdSql.execute(f'drop topic {topic_name}')
tdSql.execute(f'drop database {self.dbname}')
def drop_stream_check(self):
tdSql.execute(f'create database {self.dbname} replica {self.replicaVar}')
tdSql.execute(f'use {self.dbname}')
stbname = tdCom.getLongName(5,"letters")
stream_name = tdCom.getLongName(5,"letters")
tdSql.execute(f'create table {stbname} (ts timestamp,c0 int) tags(t0 int)')
tdSql.execute(f'create table tb using {stbname} tags(1)')
tdSql.execute(f'create stream {stream_name} into stb as select * from {self.dbname}.{stbname} partition by tbname')
tdSql.query(f'select * from information_schema.ins_streams where stream_name = "{stream_name}"')
print(tdSql.queryResult)
tdSql.checkEqual(tdSql.queryResult[0][2],f'create stream {stream_name} into stb as select * from {self.dbname}.{stbname} partition by tbname')
tdSql.execute(f'drop stream {stream_name}')
tdSql.execute(f'create stream {stream_name} into stb1 as select * from tb')
tdSql.query(f'select * from information_schema.ins_streams where stream_name = "{stream_name}"')
tdSql.checkEqual(tdSql.queryResult[0][2],f'create stream {stream_name} into stb1 as select * from tb')
tdSql.execute(f'drop database {self.dbname}')
def run(self):
self.drop_ntb_check()
self.drop_stb_ctb_check()
self.drop_topic_check()
self.drop_stream_check()
pass
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

25
tests/system-test/2-query/smaTest.py
View File

@ -44,8 +44,8 @@ class TDTestCase:
def run(self):
# insert data
dbname = "db"
self.insert_data1(f"{dbname}.t1", self.ts, 1000*10000)
self.insert_data1(f"{dbname}.t4", self.ts, 1000*10000)
self.insert_data1(f"{dbname}.t1", self.ts, 10*10000)
self.insert_data1(f"{dbname}.t4", self.ts, 10*10000)
# test base case
# self.test_case1()
tdLog.debug(" LIMIT test_case1 ............ [OK]")
@ -53,7 +53,6 @@ class TDTestCase:
# self.test_case2()
tdLog.debug(" LIMIT test_case2 ............ [OK]")
# stop
def stop(self):
tdSql.close()
@ -77,15 +76,17 @@ class TDTestCase:
# insert data1
def insert_data(self, tbname, ts_start, count):
pre_insert = "insert into %s values"%tbname
pre_insert = "insert into %s values" % tbname
sql = pre_insert
tdLog.debug("insert table %s rows=%d ..."%(tbname, count))
tdLog.debug("insert table %s rows=%d ..." % (tbname, count))
for i in range(count):
sql += " (%d,%d)"%(ts_start + i*1000, i )
if i >0 and i%30000 == 0:
sql += " (%d,%d)" % (ts_start + i*1000, i)
if i > 0 and i % 20000 == 0:
tdLog.info("%d rows inserted" % i)
tdSql.execute(sql)
sql = pre_insert
# end sql
tdLog.info("insert_data end")
if sql != pre_insert:
tdSql.execute(sql)
@ -93,15 +94,17 @@ class TDTestCase:
return
def insert_data1(self, tbname, ts_start, count):
pre_insert = "insert into %s values"%tbname
pre_insert = "insert into %s values" % tbname
sql = pre_insert
tdLog.debug("insert table %s rows=%d ..."%(tbname, count))
tdLog.debug("insert table %s rows=%d ..." % (tbname, count))
for i in range(count):
sql += " (%d,%d,%d)"%(ts_start + i*1000, i , i+1)
if i >0 and i%30000 == 0:
sql += " (%d,%d,%d)" % (ts_start + i*1000, i, i+1)
if i > 0 and i % 20000 == 0:
tdLog.info("%d rows inserted" % i)
tdSql.execute(sql)
sql = pre_insert
# end sql
tdLog.info("insert_data1 end")
if sql != pre_insert:
tdSql.execute(sql)

1
tests/system-test/test.py
View File

@ -464,6 +464,7 @@ if __name__ == "__main__":
tdDnodes.init(deployPath, masterIp)
tdDnodes.setTestCluster(testCluster)
tdDnodes.setValgrind(valgrind)
tdDnodes.setAsan(asan)
tdDnodes.stopAll()
for dnode in tdDnodes.dnodes:
tdDnodes.deploy(dnode.index,{})

3
tools/shell/inc/shellAuto.h
View File

@ -39,4 +39,7 @@ void callbackAutoTab(char* sqlstr, TAOS* pSql, bool usedb);
// introduction
void printfIntroduction();
// show all commands help
void showHelp();
#endif

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

@ -108,6 +108,7 @@ SWords shellCommands[] = {
{"drop topic <topic_name> ;", 0, 0, NULL},
{"drop stream <stream_name> ;", 0, 0, NULL},
{"explain select", 0, 0, NULL}, // 44 append sub sql
{"help;", 0, 0, NULL},
{"grant all on <anyword> to <user_name> ;", 0, 0, NULL},
{"grant read on <anyword> to <user_name> ;", 0, 0, NULL},
{"grant write on <anyword> to <user_name> ;", 0, 0, NULL},
@ -386,6 +387,8 @@ void showHelp() {
drop stream <stream_name> ;\n\
----- E ----- \n\
explain select clause ...\n\
----- H ----- \n\
help;\n\
----- I ----- \n\
insert into <tb_name> values(...) ;\n\
insert into <tb_name> using <stb_name> tags(...) values(...) ;\n\
@ -1478,24 +1481,36 @@ bool matchSelectQuery(TAOS* con, SShellCmd* cmd) {
// if is input create fields or tags area, return true
bool isCreateFieldsArea(char* p) {
char* left = strrchr(p, '(');
if (left == NULL) {
// like 'create table st'
return false;
}
// put to while, support like create table st(ts timestamp, bin1 binary(16), bin2 + blank + TAB
char* p1 = strdup(p);
bool ret = false;
while (1) {
char* left = strrchr(p1, '(');
if (left == NULL) {
// like 'create table st'
ret = false;
break;
}
char* right = strrchr(p, ')');
if (right == NULL) {
// like 'create table st( '
return true;
}
char* right = strrchr(p1, ')');
if (right == NULL) {
// like 'create table st( '
ret = true;
break;
}
if (left > right) {
// like 'create table st( ts timestamp, age int) tags(area '
return true;
}
if (left > right) {
// like 'create table st( ts timestamp, age int) tags(area '
ret = true;
break;
}
return false;
// set string end by small for next strrchr search
*left = 0;
}
taosMemoryFree(p1);
return ret;
}
bool matchCreateTable(TAOS* con, SShellCmd* cmd) {

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

@ -134,6 +134,12 @@ int32_t shellRunCommand(char *command, bool recordHistory) {
return 0;
}
// add help or help;
if(strcmp(command, "help") == 0 || strcmp(command, "help;") == 0) {
showHelp();
return 0;
}
if (recordHistory) shellRecordCommandToHistory(command);
char quote = 0, *cmd = command;