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

Merge branch '3.0' into feature/TD-11274-3.0

This commit is contained in:
Cary Xu 2022-06-15 17:31:37 +08:00
parent 53783796f4 d7a6c4976c
commit 8cded9e73b
41 changed files with 709 additions and 326 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
examples/c/stream_demo.c
View File

@ -82,7 +82,8 @@ int32_t create_stream() {
/*const char* sql = "select sum(k) from tu1 interval(10m)";*/ /*const char* sql = "select sum(k) from tu1 interval(10m)";*/
/*pRes = tmq_create_stream(pConn, "stream1", "out1", sql);*/ /*pRes = tmq_create_stream(pConn, "stream1", "out1", sql);*/
pRes = taos_query( pRes = taos_query(
pConn, "create stream stream1 trigger at_once into outstb as select _wstartts, sum(k) from st1 interval(10m)"); pConn,
"create stream stream1 trigger max_delay 10s into outstb as select _wstartts, sum(k) from st1 interval(10m)");
if (taos_errno(pRes) != 0) { if (taos_errno(pRes) != 0) {
printf("failed to create stream stream1, reason:%s\n", taos_errstr(pRes)); printf("failed to create stream stream1, reason:%s\n", taos_errstr(pRes));
return -1; return -1;

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

@ -38,8 +38,10 @@ typedef struct SReadHandle {
SMsgCb* pMsgCb; SMsgCb* pMsgCb;
} SReadHandle; } SReadHandle;
#define STREAM_DATA_TYPE_SUBMIT_BLOCK 0x1 enum {
#define STREAM_DATA_TYPE_SSDATA_BLOCK 0x2 STREAM_DATA_TYPE_SUBMIT_BLOCK = 1,
STREAM_DATA_TYPE_SSDATA_BLOCK = 2,
};
typedef enum { typedef enum {
OPTR_EXEC_MODEL_BATCH = 0x1, OPTR_EXEC_MODEL_BATCH = 0x1,
@ -102,7 +104,8 @@ int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId,
* @param tversion * @param tversion
* @return * @return
*/ */
int32_t qGetQueriedTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, char* tableName, int32_t* sversion, int32_t* tversion); int32_t qGetQueriedTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, char* tableName, int32_t* sversion,
int32_t* tversion);
/** /**
* The main task execution function, including query on both table and multiple tables, * The main task execution function, including query on both table and multiple tables,

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

@ -58,6 +58,7 @@ enum {
enum { enum {
STREAM_INPUT__DATA_SUBMIT = 1, STREAM_INPUT__DATA_SUBMIT = 1,
STREAM_INPUT__DATA_BLOCK, STREAM_INPUT__DATA_BLOCK,
STREAM_INPUT__TRIGGER,
STREAM_INPUT__CHECKPOINT, STREAM_INPUT__CHECKPOINT,
}; };
@ -85,6 +86,11 @@ typedef struct {
int8_t type; int8_t type;
} SStreamCheckpoint; } SStreamCheckpoint;
typedef struct {
int8_t type;
SSDataBlock* pBlock;
} SStreamTrigger;
enum { enum {
STREAM_QUEUE__SUCESS = 1, STREAM_QUEUE__SUCESS = 1,
STREAM_QUEUE__FAILED, STREAM_QUEUE__FAILED,
@ -98,6 +104,9 @@ typedef struct {
int8_t status; int8_t status;
} SStreamQueue; } SStreamQueue;
int32_t streamInit();
void streamCleanUp();
SStreamQueue* streamQueueOpen(); SStreamQueue* streamQueueOpen();
void streamQueueClose(SStreamQueue* queue); void streamQueueClose(SStreamQueue* queue);
@ -220,6 +229,11 @@ enum {
TASK_INPUT_TYPE__DATA_BLOCK, TASK_INPUT_TYPE__DATA_BLOCK,
}; };
enum {
TASK_TRIGGER_STATUS__IN_ACTIVE = 1,
TASK_TRIGGER_STATUS__ACTIVE,
};
struct SStreamTask { struct SStreamTask {
int64_t streamId; int64_t streamId;
int32_t taskId; int32_t taskId;
@ -262,8 +276,16 @@ struct SStreamTask {
SStreamQueue* inputQueue; SStreamQueue* inputQueue;
SStreamQueue* outputQueue; SStreamQueue* outputQueue;
// trigger
int8_t triggerStatus;
int64_t triggerParam;
void* timer;
// application storage // application storage
// void* ahandle; // void* ahandle;
// msg handle
SMsgCb* pMsgCb;
}; };
SStreamTask* tNewSStreamTask(int64_t streamId); SStreamTask* tNewSStreamTask(int64_t streamId);
@ -292,6 +314,13 @@ static FORCE_INLINE int32_t streamTaskInput(SStreamTask* pTask, SStreamQueueItem
taosWriteQitem(pTask->inputQueue->queue, pItem); taosWriteQitem(pTask->inputQueue->queue, pItem);
} else if (pItem->type == STREAM_INPUT__CHECKPOINT) { } else if (pItem->type == STREAM_INPUT__CHECKPOINT) {
taosWriteQitem(pTask->inputQueue->queue, pItem); taosWriteQitem(pTask->inputQueue->queue, pItem);
} else if (pItem->type == STREAM_INPUT__TRIGGER) {
taosWriteQitem(pTask->inputQueue->queue, pItem);
}
if (pItem->type != STREAM_INPUT__TRIGGER && pItem->type != STREAM_INPUT__CHECKPOINT && pTask->triggerParam != 0 &&
pTask->triggerStatus == TASK_TRIGGER_STATUS__IN_ACTIVE) {
atomic_store_8(&pTask->triggerStatus, TASK_TRIGGER_STATUS__ACTIVE);
} }
// TODO: back pressure // TODO: back pressure
@ -370,7 +399,8 @@ typedef struct {
int32_t tDecodeStreamDispatchReq(SDecoder* pDecoder, SStreamDispatchReq* pReq); int32_t tDecodeStreamDispatchReq(SDecoder* pDecoder, SStreamDispatchReq* pReq);
int32_t streamTriggerByWrite(SStreamTask* pTask, int32_t vgId, SMsgCb* pMsgCb); int32_t streamLaunchByWrite(SStreamTask* pTask, int32_t vgId, SMsgCb* pMsgCb);
int32_t streamSetupTrigger(SStreamTask* pTask);
int32_t streamTaskRun(SStreamTask* pTask); int32_t streamTaskRun(SStreamTask* pTask);

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

@ -24,7 +24,7 @@ extern "C" {
#include "tdef.h" #include "tdef.h"
#include "tmsgcb.h" #include "tmsgcb.h"
#define SYNC_INDEX_BEGIN 0 #define SYNC_INDEX_BEGIN 0
#define SYNC_INDEX_INVALID -1 #define SYNC_INDEX_INVALID -1
typedef uint64_t SyncNodeId; typedef uint64_t SyncNodeId;
@ -44,14 +44,6 @@ typedef enum {
TAOS_SYNC_STATE_ERROR = 103, TAOS_SYNC_STATE_ERROR = 103,
} ESyncState; } ESyncState;
typedef enum {
TAOS_SYNC_PROPOSE_SUCCESS = 0,
TAOS_SYNC_PROPOSE_NOT_LEADER = 1,
TAOS_SYNC_ONLY_ONE_REPLICA = 2,
TAOS_SYNC_NOT_IN_NEW_CONFIG = 3,
TAOS_SYNC_OTHER_ERROR = 100,
} ESyncProposeCode;
typedef enum { typedef enum {
TAOS_SYNC_FSM_CB_SUCCESS = 0, TAOS_SYNC_FSM_CB_SUCCESS = 0,
TAOS_SYNC_FSM_CB_OTHER_ERROR = 1, TAOS_SYNC_FSM_CB_OTHER_ERROR = 1,

2
include/util/taoserror.h
View File

@ -411,6 +411,8 @@ int32_t* taosGetErrno();
#define TSDB_CODE_SYN_INVALID_MSGTYPE TAOS_DEF_ERROR_CODE(0, 0x090A) #define TSDB_CODE_SYN_INVALID_MSGTYPE TAOS_DEF_ERROR_CODE(0, 0x090A)
#define TSDB_CODE_SYN_NOT_LEADER TAOS_DEF_ERROR_CODE(0, 0x0910) #define TSDB_CODE_SYN_NOT_LEADER TAOS_DEF_ERROR_CODE(0, 0x0910)
#define TSDB_CODE_SYN_ONE_REPLICA TAOS_DEF_ERROR_CODE(0, 0x0911)
#define TSDB_CODE_SYN_NOT_IN_NEW_CONFIG TAOS_DEF_ERROR_CODE(0, 0x0912)
#define TSDB_CODE_SYN_INTERNAL_ERROR TAOS_DEF_ERROR_CODE(0, 0x09FF) #define TSDB_CODE_SYN_INTERNAL_ERROR TAOS_DEF_ERROR_CODE(0, 0x09FF)
// tq // tq

84
source/client/src/tmq.c
View File

@ -992,6 +992,90 @@ CREATE_MSG_FAIL:
return -1; return -1;
} }
bool tmqUpdateEp2(tmq_t* tmq, int32_t epoch, SMqAskEpRsp* pRsp) {
bool set = false;
int32_t topicNumGet = taosArrayGetSize(pRsp->topics);
char vgKey[TSDB_TOPIC_FNAME_LEN + 22];
tscDebug("consumer %ld update ep epoch %d to epoch %d, topic num: %d", tmq->consumerId, tmq->epoch, epoch,
topicNumGet);
SArray* newTopics = taosArrayInit(topicNumGet, sizeof(SMqClientTopic));
if (newTopics == NULL) {
return false;
}
SHashObj* pHash = taosHashInit(64, MurmurHash3_32, false, HASH_NO_LOCK);
if (pHash == NULL) {
taosArrayDestroy(newTopics);
return false;
}
int32_t topicNumCur = taosArrayGetSize(tmq->clientTopics);
for (int32_t i = 0; i < topicNumCur; i++) {
// find old topic
SMqClientTopic* pTopicCur = taosArrayGet(tmq->clientTopics, i);
if (pTopicCur->vgs) {
int32_t vgNumCur = taosArrayGetSize(pTopicCur->vgs);
tscDebug("consumer %ld new vg num: %d", tmq->consumerId, vgNumCur);
if (vgNumCur == 0) break;
for (int32_t j = 0; j < vgNumCur; j++) {
SMqClientVg* pVgCur = taosArrayGet(pTopicCur->vgs, j);
sprintf(vgKey, "%s:%d", pTopicCur->topicName, pVgCur->vgId);
tscDebug("consumer %ld epoch %d vg %d build %s", tmq->consumerId, epoch, pVgCur->vgId, vgKey);
taosHashPut(pHash, vgKey, strlen(vgKey), &pVgCur->currentOffset, sizeof(int64_t));
}
break;
}
}
for (int32_t i = 0; i < topicNumGet; i++) {
SMqClientTopic topic = {0};
SMqSubTopicEp* pTopicEp = taosArrayGet(pRsp->topics, i);
topic.schema = pTopicEp->schema;
taosHashClear(pHash);
topic.topicName = strdup(pTopicEp->topic);
tstrncpy(topic.db, pTopicEp->db, TSDB_DB_FNAME_LEN);
tscDebug("consumer %ld update topic: %s", tmq->consumerId, topic.topicName);
int32_t vgNumGet = taosArrayGetSize(pTopicEp->vgs);
topic.vgs = taosArrayInit(vgNumGet, sizeof(SMqClientVg));
for (int32_t j = 0; j < vgNumGet; j++) {
SMqSubVgEp* pVgEp = taosArrayGet(pTopicEp->vgs, j);
sprintf(vgKey, "%s:%d", topic.topicName, pVgEp->vgId);
int64_t* pOffset = taosHashGet(pHash, vgKey, strlen(vgKey));
int64_t offset = tmq->resetOffsetCfg;
if (pOffset != NULL) {
offset = *pOffset;
}
tscDebug("consumer %ld(epoch %d) offset of vg %d updated to %ld", tmq->consumerId, epoch, pVgEp->vgId, offset);
SMqClientVg clientVg = {
.pollCnt = 0,
.currentOffset = offset,
.vgId = pVgEp->vgId,
.epSet = pVgEp->epSet,
.vgStatus = TMQ_VG_STATUS__IDLE,
.vgSkipCnt = 0,
};
taosArrayPush(topic.vgs, &clientVg);
set = true;
}
taosArrayPush(newTopics, &topic);
}
if (tmq->clientTopics) taosArrayDestroy(tmq->clientTopics);
taosHashCleanup(pHash);
tmq->clientTopics = newTopics;
if (taosArrayGetSize(tmq->clientTopics) == 0)
atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__NO_TOPIC);
else
atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__READY);
atomic_store_32(&tmq->epoch, epoch);
return set;
}
bool tmqUpdateEp(tmq_t* tmq, int32_t epoch, SMqAskEpRsp* pRsp) { bool tmqUpdateEp(tmq_t* tmq, int32_t epoch, SMqAskEpRsp* pRsp) {
/*printf("call update ep %d\n", epoch);*/ /*printf("call update ep %d\n", epoch);*/
bool set = false; bool set = false;

4
source/dnode/mnode/impl/inc/mndDef.h
View File

@ -554,8 +554,8 @@ typedef struct {
SVgObj fixedSinkVg; SVgObj fixedSinkVg;
int64_t smaId; // 0 for unused int64_t smaId; // 0 for unused
int8_t trigger; int8_t trigger;
int32_t triggerParam; int64_t triggerParam;
int64_t waterMark; int64_t watermark;
char* sql; char* sql;
char* physicalPlan; char* physicalPlan;
SArray* tasks; // SArray<SArray<SStreamTask>> SArray* tasks; // SArray<SArray<SStreamTask>>

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

@ -31,7 +31,7 @@ void mndReleaseStream(SMnode *pMnode, SStreamObj *pStream);
SSdbRaw *mndStreamActionEncode(SStreamObj *pStream); SSdbRaw *mndStreamActionEncode(SStreamObj *pStream);
SSdbRow *mndStreamActionDecode(SSdbRaw *pRaw); SSdbRow *mndStreamActionDecode(SSdbRaw *pRaw);
int32_t mndAddStreamToTrans(SMnode *pMnode, SStreamObj *pStream, const char *ast, int8_t triggerType, int64_t watermark, STrans *pTrans); int32_t mndAddStreamToTrans(SMnode *pMnode, SStreamObj *pStream, const char *ast, STrans *pTrans);
#ifdef __cplusplus #ifdef __cplusplus
} }

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

@ -33,8 +33,8 @@ int32_t tEncodeSStreamObj(SEncoder *pEncoder, const SStreamObj *pObj) {
if (tEncodeI8(pEncoder, pObj->status) < 0) return -1; if (tEncodeI8(pEncoder, pObj->status) < 0) return -1;
if (tEncodeI8(pEncoder, pObj->createdBy) < 0) return -1; if (tEncodeI8(pEncoder, pObj->createdBy) < 0) return -1;
if (tEncodeI8(pEncoder, pObj->trigger) < 0) return -1; if (tEncodeI8(pEncoder, pObj->trigger) < 0) return -1;
if (tEncodeI32(pEncoder, pObj->triggerParam) < 0) return -1; if (tEncodeI64(pEncoder, pObj->triggerParam) < 0) return -1;
if (tEncodeI64(pEncoder, pObj->waterMark) < 0) return -1; if (tEncodeI64(pEncoder, pObj->watermark) < 0) return -1;
if (tEncodeI32(pEncoder, pObj->fixedSinkVgId) < 0) return -1; if (tEncodeI32(pEncoder, pObj->fixedSinkVgId) < 0) return -1;
if (tEncodeI64(pEncoder, pObj->smaId) < 0) return -1; if (tEncodeI64(pEncoder, pObj->smaId) < 0) return -1;
if (tEncodeCStr(pEncoder, pObj->sql) < 0) return -1; if (tEncodeCStr(pEncoder, pObj->sql) < 0) return -1;
@ -85,8 +85,8 @@ int32_t tDecodeSStreamObj(SDecoder *pDecoder, SStreamObj *pObj) {
if (tDecodeI8(pDecoder, &pObj->status) < 0) return -1; if (tDecodeI8(pDecoder, &pObj->status) < 0) return -1;
if (tDecodeI8(pDecoder, &pObj->createdBy) < 0) return -1; if (tDecodeI8(pDecoder, &pObj->createdBy) < 0) return -1;
if (tDecodeI8(pDecoder, &pObj->trigger) < 0) return -1; if (tDecodeI8(pDecoder, &pObj->trigger) < 0) return -1;
if (tDecodeI32(pDecoder, &pObj->triggerParam) < 0) return -1; if (tDecodeI64(pDecoder, &pObj->triggerParam) < 0) return -1;
if (tDecodeI64(pDecoder, &pObj->waterMark) < 0) return -1; if (tDecodeI64(pDecoder, &pObj->watermark) < 0) return -1;
if (tDecodeI32(pDecoder, &pObj->fixedSinkVgId) < 0) return -1; if (tDecodeI32(pDecoder, &pObj->fixedSinkVgId) < 0) return -1;
if (tDecodeI64(pDecoder, &pObj->smaId) < 0) return -1; if (tDecodeI64(pDecoder, &pObj->smaId) < 0) return -1;
if (tDecodeCStrAlloc(pDecoder, &pObj->sql) < 0) return -1; if (tDecodeCStrAlloc(pDecoder, &pObj->sql) < 0) return -1;

15
source/dnode/mnode/impl/src/mndMain.c
View File

@ -380,17 +380,19 @@ void mndStop(SMnode *pMnode) {
int32_t mndProcessSyncMsg(SRpcMsg *pMsg) { int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
SMnode *pMnode = pMsg->info.node; SMnode *pMnode = pMsg->info.node;
SSyncMgmt *pMgmt = &pMnode->syncMgmt; SSyncMgmt *pMgmt = &pMnode->syncMgmt;
int32_t code = TAOS_SYNC_OTHER_ERROR; int32_t code = 0;
if (!syncEnvIsStart()) { if (!syncEnvIsStart()) {
mError("failed to process sync msg:%p type:%s since syncEnv stop", pMsg, TMSG_INFO(pMsg->msgType)); mError("failed to process sync msg:%p type:%s since syncEnv stop", pMsg, TMSG_INFO(pMsg->msgType));
return TAOS_SYNC_OTHER_ERROR; terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1;
} }
SSyncNode *pSyncNode = syncNodeAcquire(pMgmt->sync); SSyncNode *pSyncNode = syncNodeAcquire(pMgmt->sync);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
mError("failed to process sync msg:%p type:%s since syncNode is null", pMsg, TMSG_INFO(pMsg->msgType)); mError("failed to process sync msg:%p type:%s since syncNode is null", pMsg, TMSG_INFO(pMsg->msgType));
return TAOS_SYNC_OTHER_ERROR; terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1;
} }
char logBuf[512] = {0}; char logBuf[512] = {0};
@ -451,7 +453,7 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
tmsgSendRsp(&rsp); tmsgSendRsp(&rsp);
} else { } else {
mError("failed to process msg:%p since invalid type:%s", pMsg, TMSG_INFO(pMsg->msgType)); mError("failed to process msg:%p since invalid type:%s", pMsg, TMSG_INFO(pMsg->msgType));
code = TAOS_SYNC_OTHER_ERROR; code = -1;
} }
} else { } else {
if (pMsg->msgType == TDMT_SYNC_TIMEOUT) { if (pMsg->msgType == TDMT_SYNC_TIMEOUT) {
@ -492,10 +494,13 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
tmsgSendRsp(&rsp); tmsgSendRsp(&rsp);
} else { } else {
mError("failed to process msg:%p since invalid type:%s", pMsg, TMSG_INFO(pMsg->msgType)); mError("failed to process msg:%p since invalid type:%s", pMsg, TMSG_INFO(pMsg->msgType));
code = TAOS_SYNC_OTHER_ERROR; code = -1;
} }
} }
if (code != 0) {
terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
}
return code; return code;
} }

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

@ -387,6 +387,9 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
// input // input
pFinalTask->inputType = TASK_INPUT_TYPE__DATA_BLOCK; pFinalTask->inputType = TASK_INPUT_TYPE__DATA_BLOCK;
// trigger
pFinalTask->triggerParam = pStream->triggerParam;
// dispatch // dispatch
if (mndAddDispatcherToInnerTask(pMnode, pTrans, pStream, pFinalTask) < 0) { if (mndAddDispatcherToInnerTask(pMnode, pTrans, pStream, pFinalTask) < 0) {
qDestroyQueryPlan(pPlan); qDestroyQueryPlan(pPlan);

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

@ -561,6 +561,8 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
streamObj.sql = pCreate->sql; streamObj.sql = pCreate->sql;
streamObj.createdBy = STREAM_CREATED_BY__SMA; streamObj.createdBy = STREAM_CREATED_BY__SMA;
streamObj.smaId = smaObj.uid; streamObj.smaId = smaObj.uid;
streamObj.watermark = 0;
streamObj.trigger = STREAM_TRIGGER_AT_ONCE;
if (mndAllocSmaVgroup(pMnode, pDb, &streamObj.fixedSinkVg) != 0) { if (mndAllocSmaVgroup(pMnode, pDb, &streamObj.fixedSinkVg) != 0) {
mError("sma:%s, failed to create since %s", smaObj.name, terrstr()); mError("sma:%s, failed to create since %s", smaObj.name, terrstr());
@ -583,7 +585,7 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
if (mndSetUpdateSmaStbCommitLogs(pMnode, pTrans, pStb) != 0) goto _OVER; if (mndSetUpdateSmaStbCommitLogs(pMnode, pTrans, pStb) != 0) goto _OVER;
// if (mndSetCreateSmaRedoActions(pMnode, pTrans, pDb, &smaObj) != 0) goto _OVER; // if (mndSetCreateSmaRedoActions(pMnode, pTrans, pDb, &smaObj) != 0) goto _OVER;
if (mndSetCreateSmaVgroupRedoActions(pMnode, pTrans, pDb, &streamObj.fixedSinkVg, &smaObj) != 0) goto _OVER; if (mndSetCreateSmaVgroupRedoActions(pMnode, pTrans, pDb, &streamObj.fixedSinkVg, &smaObj) != 0) goto _OVER;
if (mndAddStreamToTrans(pMnode, &streamObj, pCreate->ast, STREAM_TRIGGER_AT_ONCE, 0, pTrans) != 0) goto _OVER; if (mndAddStreamToTrans(pMnode, &streamObj, pCreate->ast, pTrans) != 0) goto _OVER;
if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER; if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
code = 0; code = 0;

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

@ -228,23 +228,22 @@ static int32_t mndStreamGetPlanString(const char *ast, int8_t triggerType, int64
.pAstRoot = pAst, .pAstRoot = pAst,
.topicQuery = false, .topicQuery = false,
.streamQuery = true, .streamQuery = true,
.triggerType = triggerType, .triggerType = triggerType == STREAM_TRIGGER_MAX_DELAY ? STREAM_TRIGGER_WINDOW_CLOSE : triggerType,
.watermark = watermark, .watermark = watermark,
}; };
code = qCreateQueryPlan(&cxt, &pPlan, NULL); code = qCreateQueryPlan(&cxt, &pPlan, NULL);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = nodesNodeToString((SNode*)pPlan, false, pStr, NULL); code = nodesNodeToString((SNode *)pPlan, false, pStr, NULL);
} }
nodesDestroyNode(pAst); nodesDestroyNode(pAst);
nodesDestroyNode((SNode*)pPlan); nodesDestroyNode((SNode *)pPlan);
terrno = code; terrno = code;
return code; return code;
} }
int32_t mndAddStreamToTrans(SMnode *pMnode, SStreamObj *pStream, const char *ast, int8_t triggerType, int64_t watermark, int32_t mndAddStreamToTrans(SMnode *pMnode, SStreamObj *pStream, const char *ast, STrans *pTrans) {
STrans *pTrans) {
SNode *pAst = NULL; SNode *pAst = NULL;
if (nodesStringToNode(ast, &pAst) < 0) { if (nodesStringToNode(ast, &pAst) < 0) {
@ -258,7 +257,6 @@ int32_t mndAddStreamToTrans(SMnode *pMnode, SStreamObj *pStream, const char *ast
// free // free
nodesDestroyNode(pAst); nodesDestroyNode(pAst);
#if 0 #if 0
printf("|"); printf("|");
for (int i = 0; i < pStream->outputSchema.nCols; i++) { for (int i = 0; i < pStream->outputSchema.nCols; i++) {
@ -268,7 +266,7 @@ int32_t mndAddStreamToTrans(SMnode *pMnode, SStreamObj *pStream, const char *ast
#endif #endif
if (TSDB_CODE_SUCCESS != mndStreamGetPlanString(ast, triggerType, watermark, &pStream->physicalPlan)) { if (TSDB_CODE_SUCCESS != mndStreamGetPlanString(ast, pStream->trigger, pStream->watermark, &pStream->physicalPlan)) {
mError("topic:%s, failed to get plan since %s", pStream->name, terrstr()); mError("topic:%s, failed to get plan since %s", pStream->name, terrstr());
return -1; return -1;
} }
@ -391,7 +389,8 @@ static int32_t mndCreateStream(SMnode *pMnode, SRpcMsg *pReq, SCMCreateStreamReq
streamObj.smaId = 0; streamObj.smaId = 0;
/*streamObj.physicalPlan = "";*/ /*streamObj.physicalPlan = "";*/
streamObj.trigger = pCreate->triggerType; streamObj.trigger = pCreate->triggerType;
streamObj.waterMark = pCreate->watermark; streamObj.watermark = pCreate->watermark;
streamObj.triggerParam = pCreate->maxDelay;
if (streamObj.targetSTbName[0]) { if (streamObj.targetSTbName[0]) {
pDb = mndAcquireDbByStb(pMnode, streamObj.targetSTbName); pDb = mndAcquireDbByStb(pMnode, streamObj.targetSTbName);
@ -409,7 +408,7 @@ static int32_t mndCreateStream(SMnode *pMnode, SRpcMsg *pReq, SCMCreateStreamReq
} }
mDebug("trans:%d, used to create stream:%s", pTrans->id, pCreate->name); mDebug("trans:%d, used to create stream:%s", pTrans->id, pCreate->name);
if (mndAddStreamToTrans(pMnode, &streamObj, pCreate->ast, pCreate->triggerType, pCreate->watermark, pTrans) != 0) { if (mndAddStreamToTrans(pMnode, &streamObj, pCreate->ast, pTrans) != 0) {
mError("trans:%d, failed to add stream since %s", pTrans->id, terrstr()); mError("trans:%d, failed to add stream since %s", pTrans->id, terrstr());
mndTransDrop(pTrans); mndTransDrop(pTrans);
return -1; return -1;
@ -566,7 +565,7 @@ static int32_t mndRetrieveStream(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
colDataAppend(pColInfo, numOfRows, (const char *)&pStream->targetSTbName, true); colDataAppend(pColInfo, numOfRows, (const char *)&pStream->targetSTbName, true);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)&pStream->waterMark, false); colDataAppend(pColInfo, numOfRows, (const char *)&pStream->watermark, false);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)&pStream->trigger, false); colDataAppend(pColInfo, numOfRows, (const char *)&pStream->trigger, false);

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

@ -234,9 +234,9 @@ int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw, int32_t transId) {
int32_t code = syncPropose(pMgmt->sync, &rsp, isWeak); int32_t code = syncPropose(pMgmt->sync, &rsp, isWeak);
if (code == 0) { if (code == 0) {
tsem_wait(&pMgmt->syncSem); tsem_wait(&pMgmt->syncSem);
} else if (code == TAOS_SYNC_PROPOSE_NOT_LEADER) { } else if (code == -1 && terrno == TSDB_CODE_SYN_NOT_LEADER) {
terrno = TSDB_CODE_APP_NOT_READY; terrno = TSDB_CODE_APP_NOT_READY;
} else if (code == TAOS_SYNC_OTHER_ERROR) { } else if (code == -1 && terrno == TSDB_CODE_SYN_INTERNAL_ERROR) {
terrno = TSDB_CODE_SYN_INTERNAL_ERROR; terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
} else { } else {
terrno = TSDB_CODE_APP_ERROR; terrno = TSDB_CODE_APP_ERROR;
@ -257,13 +257,13 @@ void mndSyncStart(SMnode *pMnode) {
syncStart(pMgmt->sync); syncStart(pMgmt->sync);
mDebug("mnode sync started, id:%" PRId64 " standby:%d", pMgmt->sync, pMgmt->standby); mDebug("mnode sync started, id:%" PRId64 " standby:%d", pMgmt->sync, pMgmt->standby);
/* /*
if (pMgmt->standby) { if (pMgmt->standby) {
syncStartStandBy(pMgmt->sync); syncStartStandBy(pMgmt->sync);
} else { } else {
syncStart(pMgmt->sync); syncStart(pMgmt->sync);
} }
*/ */
} }
void mndSyncStop(SMnode *pMnode) {} void mndSyncStop(SMnode *pMnode) {}

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

@ -375,6 +375,8 @@ int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) {
if (pTask->inputQueue == NULL || pTask->outputQueue == NULL) goto FAIL; if (pTask->inputQueue == NULL || pTask->outputQueue == NULL) goto FAIL;
pTask->pMsgCb = &pTq->pVnode->msgCb;
// exec // exec
if (pTask->execType != TASK_EXEC__NONE) { if (pTask->execType != TASK_EXEC__NONE) {
// expand runners // expand runners
@ -406,9 +408,12 @@ int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) {
tdGetSTSChemaFromSSChema(&pTask->tbSink.pSchemaWrapper->pSchema, pTask->tbSink.pSchemaWrapper->nCols); tdGetSTSChemaFromSSChema(&pTask->tbSink.pSchemaWrapper->pSchema, pTask->tbSink.pSchemaWrapper->nCols);
ASSERT(pTask->tbSink.pTSchema); ASSERT(pTask->tbSink.pTSchema);
} }
streamSetupTrigger(pTask);
tqInfo("deploy stream task id %d child id %d on vg %d", pTask->taskId, pTask->childId, pTq->pVnode->config.vgId); tqInfo("deploy stream task id %d child id %d on vg %d", pTask->taskId, pTask->childId, pTq->pVnode->config.vgId);
taosHashPut(pTq->pStreamTasks, &pTask->taskId, sizeof(int32_t), pTask, sizeof(SStreamTask)); taosHashPut(pTq->pStreamTasks, &pTask->taskId, sizeof(int32_t), &pTask, sizeof(void*));
return 0; return 0;
FAIL: FAIL:
@ -431,7 +436,7 @@ int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* pReq) {
while (1) { while (1) {
pIter = taosHashIterate(pTq->pStreamTasks, pIter); pIter = taosHashIterate(pTq->pStreamTasks, pIter);
if (pIter == NULL) break; if (pIter == NULL) break;
SStreamTask* pTask = (SStreamTask*)pIter; SStreamTask* pTask = *(SStreamTask**)pIter;
if (pTask->inputType != STREAM_INPUT__DATA_SUBMIT) continue; if (pTask->inputType != STREAM_INPUT__DATA_SUBMIT) continue;
if (!failed) { if (!failed) {
@ -439,7 +444,7 @@ int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* pReq) {
continue; continue;
} }
if (streamTriggerByWrite(pTask, pTq->pVnode->config.vgId, &pTq->pVnode->msgCb) < 0) { if (streamLaunchByWrite(pTask, pTq->pVnode->config.vgId, &pTq->pVnode->msgCb) < 0) {
continue; continue;
} }
} else { } else {
@ -459,7 +464,7 @@ int32_t tqProcessTaskRunReq(STQ* pTq, SRpcMsg* pMsg) {
// //
SStreamTaskRunReq* pReq = pMsg->pCont; SStreamTaskRunReq* pReq = pMsg->pCont;
int32_t taskId = pReq->taskId; int32_t taskId = pReq->taskId;
SStreamTask* pTask = taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t)); SStreamTask* pTask = *(SStreamTask**)taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
streamTaskProcessRunReq(pTask, &pTq->pVnode->msgCb); streamTaskProcessRunReq(pTask, &pTq->pVnode->msgCb);
return 0; return 0;
} }
@ -473,7 +478,7 @@ int32_t tqProcessTaskDispatchReq(STQ* pTq, SRpcMsg* pMsg) {
tDecoderInit(&decoder, msgBody, msgLen); tDecoderInit(&decoder, msgBody, msgLen);
tDecodeStreamDispatchReq(&decoder, &req); tDecodeStreamDispatchReq(&decoder, &req);
int32_t taskId = req.taskId; int32_t taskId = req.taskId;
SStreamTask* pTask = taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t)); SStreamTask* pTask = *(SStreamTask**)taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
SRpcMsg rsp = { SRpcMsg rsp = {
.info = pMsg->info, .info = pMsg->info,
.code = 0, .code = 0,
@ -485,7 +490,7 @@ int32_t tqProcessTaskDispatchReq(STQ* pTq, SRpcMsg* pMsg) {
int32_t tqProcessTaskRecoverReq(STQ* pTq, SRpcMsg* pMsg) { int32_t tqProcessTaskRecoverReq(STQ* pTq, SRpcMsg* pMsg) {
SStreamTaskRecoverReq* pReq = pMsg->pCont; SStreamTaskRecoverReq* pReq = pMsg->pCont;
int32_t taskId = pReq->taskId; int32_t taskId = pReq->taskId;
SStreamTask* pTask = taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t)); SStreamTask* pTask = *(SStreamTask**)taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
streamProcessRecoverReq(pTask, &pTq->pVnode->msgCb, pReq, pMsg); streamProcessRecoverReq(pTask, &pTq->pVnode->msgCb, pReq, pMsg);
return 0; return 0;
} }
@ -493,7 +498,7 @@ int32_t tqProcessTaskRecoverReq(STQ* pTq, SRpcMsg* pMsg) {
int32_t tqProcessTaskDispatchRsp(STQ* pTq, SRpcMsg* pMsg) { int32_t tqProcessTaskDispatchRsp(STQ* pTq, SRpcMsg* pMsg) {
SStreamDispatchRsp* pRsp = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)); SStreamDispatchRsp* pRsp = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
int32_t taskId = pRsp->taskId; int32_t taskId = pRsp->taskId;
SStreamTask* pTask = taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t)); SStreamTask* pTask = *(SStreamTask**)taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
streamProcessDispatchRsp(pTask, &pTq->pVnode->msgCb, pRsp); streamProcessDispatchRsp(pTask, &pTq->pVnode->msgCb, pRsp);
return 0; return 0;
} }
@ -501,7 +506,7 @@ int32_t tqProcessTaskDispatchRsp(STQ* pTq, SRpcMsg* pMsg) {
int32_t tqProcessTaskRecoverRsp(STQ* pTq, SRpcMsg* pMsg) { int32_t tqProcessTaskRecoverRsp(STQ* pTq, SRpcMsg* pMsg) {
SStreamTaskRecoverRsp* pRsp = pMsg->pCont; SStreamTaskRecoverRsp* pRsp = pMsg->pCont;
int32_t taskId = pRsp->taskId; int32_t taskId = pRsp->taskId;
SStreamTask* pTask = taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t)); SStreamTask* pTask = *(SStreamTask**)taosHashGet(pTq->pStreamTasks, &taskId, sizeof(int32_t));
streamProcessRecoverRsp(pTask, pRsp); streamProcessRecoverRsp(pTask, pRsp);
return 0; return 0;
} }

9
source/dnode/vnode/src/vnd/vnodeSvr.c
View File

@ -296,7 +296,7 @@ void vnodeUpdateMetaRsp(SVnode *pVnode, STableMetaRsp *pMetaRsp) {
} }
int32_t vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) { int32_t vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
int32_t ret = TAOS_SYNC_OTHER_ERROR; int32_t ret = 0;
if (syncEnvIsStart()) { if (syncEnvIsStart()) {
SSyncNode *pSyncNode = syncNodeAcquire(pVnode->sync); SSyncNode *pSyncNode = syncNodeAcquire(pVnode->sync);
@ -381,15 +381,18 @@ int32_t vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
tmsgSendRsp(&rsp); tmsgSendRsp(&rsp);
} else { } else {
vError("==vnodeProcessSyncReq== error msg type:%d", pRpcMsg->msgType); vError("==vnodeProcessSyncReq== error msg type:%d", pRpcMsg->msgType);
ret = TAOS_SYNC_OTHER_ERROR; ret = -1;
} }
syncNodeRelease(pSyncNode); syncNodeRelease(pSyncNode);
} else { } else {
vError("==vnodeProcessSyncReq== error syncEnv stop"); vError("==vnodeProcessSyncReq== error syncEnv stop");
ret = TAOS_SYNC_OTHER_ERROR; ret = -1;
} }
if (ret != 0) {
terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
}
return ret; return ret;
} }

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

@ -98,7 +98,8 @@ void vnodeProposeMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
if (code == 0) { if (code == 0) {
vnodeAccumBlockMsg(pVnode, pMsg->msgType); vnodeAccumBlockMsg(pVnode, pMsg->msgType);
} else if (code == TAOS_SYNC_PROPOSE_NOT_LEADER) {
} else if (code == -1 && terrno == TSDB_CODE_SYN_NOT_LEADER) {
SEpSet newEpSet = {0}; SEpSet newEpSet = {0};
syncGetEpSet(pVnode->sync, &newEpSet); syncGetEpSet(pVnode->sync, &newEpSet);
SEp *pEp = &newEpSet.eps[newEpSet.inUse]; SEp *pEp = &newEpSet.eps[newEpSet.inUse];
@ -247,29 +248,17 @@ static void vnodeSyncRollBackMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta
syncRpcMsgLog2(logBuf, (SRpcMsg *)pMsg); syncRpcMsgLog2(logBuf, (SRpcMsg *)pMsg);
} }
int32_t vnodeSnapshotStartRead(struct SSyncFSM *pFsm, void **ppReader) { int32_t vnodeSnapshotStartRead(struct SSyncFSM *pFsm, void **ppReader) { return 0; }
return 0;
}
int32_t vnodeSnapshotStopRead(struct SSyncFSM *pFsm, void *pReader) { int32_t vnodeSnapshotStopRead(struct SSyncFSM *pFsm, void *pReader) { return 0; }
return 0;
}
int32_t vnodeSnapshotDoRead(struct SSyncFSM *pFsm, void *pReader, void **ppBuf, int32_t *len) { int32_t vnodeSnapshotDoRead(struct SSyncFSM *pFsm, void *pReader, void **ppBuf, int32_t *len) { return 0; }
return 0;
}
int32_t vnodeSnapshotStartWrite(struct SSyncFSM *pFsm, void **ppWriter) { int32_t vnodeSnapshotStartWrite(struct SSyncFSM *pFsm, void **ppWriter) { return 0; }
return 0;
}
int32_t vnodeSnapshotStopWrite(struct SSyncFSM *pFsm, void *pWriter, bool isApply) { int32_t vnodeSnapshotStopWrite(struct SSyncFSM *pFsm, void *pWriter, bool isApply) { return 0; }
return 0;
}
int32_t vnodeSnapshotDoWrite(struct SSyncFSM *pFsm, void *pWriter, void *pBuf, int32_t len) { int32_t vnodeSnapshotDoWrite(struct SSyncFSM *pFsm, void *pWriter, void *pBuf, int32_t len) { return 0; }
return 0;
}
static SSyncFSM *vnodeSyncMakeFsm(SVnode *pVnode) { static SSyncFSM *vnodeSyncMakeFsm(SVnode *pVnode) {
SSyncFSM *pFsm = taosMemoryCalloc(1, sizeof(SSyncFSM)); SSyncFSM *pFsm = taosMemoryCalloc(1, sizeof(SSyncFSM));

2
source/libs/catalog/src/ctgCache.c
View File

@ -1899,7 +1899,7 @@ _return:
void ctgUpdateThreadUnexpectedStopped(void) { void ctgUpdateThreadUnexpectedStopped(void) {
if (CTG_IS_LOCKED(&gCtgMgmt.lock) > 0) CTG_UNLOCK(CTG_READ, &gCtgMgmt.lock); if (!atomic_load_8((int8_t*)&gCtgMgmt.exit) && CTG_IS_LOCKED(&gCtgMgmt.lock) > 0) CTG_UNLOCK(CTG_READ, &gCtgMgmt.lock);
} }
void ctgCleanupCacheQueue(void) { void ctgCleanupCacheQueue(void) {

12
source/libs/executor/inc/executorimpl.h
View File

@ -365,7 +365,9 @@ typedef struct SCatchSupporter {
} SCatchSupporter; } SCatchSupporter;
typedef struct SStreamAggSupporter { typedef struct SStreamAggSupporter {
SArray* pResultRows; SHashObj* pResultRows;
SArray* pCurWins;
int32_t valueSize;
int32_t keySize; int32_t keySize;
char* pKeyBuf; // window key buffer char* pKeyBuf; // window key buffer
SDiskbasedBuf* pResultBuf; // query result buffer based on blocked-wised disk file SDiskbasedBuf* pResultBuf; // query result buffer based on blocked-wised disk file
@ -391,7 +393,9 @@ typedef struct SStreamBlockScanInfo {
void* streamBlockReader;// stream block reader handle void* streamBlockReader;// stream block reader handle
SArray* pColMatchInfo; // SArray* pColMatchInfo; //
SNode* pCondition; SNode* pCondition;
int32_t tsArrayIndex;
SArray* tsArray; SArray* tsArray;
uint64_t groupId;
SUpdateInfo* pUpdateInfo; SUpdateInfo* pUpdateInfo;
SExprInfo* pPseudoExpr; SExprInfo* pPseudoExpr;
@ -582,6 +586,7 @@ typedef struct SPartitionOperatorInfo {
int32_t* columnOffset; // start position for each column data int32_t* columnOffset; // start position for each column data
void* pGroupIter; // group iterator void* pGroupIter; // group iterator
int32_t pageIndex; // page index of current group int32_t pageIndex; // page index of current group
SSDataBlock* pUpdateRes;
} SPartitionOperatorInfo; } SPartitionOperatorInfo;
typedef struct SWindowRowsSup { typedef struct SWindowRowsSup {
@ -896,9 +901,9 @@ int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimary
__block_search_fn_t searchFn, STableQueryInfo* item, int32_t order); __block_search_fn_t searchFn, STableQueryInfo* item, int32_t order);
int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order); int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order);
int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, const char* pKey, int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, const char* pKey,
SqlFunctionCtx* pCtx, int32_t numOfOutput, size_t size); SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t size);
SResultRow* getNewResultRow(SDiskbasedBuf* pResultBuf, int64_t tableGroupId, int32_t interBufSize); SResultRow* getNewResultRow(SDiskbasedBuf* pResultBuf, int64_t tableGroupId, int32_t interBufSize);
SResultWindowInfo* getSessionTimeWindow(SArray* pWinInfos, TSKEY ts, int64_t gap, int32_t* pIndex); SResultWindowInfo* getSessionTimeWindow(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId, int64_t gap, int32_t* pIndex);
int32_t updateSessionWindowInfo(SResultWindowInfo* pWinInfo, TSKEY* pTs, int32_t rows, int32_t updateSessionWindowInfo(SResultWindowInfo* pWinInfo, TSKEY* pTs, int32_t rows,
int32_t start, int64_t gap, SHashObj* pStDeleted); int32_t start, int64_t gap, SHashObj* pStDeleted);
bool functionNeedToExecute(SqlFunctionCtx* pCtx); bool functionNeedToExecute(SqlFunctionCtx* pCtx);
@ -907,6 +912,7 @@ int32_t compareTimeWindow(const void* p1, const void* p2, const void* param);
int32_t finalizeResultRowIntoResultDataBlock(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPosition, int32_t finalizeResultRowIntoResultDataBlock(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPosition,
SqlFunctionCtx* pCtx, SExprInfo* pExprInfo, int32_t numOfExprs, const int32_t* rowCellOffset, SqlFunctionCtx* pCtx, SExprInfo* pExprInfo, int32_t numOfExprs, const int32_t* rowCellOffset,
SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo); SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);
void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_t tsColIndex);
#ifdef __cplusplus #ifdef __cplusplus
} }

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

@ -19,7 +19,8 @@
#include "tdatablock.h" #include "tdatablock.h"
#include "vnode.h" #include "vnode.h"
static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t numOfBlocks, int32_t type, bool assignUid, char* id) { static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t numOfBlocks, int32_t type, bool assignUid,
char* id) {
ASSERT(pOperator != NULL); ASSERT(pOperator != NULL);
if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) { if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
if (pOperator->numOfDownstream == 0) { if (pOperator->numOfDownstream == 0) {
@ -43,6 +44,7 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu
if (pInfo->blockType == 0) { if (pInfo->blockType == 0) {
pInfo->blockType = type; pInfo->blockType = type;
} else if (pInfo->blockType != type) { } else if (pInfo->blockType != type) {
ASSERT(0);
return TSDB_CODE_QRY_APP_ERROR; return TSDB_CODE_QRY_APP_ERROR;
} }
@ -51,7 +53,7 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu
qError("submit msg messed up when initing stream block, %s" PRIx64, id); qError("submit msg messed up when initing stream block, %s" PRIx64, id);
return TSDB_CODE_QRY_APP_ERROR; return TSDB_CODE_QRY_APP_ERROR;
} }
} else { } else if (type == STREAM_DATA_TYPE_SSDATA_BLOCK) {
for (int32_t i = 0; i < numOfBlocks; ++i) { for (int32_t i = 0; i < numOfBlocks; ++i) {
SSDataBlock* pDataBlock = &((SSDataBlock*)input)[i]; SSDataBlock* pDataBlock = &((SSDataBlock*)input)[i];
@ -62,6 +64,8 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu
taosArrayAddAll(p->pDataBlock, pDataBlock->pDataBlock); taosArrayAddAll(p->pDataBlock, pDataBlock->pDataBlock);
taosArrayPush(pInfo->pBlockLists, &p); taosArrayPush(pInfo->pBlockLists, &p);
} }
} else {
ASSERT(0);
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -83,7 +87,8 @@ int32_t qSetMultiStreamInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numO
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo; SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
int32_t code = doSetStreamBlock(pTaskInfo->pRoot, (void**)pBlocks, numOfBlocks, type, assignUid, GET_TASKID(pTaskInfo)); int32_t code =
doSetStreamBlock(pTaskInfo->pRoot, (void**)pBlocks, numOfBlocks, type, assignUid, GET_TASKID(pTaskInfo));
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
qError("%s failed to set the stream block data", GET_TASKID(pTaskInfo)); qError("%s failed to set the stream block data", GET_TASKID(pTaskInfo));
} else { } else {
@ -162,7 +167,7 @@ int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, const SArray* tableIdList, bo
pInfo = pInfo->pDownstream[0]; pInfo = pInfo->pDownstream[0];
} }
int32_t code = 0; int32_t code = 0;
SStreamBlockScanInfo* pScanInfo = pInfo->info; SStreamBlockScanInfo* pScanInfo = pInfo->info;
if (isAdd) { // add new table id if (isAdd) { // add new table id
SArray* qa = filterQualifiedChildTables(pScanInfo, tableIdList); SArray* qa = filterQualifiedChildTables(pScanInfo, tableIdList);
@ -178,9 +183,10 @@ int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, const SArray* tableIdList, bo
return code; return code;
} }
int32_t qGetQueriedTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, char* tableName, int32_t* sversion, int32_t* tversion) { int32_t qGetQueriedTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, char* tableName, int32_t* sversion,
int32_t* tversion) {
ASSERT(tinfo != NULL && dbName != NULL && tableName != NULL); ASSERT(tinfo != NULL && dbName != NULL && tableName != NULL);
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*) tinfo; SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
*sversion = pTaskInfo->schemaVer.sversion; *sversion = pTaskInfo->schemaVer.sversion;
*tversion = pTaskInfo->schemaVer.tversion; *tversion = pTaskInfo->schemaVer.tversion;
@ -196,4 +202,4 @@ int32_t qGetQueriedTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, char* tab
} }
return 0; return 0;
} }

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

@ -4814,6 +4814,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
.calTrigger = pIntervalPhyNode->window.triggerType, .calTrigger = pIntervalPhyNode->window.triggerType,
.maxTs = INT64_MIN, .maxTs = INT64_MIN,
}; };
ASSERT(as.calTrigger != STREAM_TRIGGER_MAX_DELAY);
int32_t tsSlotId = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId; int32_t tsSlotId = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId;
bool isStream = (QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL == type); bool isStream = (QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL == type);
@ -5498,14 +5499,16 @@ int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SExplainExecInfo
} }
int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, const char* pKey, SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, const char* pKey, SqlFunctionCtx* pCtx, int32_t numOfOutput,
size_t size) { int32_t size) {
pSup->resultRowSize = getResultRowSize(pCtx, numOfOutput); pSup->resultRowSize = getResultRowSize(pCtx, numOfOutput);
pSup->keySize = sizeof(int64_t) + sizeof(TSKEY); pSup->keySize = sizeof(int64_t) + sizeof(TSKEY);
pSup->pKeyBuf = taosMemoryCalloc(1, pSup->keySize); pSup->pKeyBuf = taosMemoryCalloc(1, pSup->keySize);
pSup->pResultRows = taosArrayInit(1024, size); _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
pSup->pResultRows = taosHashInit(1024, hashFn, false, HASH_NO_LOCK);
if (pSup->pKeyBuf == NULL || pSup->pResultRows == NULL) { if (pSup->pKeyBuf == NULL || pSup->pResultRows == NULL) {
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;
} }
pSup->valueSize = size;
pSup->pScanWindow = taosArrayInit(4, sizeof(STimeWindow)); pSup->pScanWindow = taosArrayInit(4, sizeof(STimeWindow));

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

@ -716,7 +716,7 @@ static bool prepareDataScan(SStreamBlockScanInfo* pInfo) {
int64_t gap = pInfo->sessionSup.gap; int64_t gap = pInfo->sessionSup.gap;
int32_t winIndex = 0; int32_t winIndex = 0;
SResultWindowInfo* pCurWin = SResultWindowInfo* pCurWin =
getSessionTimeWindow(pAggSup->pResultRows, tsCols[pInfo->updateResIndex], gap, &winIndex); getSessionTimeWindow(pAggSup, tsCols[pInfo->updateResIndex], pSDB->info.groupId, gap, &winIndex);
win = pCurWin->win; win = pCurWin->win;
pInfo->updateResIndex += pInfo->updateResIndex +=
updateSessionWindowInfo(pCurWin, tsCols, pSDB->info.rows, pInfo->updateResIndex, gap, NULL); updateSessionWindowInfo(pCurWin, tsCols, pSDB->info.rows, pInfo->updateResIndex, gap, NULL);
@ -750,16 +750,103 @@ static bool prepareDataScan(SStreamBlockScanInfo* pInfo) {
return true; return true;
} }
static void copyOneRow(SSDataBlock* dest, SSDataBlock* source, int32_t sourceRowId) {
for (int32_t j = 0; j < source->info.numOfCols; j++) {
SColumnInfoData* pDestCol = (SColumnInfoData*)taosArrayGet(dest->pDataBlock, j);
SColumnInfoData* pSourceCol = (SColumnInfoData*)taosArrayGet(source->pDataBlock, j);
if (colDataIsNull_s(pSourceCol, sourceRowId)) {
colDataAppendNULL(pDestCol, dest->info.rows);
} else {
colDataAppend(pDestCol, dest->info.rows, colDataGetData(pSourceCol, sourceRowId), false);
}
}
dest->info.rows++;
}
static uint64_t getGroupId(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t rowId) {
uint64_t* groupId = taosHashGet(pOperator->pTaskInfo->tableqinfoList.map, &pBlock->info.uid, sizeof(int64_t));
if (groupId) {
return *groupId;
}
return 0;
/* Todo(liuyao) for partition by column
recordNewGroupKeys(pTableScanInfo->pGroupCols, pTableScanInfo->pGroupColVals, pBlock, rowId);
int32_t len = buildGroupKeys(pTableScanInfo->keyBuf, pTableScanInfo->pGroupColVals);
uint64_t resId = 0;
uint64_t* groupId = taosHashGet(pTableScanInfo->pGroupSet, pTableScanInfo->keyBuf, len);
if (groupId) {
return *groupId;
} else if (len != 0) {
resId = calcGroupId(pTableScanInfo->keyBuf, len);
taosHashPut(pTableScanInfo->pGroupSet, pTableScanInfo->keyBuf, len, &resId, sizeof(uint64_t));
}
return resId;
*/
}
static SSDataBlock* doDataScan(SStreamBlockScanInfo* pInfo) { static SSDataBlock* doDataScan(SStreamBlockScanInfo* pInfo) {
SSDataBlock* pResult = NULL; while (1) {
pResult = doTableScan(pInfo->pOperatorDumy); SSDataBlock* pResult = NULL;
if (pResult == NULL) { pResult = doTableScan(pInfo->pOperatorDumy);
if (prepareDataScan(pInfo)) { if (pResult == NULL) {
// scan next window data if (prepareDataScan(pInfo)) {
pResult = doTableScan(pInfo->pOperatorDumy); // scan next window data
pResult = doTableScan(pInfo->pOperatorDumy);
}
}
if (!pResult) {
return NULL;
}
if (pResult->info.groupId == pInfo->groupId) {
return pResult;
}
}
/* Todo(liuyao) for partition by column
SSDataBlock* pBlock = createOneDataBlock(pResult, true);
blockDataCleanup(pResult);
for (int32_t i = 0; i < pBlock->info.rows; i++) {
uint64_t id = getGroupId(pInfo->pOperatorDumy, pBlock, i);
if (id == pInfo->groupId) {
copyOneRow(pResult, pBlock, i);
} }
} }
return pResult; return pResult;
*/
}
static void setUpdateData(SStreamBlockScanInfo* pInfo, SSDataBlock* pBlock, SSDataBlock* pUpdateBlock) {
blockDataCleanup(pUpdateBlock);
int32_t size = taosArrayGetSize(pInfo->tsArray);
if (pInfo->tsArrayIndex < size) {
SColumnInfoData* pCol = (SColumnInfoData*)taosArrayGet(pUpdateBlock->pDataBlock, pInfo->primaryTsIndex);
ASSERT(pCol->info.type == TSDB_DATA_TYPE_TIMESTAMP);
blockDataEnsureCapacity(pUpdateBlock, size);
ASSERT(pBlock->info.numOfCols == pUpdateBlock->info.numOfCols);
int32_t rowId = *(int32_t*)taosArrayGet(pInfo->tsArray, pInfo->tsArrayIndex);
pInfo->groupId = getGroupId(pInfo->pOperatorDumy, pBlock, rowId);
int32_t i = 0;
for ( ; i < size; i++) {
rowId = *(int32_t*)taosArrayGet(pInfo->tsArray, i + pInfo->tsArrayIndex);
uint64_t id = getGroupId(pInfo->pOperatorDumy, pBlock, rowId);
if (pInfo->groupId != id) {
break;
}
copyOneRow(pUpdateBlock, pBlock, rowId);
}
pUpdateBlock->info.rows = i;
pInfo->tsArrayIndex += i;
pUpdateBlock->info.groupId = pInfo->groupId;
pUpdateBlock->info.type = STREAM_REPROCESS;
blockDataUpdateTsWindow(pUpdateBlock, 0);
}
// all rows have same group id
ASSERT(pInfo->tsArrayIndex >= size);
if (size > 0 && pInfo->tsArrayIndex == size) {
taosArrayClear(pInfo->tsArray);
}
} }
static void getUpdateDataBlock(SStreamBlockScanInfo* pInfo, bool invertible, SSDataBlock* pBlock, static void getUpdateDataBlock(SStreamBlockScanInfo* pInfo, bool invertible, SSDataBlock* pBlock,
@ -767,41 +854,21 @@ static void getUpdateDataBlock(SStreamBlockScanInfo* pInfo, bool invertible, SSD
SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, pInfo->primaryTsIndex); SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, pInfo->primaryTsIndex);
ASSERT(pColDataInfo->info.type == TSDB_DATA_TYPE_TIMESTAMP); ASSERT(pColDataInfo->info.type == TSDB_DATA_TYPE_TIMESTAMP);
TSKEY* ts = (TSKEY*)pColDataInfo->pData; TSKEY* ts = (TSKEY*)pColDataInfo->pData;
for (int32_t i = 0; i < pBlock->info.rows; i++) { for (int32_t rowId = 0; rowId < pBlock->info.rows; rowId++) {
if (updateInfoIsUpdated(pInfo->pUpdateInfo, pBlock->info.uid, ts[i])) { if (updateInfoIsUpdated(pInfo->pUpdateInfo, pBlock->info.uid, ts[rowId])) {
taosArrayPush(pInfo->tsArray, ts + i); taosArrayPush(pInfo->tsArray, &rowId);
} }
} }
if (!pUpdateBlock) { if (!pUpdateBlock) {
taosArrayClear(pInfo->tsArray); taosArrayClear(pInfo->tsArray);
return; return;
} }
int32_t size = taosArrayGetSize(pInfo->tsArray); setUpdateData(pInfo, pBlock, pUpdateBlock);
if (size > 0 && invertible) { // Todo(liuyao) get from tsdb
// Todo(liuyao) get from tsdb // SSDataBlock* p = createOneDataBlock(pBlock, true);
// SSDataBlock* p = createOneDataBlock(pBlock, true); // p->info.type = STREAM_INVERT;
// p->info.type = STREAM_INVERT; // taosArrayClear(pInfo->tsArray);
// taosArrayClear(pInfo->tsArray); // return p;
// return p;
SColumnInfoData* pCol = (SColumnInfoData*)taosArrayGet(pUpdateBlock->pDataBlock, pInfo->primaryTsIndex);
ASSERT(pCol->info.type == TSDB_DATA_TYPE_TIMESTAMP);
blockDataEnsureCapacity(pUpdateBlock, size);
for (int32_t i = 0; i < size; i++) {
TSKEY* pTs = (TSKEY*)taosArrayGet(pInfo->tsArray, i);
colDataAppend(pCol, i, (char*)pTs, false);
}
for (int32_t i = 0; i < pUpdateBlock->info.numOfCols; i++) {
if (i == pInfo->primaryTsIndex) {
continue;
}
SColumnInfoData* pCol = (SColumnInfoData*)taosArrayGet(pUpdateBlock->pDataBlock, i);
colDataAppendNNULL(pCol, 0, size);
}
pUpdateBlock->info.rows = size;
pUpdateBlock->info.type = STREAM_REPROCESS;
blockDataUpdateTsWindow(pUpdateBlock, 0);
taosArrayClear(pInfo->tsArray);
}
} }
static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) { static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
@ -833,7 +900,6 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE; pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
return pInfo->pRes; return pInfo->pRes;
} else if (pInfo->scanMode == STREAM_SCAN_FROM_UPDATERES) { } else if (pInfo->scanMode == STREAM_SCAN_FROM_UPDATERES) {
blockDataCleanup(pInfo->pRes);
pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER; pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER;
if (!isStateWindow(pInfo)) { if (!isStateWindow(pInfo)) {
prepareDataScan(pInfo); prepareDataScan(pInfo);
@ -848,7 +914,15 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
if (pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER) { if (pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER) {
SSDataBlock* pSDB = doDataScan(pInfo); SSDataBlock* pSDB = doDataScan(pInfo);
if (pSDB == NULL) { if (pSDB == NULL) {
pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE; setUpdateData(pInfo, pInfo->pRes, pInfo->pUpdateRes);
if (pInfo->pUpdateRes->info.rows > 0) {
if (!isStateWindow(pInfo)) {
prepareDataScan(pInfo);
}
return pInfo->pUpdateRes;
} else {
pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
}
} else { } else {
getUpdateDataBlock(pInfo, true, pSDB, NULL); getUpdateDataBlock(pInfo, true, pSDB, NULL);
return pSDB; return pSDB;
@ -941,7 +1015,7 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
if (rows == 0) { if (rows == 0) {
pOperator->status = OP_EXEC_DONE; pOperator->status = OP_EXEC_DONE;
} else if (pInfo->pUpdateInfo) { } else if (pInfo->pUpdateInfo) {
blockDataCleanup(pInfo->pUpdateRes); pInfo->tsArrayIndex = 0;
getUpdateDataBlock(pInfo, true, pInfo->pRes, pInfo->pUpdateRes); getUpdateDataBlock(pInfo, true, pInfo->pRes, pInfo->pUpdateRes);
if (pInfo->pUpdateRes->info.rows > 0) { if (pInfo->pUpdateRes->info.rows > 0) {
if (pInfo->pUpdateRes->info.type == STREAM_REPROCESS) { if (pInfo->pUpdateRes->info.type == STREAM_REPROCESS) {
@ -1020,7 +1094,7 @@ SOperatorInfo* createStreamScanOperatorInfo(void* pDataReader, SReadHandle* pHan
goto _error; goto _error;
} }
pInfo->tsArray = taosArrayInit(4, sizeof(TSKEY)); pInfo->tsArray = taosArrayInit(4, sizeof(int32_t));
if (pInfo->tsArray == NULL) { if (pInfo->tsArray == NULL) {
goto _error; goto _error;
} }
@ -1047,6 +1121,8 @@ SOperatorInfo* createStreamScanOperatorInfo(void* pDataReader, SReadHandle* pHan
pInfo->pOperatorDumy = pTableScanDummy; pInfo->pOperatorDumy = pTableScanDummy;
pInfo->interval = pSTInfo->interval; pInfo->interval = pSTInfo->interval;
pInfo->sessionSup = (SessionWindowSupporter){.pStreamAggSup = NULL, .gap = -1}; pInfo->sessionSup = (SessionWindowSupporter){.pStreamAggSup = NULL, .gap = -1};
pInfo->groupId = 0;
pOperator->name = "StreamBlockScanOperator"; pOperator->name = "StreamBlockScanOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN;
pOperator->blocking = false; pOperator->blocking = false;

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

@ -1320,6 +1320,15 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
break; break;
} }
if (pBlock->info.type == STREAM_REPROCESS) {
doClearWindows(&pInfo->aggSup, &pInfo->binfo, &pInfo->interval, 0, pOperator->numOfExprs, pBlock, NULL);
qDebug("%s clear existed time window results for updates checked", GET_TASKID(pTaskInfo));
continue;
} else if (pBlock->info.type == STREAM_GET_ALL) {
getAllIntervalWindow(pInfo->aggSup.pResultRowHashTable, pUpdated);
continue;
}
// The timewindow that overlaps the timestamps of the input pBlock need to be recalculated and return to the // The timewindow that overlaps the timestamps of the input pBlock need to be recalculated and return to the
// caller. Note that all the time window are not close till now. // caller. Note that all the time window are not close till now.
// the pDataBlock are always the same one, no need to call this again // the pDataBlock are always the same one, no need to call this again
@ -1328,16 +1337,6 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
setInverFunction(pInfo->binfo.pCtx, pOperator->numOfExprs, pBlock->info.type); setInverFunction(pInfo->binfo.pCtx, pOperator->numOfExprs, pBlock->info.type);
} }
if (pBlock->info.type == STREAM_REPROCESS) {
doClearWindows(&pInfo->aggSup, &pInfo->binfo, &pInfo->interval, 0, pOperator->numOfExprs, pBlock, NULL);
qDebug("%s clear existed time window results for updates checked", GET_TASKID(pTaskInfo));
continue;
} else if (pBlock->info.type == STREAM_GET_ALL &&
pInfo->twAggSup.calTrigger == STREAM_TRIGGER_MAX_DELAY) {
getAllIntervalWindow(pInfo->aggSup.pResultRowHashTable, pUpdated);
continue;
}
pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.window.ekey); pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.window.ekey);
hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, MAIN_SCAN, pUpdated); hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, MAIN_SCAN, pUpdated);
} }
@ -1985,7 +1984,7 @@ static void clearUpdateDataBlock(SSDataBlock* pBlock) {
blockDataCleanup(pBlock); blockDataCleanup(pBlock);
} }
static void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_t tsColIndex) { void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_t tsColIndex) {
ASSERT(pDest->info.capacity >= pSource->info.rows); ASSERT(pDest->info.capacity >= pSource->info.rows);
clearUpdateDataBlock(pDest); clearUpdateDataBlock(pDest);
SColumnInfoData* pDestCol = taosArrayGet(pDest->pDataBlock, 0); SColumnInfoData* pDestCol = taosArrayGet(pDest->pDataBlock, 0);
@ -1997,6 +1996,8 @@ static void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_
colDataAppendNNULL(pCol, 0, pSource->info.rows); colDataAppendNNULL(pCol, 0, pSource->info.rows);
} }
pDest->info.rows = pSource->info.rows; pDest->info.rows = pSource->info.rows;
pDest->info.groupId = pSource->info.groupId;
pDest->info.type = pSource->info.type;
blockDataUpdateTsWindow(pDest, 0); blockDataUpdateTsWindow(pDest, 0);
} }
@ -2036,7 +2037,6 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
break; break;
} }
setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, pInfo->order, MAIN_SCAN, true);
if (pBlock->info.type == STREAM_REPROCESS) { if (pBlock->info.type == STREAM_REPROCESS) {
SArray* pUpWins = taosArrayInit(8, sizeof(STimeWindow)); SArray* pUpWins = taosArrayInit(8, sizeof(STimeWindow));
doClearWindows(&pInfo->aggSup, &pInfo->binfo, &pInfo->interval, pInfo->primaryTsIndex, pOperator->numOfExprs, doClearWindows(&pInfo->aggSup, &pInfo->binfo, &pInfo->interval, pInfo->primaryTsIndex, pOperator->numOfExprs,
@ -2056,12 +2056,12 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
copyUpdateDataBlock(pInfo->pUpdateRes, pBlock, pInfo->primaryTsIndex); copyUpdateDataBlock(pInfo->pUpdateRes, pBlock, pInfo->primaryTsIndex);
taosArrayDestroy(pUpWins); taosArrayDestroy(pUpWins);
break; break;
} else if (pBlock->info.type == STREAM_GET_ALL && isFinalInterval(pInfo) && } else if (pBlock->info.type == STREAM_GET_ALL && isFinalInterval(pInfo)) {
pInfo->twAggSup.calTrigger == STREAM_TRIGGER_MAX_DELAY) {
getAllIntervalWindow(pInfo->aggSup.pResultRowHashTable, pUpdated); getAllIntervalWindow(pInfo->aggSup.pResultRowHashTable, pUpdated);
continue; continue;
} }
setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, pInfo->order, MAIN_SCAN, true);
if (isFinalInterval(pInfo)) { if (isFinalInterval(pInfo)) {
int32_t chIndex = getChildIndex(pBlock); int32_t chIndex = getChildIndex(pBlock);
int32_t size = taosArrayGetSize(pInfo->pChildren); int32_t size = taosArrayGetSize(pInfo->pChildren);
@ -2123,6 +2123,7 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
.calTrigger = pIntervalPhyNode->window.triggerType, .calTrigger = pIntervalPhyNode->window.triggerType,
.maxTs = INT64_MIN, .maxTs = INT64_MIN,
}; };
ASSERT(pInfo->twAggSup.calTrigger != STREAM_TRIGGER_MAX_DELAY);
pInfo->primaryTsIndex = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId; pInfo->primaryTsIndex = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId;
size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES; size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
initResultSizeInfo(pOperator, 4096); initResultSizeInfo(pOperator, 4096);
@ -2188,8 +2189,13 @@ _error:
} }
void destroyStreamAggSupporter(SStreamAggSupporter* pSup) { void destroyStreamAggSupporter(SStreamAggSupporter* pSup) {
taosArrayDestroy(pSup->pResultRows);
taosMemoryFreeClear(pSup->pKeyBuf); taosMemoryFreeClear(pSup->pKeyBuf);
void **pIte = NULL;
while ((pIte = taosHashIterate(pSup->pResultRows, pIte)) != NULL) {
SArray *pWins = (SArray *) (*pIte);
taosArrayDestroy(pWins);
}
taosHashCleanup(pSup->pResultRows);
destroyDiskbasedBuf(pSup->pResultBuf); destroyDiskbasedBuf(pSup->pResultBuf);
} }
@ -2331,7 +2337,22 @@ static SResultWindowInfo* addNewSessionWindow(SArray* pWinInfos, TSKEY ts) {
return taosArrayPush(pWinInfos, &win); return taosArrayPush(pWinInfos, &win);
} }
SResultWindowInfo* getSessionTimeWindow(SArray* pWinInfos, TSKEY ts, int64_t gap, int32_t* pIndex) { SArray* getWinInfos(SStreamAggSupporter* pAggSup, uint64_t groupId) {
void** ite = taosHashGet(pAggSup->pResultRows, &groupId, sizeof(uint64_t));
SArray* pWinInfos = NULL;
if (ite == NULL) {
pWinInfos = taosArrayInit(1024, pAggSup->valueSize);
taosHashPut(pAggSup->pResultRows, &groupId, sizeof(uint64_t), &pWinInfos, sizeof(void *));
} else {
pWinInfos = *ite;
}
return pWinInfos;
}
SResultWindowInfo* getSessionTimeWindow(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId, int64_t gap, int32_t* pIndex) {
SArray* pWinInfos = getWinInfos(pAggSup, groupId);
pAggSup->pCurWins = pWinInfos;
int32_t size = taosArrayGetSize(pWinInfos); int32_t size = taosArrayGetSize(pWinInfos);
if (size == 0) { if (size == 0) {
*pIndex = 0; *pIndex = 0;
@ -2387,7 +2408,7 @@ static int32_t setWindowOutputBuf(SResultWindowInfo* pWinInfo, SResultRow** pRes
SStreamAggSupporter* pAggSup, SExecTaskInfo* pTaskInfo) { SStreamAggSupporter* pAggSup, SExecTaskInfo* pTaskInfo) {
assert(pWinInfo->win.skey <= pWinInfo->win.ekey); assert(pWinInfo->win.skey <= pWinInfo->win.ekey);
// too many time window in query // too many time window in query
int32_t size = taosArrayGetSize(pAggSup->pResultRows); int32_t size = taosArrayGetSize(pAggSup->pCurWins);
if (size > MAX_INTERVAL_TIME_WINDOW) { if (size > MAX_INTERVAL_TIME_WINDOW) {
longjmp(pTaskInfo->env, TSDB_CODE_QRY_TOO_MANY_TIMEWINDOW); longjmp(pTaskInfo->env, TSDB_CODE_QRY_TOO_MANY_TIMEWINDOW);
} }
@ -2447,25 +2468,6 @@ static int32_t doOneStateWindowAgg(SStreamStateAggOperatorInfo* pInfo, SSDataBlo
pSDataBlock, pCurWin, pResult, startIndex, winRows, numOutput, pTaskInfo); pSDataBlock, pCurWin, pResult, startIndex, winRows, numOutput, pTaskInfo);
} }
int32_t copyWinInfoToDataBlock(SSDataBlock* pBlock, SStreamAggSupporter* pAggSup, int32_t start, int32_t num,
int32_t numOfExprs, SOptrBasicInfo* pBinfo) {
for (int32_t i = start; i < num; i += 1) {
SResultWindowInfo* pWinInfo = taosArrayGet(pAggSup->pResultRows, start);
SFilePage* bufPage = getBufPage(pAggSup->pResultBuf, pWinInfo->pos.pageId);
SResultRow* pRow = (SResultRow*)((char*)bufPage + pWinInfo->pos.offset);
for (int32_t j = 0; j < numOfExprs; ++j) {
SResultRowEntryInfo* pResultInfo = getResultCell(pRow, j, pBinfo->rowCellInfoOffset);
SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, j);
char* in = GET_ROWCELL_INTERBUF(pBinfo->pCtx[j].resultInfo);
colDataAppend(pColInfoData, pBlock->info.rows, in, pResultInfo->isNullRes);
}
pBlock->info.rows += pRow->numOfRows;
releaseBufPage(pAggSup->pResultBuf, bufPage);
}
blockDataUpdateTsWindow(pBlock, -1);
return TSDB_CODE_SUCCESS;
}
int32_t getNumCompactWindow(SArray* pWinInfos, int32_t startIndex, int64_t gap) { int32_t getNumCompactWindow(SArray* pWinInfos, int32_t startIndex, int64_t gap) {
SResultWindowInfo* pCurWin = taosArrayGet(pWinInfos, startIndex); SResultWindowInfo* pCurWin = taosArrayGet(pWinInfos, startIndex);
int32_t size = taosArrayGetSize(pWinInfos); int32_t size = taosArrayGetSize(pWinInfos);
@ -2482,15 +2484,15 @@ int32_t getNumCompactWindow(SArray* pWinInfos, int32_t startIndex, int64_t gap)
void compactTimeWindow(SStreamSessionAggOperatorInfo* pInfo, int32_t startIndex, int32_t num, int32_t groupId, void compactTimeWindow(SStreamSessionAggOperatorInfo* pInfo, int32_t startIndex, int32_t num, int32_t groupId,
int32_t numOfOutput, SExecTaskInfo* pTaskInfo, SHashObj* pStUpdated, SHashObj* pStDeleted) { int32_t numOfOutput, SExecTaskInfo* pTaskInfo, SHashObj* pStUpdated, SHashObj* pStDeleted) {
SResultWindowInfo* pCurWin = taosArrayGet(pInfo->streamAggSup.pResultRows, startIndex); SResultWindowInfo* pCurWin = taosArrayGet(pInfo->streamAggSup.pCurWins, startIndex);
SResultRow* pCurResult = NULL; SResultRow* pCurResult = NULL;
setWindowOutputBuf(pCurWin, &pCurResult, pInfo->binfo.pCtx, groupId, numOfOutput, pInfo->binfo.rowCellInfoOffset, setWindowOutputBuf(pCurWin, &pCurResult, pInfo->binfo.pCtx, groupId, numOfOutput, pInfo->binfo.rowCellInfoOffset,
&pInfo->streamAggSup, pTaskInfo); &pInfo->streamAggSup, pTaskInfo);
num += startIndex + 1; num += startIndex + 1;
ASSERT(num <= taosArrayGetSize(pInfo->streamAggSup.pResultRows)); ASSERT(num <= taosArrayGetSize(pInfo->streamAggSup.pCurWins));
// Just look for the window behind StartIndex // Just look for the window behind StartIndex
for (int32_t i = startIndex + 1; i < num; i++) { for (int32_t i = startIndex + 1; i < num; i++) {
SResultWindowInfo* pWinInfo = taosArrayGet(pInfo->streamAggSup.pResultRows, i); SResultWindowInfo* pWinInfo = taosArrayGet(pInfo->streamAggSup.pCurWins, i);
SResultRow* pWinResult = NULL; SResultRow* pWinResult = NULL;
setWindowOutputBuf(pWinInfo, &pWinResult, pInfo->pDummyCtx, groupId, numOfOutput, pInfo->binfo.rowCellInfoOffset, setWindowOutputBuf(pWinInfo, &pWinResult, pInfo->pDummyCtx, groupId, numOfOutput, pInfo->binfo.rowCellInfoOffset,
&pInfo->streamAggSup, pTaskInfo); &pInfo->streamAggSup, pTaskInfo);
@ -2501,7 +2503,7 @@ void compactTimeWindow(SStreamSessionAggOperatorInfo* pInfo, int32_t startIndex,
taosHashPut(pStDeleted, &pWinInfo->pos, sizeof(SResultRowPosition), &pWinInfo->win.skey, sizeof(TSKEY)); taosHashPut(pStDeleted, &pWinInfo->pos, sizeof(SResultRowPosition), &pWinInfo->win.skey, sizeof(TSKEY));
pWinInfo->isOutput = false; pWinInfo->isOutput = false;
} }
taosArrayRemove(pInfo->streamAggSup.pResultRows, i); taosArrayRemove(pInfo->streamAggSup.pCurWins, i);
} }
} }
@ -2531,7 +2533,7 @@ static void doStreamSessionAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSData
SStreamAggSupporter* pAggSup = &pInfo->streamAggSup; SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
for (int32_t i = 0; i < pSDataBlock->info.rows;) { for (int32_t i = 0; i < pSDataBlock->info.rows;) {
int32_t winIndex = 0; int32_t winIndex = 0;
SResultWindowInfo* pCurWin = getSessionTimeWindow(pAggSup->pResultRows, tsCols[i], gap, &winIndex); SResultWindowInfo* pCurWin = getSessionTimeWindow(pAggSup, tsCols[i], pSDataBlock->info.groupId, gap, &winIndex);
winRows = updateSessionWindowInfo(pCurWin, tsCols, pSDataBlock->info.rows, i, pInfo->gap, pStDeleted); winRows = updateSessionWindowInfo(pCurWin, tsCols, pSDataBlock->info.rows, i, pInfo->gap, pStDeleted);
code = doOneWindowAgg(pInfo, pSDataBlock, pCurWin, &pResult, i, winRows, numOfOutput, pTaskInfo); code = doOneWindowAgg(pInfo, pSDataBlock, pCurWin, &pResult, i, winRows, numOfOutput, pTaskInfo);
if (code != TSDB_CODE_SUCCESS || pResult == NULL) { if (code != TSDB_CODE_SUCCESS || pResult == NULL) {
@ -2541,7 +2543,7 @@ static void doStreamSessionAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSData
// doWindowBorderInterpolation(pOperatorInfo, pSDataBlock, pInfo->binfo.pCtx, pResult, &nextWin, startPos, // doWindowBorderInterpolation(pOperatorInfo, pSDataBlock, pInfo->binfo.pCtx, pResult, &nextWin, startPos,
// forwardRows, // forwardRows,
// pInfo->order, false); // pInfo->order, false);
int32_t winNum = getNumCompactWindow(pAggSup->pResultRows, winIndex, gap); int32_t winNum = getNumCompactWindow(pAggSup->pCurWins, winIndex, gap);
if (winNum > 0) { if (winNum > 0) {
compactTimeWindow(pInfo, winIndex, winNum, groupId, numOfOutput, pTaskInfo, pStUpdated, pStDeleted); compactTimeWindow(pInfo, winIndex, winNum, groupId, numOfOutput, pTaskInfo, pStUpdated, pStDeleted);
} }
@ -2564,7 +2566,7 @@ static void doClearSessionWindows(SStreamAggSupporter* pAggSup, SOptrBasicInfo*
int32_t step = 0; int32_t step = 0;
for (int32_t i = 0; i < pBlock->info.rows; i += step) { for (int32_t i = 0; i < pBlock->info.rows; i += step) {
int32_t winIndex = 0; int32_t winIndex = 0;
SResultWindowInfo* pCurWin = getSessionTimeWindow(pAggSup->pResultRows, tsCols[i], gap, &winIndex); SResultWindowInfo* pCurWin = getSessionTimeWindow(pAggSup, tsCols[i], pBlock->info.groupId, gap, &winIndex);
step = updateSessionWindowInfo(pCurWin, tsCols, pBlock->info.rows, i, gap, NULL); step = updateSessionWindowInfo(pCurWin, tsCols, pBlock->info.rows, i, gap, NULL);
ASSERT(isInWindow(pCurWin, tsCols[i], gap)); ASSERT(isInWindow(pCurWin, tsCols[i], gap));
doClearWindowImpl(&pCurWin->pos, pAggSup->pResultBuf, pBinfo, numOfOutput); doClearWindowImpl(&pCurWin->pos, pAggSup->pResultBuf, pBinfo, numOfOutput);
@ -2625,7 +2627,7 @@ static void rebuildTimeWindow(SStreamSessionAggOperatorInfo* pInfo, SArray* pWin
for (int32_t j = 0; j < numOfChildren; j++) { for (int32_t j = 0; j < numOfChildren; j++) {
SOperatorInfo* pChild = taosArrayGetP(pInfo->pChildren, j); SOperatorInfo* pChild = taosArrayGetP(pInfo->pChildren, j);
SStreamSessionAggOperatorInfo* pChInfo = pChild->info; SStreamSessionAggOperatorInfo* pChInfo = pChild->info;
SArray* pChWins = pChInfo->streamAggSup.pResultRows; SArray* pChWins = getWinInfos(&pChInfo->streamAggSup, groupId);
int32_t chWinSize = taosArrayGetSize(pChWins); int32_t chWinSize = taosArrayGetSize(pChWins);
int32_t index = binarySearch(pChWins, chWinSize, pParentWin->win.skey, TSDB_ORDER_DESC, getSessionWindowEndkey); int32_t index = binarySearch(pChWins, chWinSize, pParentWin->win.skey, TSDB_ORDER_DESC, getSessionWindowEndkey);
for (int32_t k = index; k > 0 && k < chWinSize; k++) { for (int32_t k = index; k > 0 && k < chWinSize; k++) {
@ -2649,36 +2651,44 @@ typedef SResultWindowInfo* (*__get_win_info_)(void*);
SResultWindowInfo* getSessionWinInfo(void* pData) { return (SResultWindowInfo*)pData; } SResultWindowInfo* getSessionWinInfo(void* pData) { return (SResultWindowInfo*)pData; }
SResultWindowInfo* getStateWinInfo(void* pData) { return &((SStateWindowInfo*)pData)->winInfo; } SResultWindowInfo* getStateWinInfo(void* pData) { return &((SStateWindowInfo*)pData)->winInfo; }
int32_t closeSessionWindow(SArray* pWins, STimeWindowAggSupp* pTwSup, SArray* pClosed, int32_t closeSessionWindow(SHashObj* pHashMap, STimeWindowAggSupp* pTwSup, SArray* pClosed,
__get_win_info_ fn) { __get_win_info_ fn) {
// Todo(liuyao) save window to tdb // Todo(liuyao) save window to tdb
int32_t size = taosArrayGetSize(pWins); void **pIte = NULL;
for (int32_t i = 0; i < size; i++) { while ((pIte = taosHashIterate(pHashMap, pIte)) != NULL) {
void* pWin = taosArrayGet(pWins, i); SArray *pWins = (SArray *) (*pIte);
SResultWindowInfo* pSeWin = fn(pWin); int32_t size = taosArrayGetSize(pWins);
if (pSeWin->win.ekey < pTwSup->maxTs - pTwSup->waterMark) { for (int32_t i = 0; i < size; i++) {
if (!pSeWin->isClosed) { void* pWin = taosArrayGet(pWins, i);
pSeWin->isClosed = true; SResultWindowInfo* pSeWin = fn(pWin);
if (pTwSup->calTrigger == STREAM_TRIGGER_WINDOW_CLOSE) { if (pSeWin->win.ekey < pTwSup->maxTs - pTwSup->waterMark) {
int32_t code = saveResult(pSeWin->win.skey, pSeWin->pos.pageId, pSeWin->pos.offset, 0, pClosed); if (!pSeWin->isClosed) {
pSeWin->isOutput = true; pSeWin->isClosed = true;
if (pTwSup->calTrigger == STREAM_TRIGGER_WINDOW_CLOSE) {
int32_t code = saveResult(pSeWin->win.skey, pSeWin->pos.pageId, pSeWin->pos.offset, 0, pClosed);
pSeWin->isOutput = true;
}
} }
continue;
} }
continue; break;
} }
break;
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t getAllSessionWindow(SArray* pWins, SArray* pClosed, __get_win_info_ fn) { int32_t getAllSessionWindow(SHashObj* pHashMap, SArray* pClosed, __get_win_info_ fn) {
int32_t size = taosArrayGetSize(pWins); void **pIte = NULL;
for (int32_t i = 0; i < size; i++) { while ((pIte = taosHashIterate(pHashMap, pIte)) != NULL) {
void* pWin = taosArrayGet(pWins, i); SArray *pWins = (SArray *) (*pIte);
SResultWindowInfo* pSeWin = fn(pWin); int32_t size = taosArrayGetSize(pWins);
if (!pSeWin->isClosed) { for (int32_t i = 0; i < size; i++) {
int32_t code = saveResult(pSeWin->win.skey, pSeWin->pos.pageId, pSeWin->pos.offset, 0, pClosed); void* pWin = taosArrayGet(pWins, i);
pSeWin->isOutput = true; SResultWindowInfo* pSeWin = fn(pWin);
if (!pSeWin->isClosed) {
int32_t code = saveResult(pSeWin->win.skey, pSeWin->pos.pageId, pSeWin->pos.offset, 0, pClosed);
pSeWin->isOutput = true;
}
} }
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -2712,8 +2722,7 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
if (pBlock == NULL) { if (pBlock == NULL) {
break; break;
} }
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pOperator, pBInfo->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
if (pBlock->info.type == STREAM_REPROCESS) { if (pBlock->info.type == STREAM_REPROCESS) {
SArray* pWins = taosArrayInit(16, sizeof(SResultWindowInfo)); SArray* pWins = taosArrayInit(16, sizeof(SResultWindowInfo));
doClearSessionWindows(&pInfo->streamAggSup, &pInfo->binfo, pBlock, 0, pOperator->numOfExprs, pInfo->gap, pWins); doClearSessionWindows(&pInfo->streamAggSup, &pInfo->binfo, pBlock, 0, pOperator->numOfExprs, pInfo->gap, pWins);
@ -2727,12 +2736,13 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
} }
taosArrayDestroy(pWins); taosArrayDestroy(pWins);
continue; continue;
} else if (pBlock->info.type == STREAM_GET_ALL && } else if (pBlock->info.type == STREAM_GET_ALL) {
pInfo->twAggSup.calTrigger == STREAM_TRIGGER_MAX_DELAY) {
getAllSessionWindow(pInfo->streamAggSup.pResultRows, pUpdated, getSessionWinInfo); getAllSessionWindow(pInfo->streamAggSup.pResultRows, pUpdated, getSessionWinInfo);
continue; continue;
} }
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pOperator, pBInfo->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
if (isFinalSession(pInfo)) { if (isFinalSession(pInfo)) {
int32_t childIndex = 0; // Todo(liuyao) get child id from SSDataBlock int32_t childIndex = 0; // Todo(liuyao) get child id from SSDataBlock
SOptrBasicInfo* pChildOp = taosArrayGetP(pInfo->pChildren, childIndex); SOptrBasicInfo* pChildOp = taosArrayGetP(pInfo->pChildren, childIndex);
@ -2871,7 +2881,9 @@ bool isEqualStateKey(SStateWindowInfo* pWin, char* pKeyData) {
return pKeyData && compareVal(pKeyData, &pWin->stateKey); return pKeyData && compareVal(pKeyData, &pWin->stateKey);
} }
SStateWindowInfo* getStateWindowByTs(SArray* pWinInfos, TSKEY ts, int32_t* pIndex) { SStateWindowInfo* getStateWindowByTs(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId, int32_t* pIndex) {
SArray* pWinInfos = getWinInfos(pAggSup, groupId);
pAggSup->pCurWins = pWinInfos;
int32_t size = taosArrayGetSize(pWinInfos); int32_t size = taosArrayGetSize(pWinInfos);
int32_t index = binarySearch(pWinInfos, size, ts, TSDB_ORDER_DESC, getStateWinTsKey); int32_t index = binarySearch(pWinInfos, size, ts, TSDB_ORDER_DESC, getStateWinTsKey);
SStateWindowInfo* pWin = NULL; SStateWindowInfo* pWin = NULL;
@ -2894,7 +2906,10 @@ SStateWindowInfo* getStateWindowByTs(SArray* pWinInfos, TSKEY ts, int32_t* pInde
return NULL; return NULL;
} }
SStateWindowInfo* getStateWindow(SArray* pWinInfos, TSKEY ts, char* pKeyData, SColumn* pCol, int32_t* pIndex) { SStateWindowInfo* getStateWindow(SStreamAggSupporter* pAggSup, TSKEY ts,
uint64_t groupId, char* pKeyData, SColumn* pCol, int32_t* pIndex) {
SArray* pWinInfos = getWinInfos(pAggSup, groupId);
pAggSup->pCurWins = pWinInfos;
int32_t size = taosArrayGetSize(pWinInfos); int32_t size = taosArrayGetSize(pWinInfos);
if (size == 0) { if (size == 0) {
*pIndex = 0; *pIndex = 0;
@ -2985,16 +3000,16 @@ static void doClearStateWindows(SStreamAggSupporter* pAggSup, SSDataBlock* pBloc
for (int32_t i = 0; i < pBlock->info.rows; i += step) { for (int32_t i = 0; i < pBlock->info.rows; i += step) {
char* pKeyData = colDataGetData(pKeyColInfo, i); char* pKeyData = colDataGetData(pKeyColInfo, i);
int32_t winIndex = 0; int32_t winIndex = 0;
SStateWindowInfo* pCurWin = getStateWindowByTs(pAggSup->pResultRows, tsCol[i], &winIndex); SStateWindowInfo* pCurWin = getStateWindowByTs(pAggSup, tsCol[i], pBlock->info.groupId, &winIndex);
if (!pCurWin) { if (!pCurWin) {
continue; continue;
} }
step = updateStateWindowInfo(pAggSup->pResultRows, winIndex, tsCol, pKeyColInfo, pBlock->info.rows, i, &allEqual, step = updateStateWindowInfo(pAggSup->pCurWins, winIndex, tsCol, pKeyColInfo, pBlock->info.rows, i, &allEqual,
pSeDeleted); pSeDeleted);
ASSERT(isTsInWindow(pCurWin, tsCol[i]) || isEqualStateKey(pCurWin, pKeyData)); ASSERT(isTsInWindow(pCurWin, tsCol[i]) || isEqualStateKey(pCurWin, pKeyData));
taosArrayPush(pAggSup->pScanWindow, &pCurWin->winInfo.win); taosArrayPush(pAggSup->pScanWindow, &pCurWin->winInfo.win);
taosHashRemove(pSeUpdated, &pCurWin->winInfo.pos, sizeof(SResultRowPosition)); taosHashRemove(pSeUpdated, &pCurWin->winInfo.pos, sizeof(SResultRowPosition));
deleteWindow(pAggSup->pResultRows, winIndex); deleteWindow(pAggSup->pCurWins, winIndex);
} }
} }
@ -3024,13 +3039,15 @@ static void doStreamStateAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl
char* pKeyData = colDataGetData(pKeyColInfo, i); char* pKeyData = colDataGetData(pKeyColInfo, i);
int32_t winIndex = 0; int32_t winIndex = 0;
bool allEqual = true; bool allEqual = true;
SStateWindowInfo* pCurWin = getStateWindow(pAggSup->pResultRows, tsCols[i], pKeyData, &pInfo->stateCol, &winIndex); SStateWindowInfo* pCurWin =
winRows = updateStateWindowInfo(pAggSup->pResultRows, winIndex, tsCols, pKeyColInfo, pSDataBlock->info.rows, i, getStateWindow(pAggSup, tsCols[i], pSDataBlock->info.groupId, pKeyData,
&allEqual, pInfo->pSeDeleted); &pInfo->stateCol, &winIndex);
winRows = updateStateWindowInfo(pAggSup->pCurWins, winIndex, tsCols, pKeyColInfo,
pSDataBlock->info.rows, i, &allEqual, pInfo->pSeDeleted);
if (!allEqual) { if (!allEqual) {
taosArrayPush(pAggSup->pScanWindow, &pCurWin->winInfo.win); taosArrayPush(pAggSup->pScanWindow, &pCurWin->winInfo.win);
taosHashRemove(pSeUpdated, &pCurWin->winInfo.pos, sizeof(SResultRowPosition)); taosHashRemove(pSeUpdated, &pCurWin->winInfo.pos, sizeof(SResultRowPosition));
deleteWindow(pAggSup->pResultRows, winIndex); deleteWindow(pAggSup->pCurWins, winIndex);
continue; continue;
} }
code = doOneStateWindowAgg(pInfo, pSDataBlock, &pCurWin->winInfo, &pResult, i, winRows, numOfOutput, pTaskInfo); code = doOneStateWindowAgg(pInfo, pSDataBlock, &pCurWin->winInfo, &pResult, i, winRows, numOfOutput, pTaskInfo);
@ -3077,17 +3094,18 @@ static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) {
if (pBlock == NULL) { if (pBlock == NULL) {
break; break;
} }
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pOperator, pBInfo->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
if (pBlock->info.type == STREAM_REPROCESS) { if (pBlock->info.type == STREAM_REPROCESS) {
doClearStateWindows(&pInfo->streamAggSup, pBlock, pInfo->primaryTsIndex, &pInfo->stateCol, pInfo->stateCol.slotId, doClearStateWindows(&pInfo->streamAggSup, pBlock, pInfo->primaryTsIndex, &pInfo->stateCol, pInfo->stateCol.slotId,
pSeUpdated, pInfo->pSeDeleted); pSeUpdated, pInfo->pSeDeleted);
continue; continue;
} else if (pBlock->info.type == STREAM_GET_ALL && } else if (pBlock->info.type == STREAM_GET_ALL) {
pInfo->twAggSup.calTrigger == STREAM_TRIGGER_MAX_DELAY) {
getAllSessionWindow(pInfo->streamAggSup.pResultRows, pUpdated, getStateWinInfo); getAllSessionWindow(pInfo->streamAggSup.pResultRows, pUpdated, getStateWinInfo);
continue; continue;
} }
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pOperator, pBInfo->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
doStreamStateAggImpl(pOperator, pBlock, pSeUpdated, pInfo->pSeDeleted); doStreamStateAggImpl(pOperator, pBlock, pSeUpdated, pInfo->pSeDeleted);
pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.window.ekey); pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.window.ekey);
} }

4
source/libs/function/src/builtins.c
View File

@ -1682,7 +1682,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{ {
.name = "top", .name = "top",
.type = FUNCTION_TYPE_TOP, .type = FUNCTION_TYPE_TOP,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_INDEFINITE_ROWS_FUNC, .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
.translateFunc = translateTopBot, .translateFunc = translateTopBot,
.getEnvFunc = getTopBotFuncEnv, .getEnvFunc = getTopBotFuncEnv,
.initFunc = topBotFunctionSetup, .initFunc = topBotFunctionSetup,
@ -1717,7 +1717,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{ {
.name = "bottom", .name = "bottom",
.type = FUNCTION_TYPE_BOTTOM, .type = FUNCTION_TYPE_BOTTOM,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_INDEFINITE_ROWS_FUNC, .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
.translateFunc = translateTopBot, .translateFunc = translateTopBot,
.getEnvFunc = getTopBotFuncEnv, .getEnvFunc = getTopBotFuncEnv,
.initFunc = topBotFunctionSetup, .initFunc = topBotFunctionSetup,

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

@ -2345,14 +2345,7 @@ int32_t apercentileCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx)
SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx); SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx);
SAPercentileInfo* pSBuf = GET_ROWCELL_INTERBUF(pSResInfo); SAPercentileInfo* pSBuf = GET_ROWCELL_INTERBUF(pSResInfo);
ASSERT(pDBuf->algo == pSBuf->algo); ASSERT(pDBuf->algo == pSBuf->algo);
if (pDBuf->algo == APERCT_ALGO_TDIGEST) { apercentileTransferInfo(pSBuf, pDBuf);
tdigestMerge(pDBuf->pTDigest, pSBuf->pTDigest);
} else {
SHistogramInfo* pTmp = tHistogramMerge(pDBuf->pHisto, pSBuf->pHisto, MAX_HISTOGRAM_BIN);
memcpy(pDBuf->pHisto, pTmp, sizeof(SHistogramInfo) + sizeof(SHistBin) * (MAX_HISTOGRAM_BIN + 1));
pDBuf->pHisto->elems = (SHistBin*)((char*)pDBuf->pHisto + sizeof(SHistogramInfo));
tHistogramDestroy(&pTmp);
}
pDResInfo->numOfRes = TMAX(pDResInfo->numOfRes, pSResInfo->numOfRes); pDResInfo->numOfRes = TMAX(pDResInfo->numOfRes, pSResInfo->numOfRes);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -2392,6 +2385,7 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
SInputColumnInfoData* pInput = &pCtx->input; SInputColumnInfoData* pInput = &pCtx->input;
SColumnInfoData* pInputCol = pInput->pData[0]; SColumnInfoData* pInputCol = pInput->pData[0];
int32_t type = pInputCol->info.type;
int32_t bytes = pInputCol->info.bytes; int32_t bytes = pInputCol->info.bytes;
pInfo->bytes = bytes; pInfo->bytes = bytes;
@ -2428,6 +2422,10 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
TSKEY cts = getRowPTs(pInput->pPTS, i); TSKEY cts = getRowPTs(pInput->pPTS, i);
if (pResInfo->numOfRes == 0 || *(TSKEY*)(pInfo->buf + bytes) > cts) { if (pResInfo->numOfRes == 0 || *(TSKEY*)(pInfo->buf + bytes) > cts) {
if (IS_VAR_DATA_TYPE(type)) {
bytes = varDataTLen(data);
pInfo->bytes = bytes;
}
memcpy(pInfo->buf, data, bytes); memcpy(pInfo->buf, data, bytes);
*(TSKEY*)(pInfo->buf + bytes) = cts; *(TSKEY*)(pInfo->buf + bytes) = cts;
pInfo->hasResult = true; pInfo->hasResult = true;
@ -2458,6 +2456,10 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
TSKEY cts = getRowPTs(pInput->pPTS, i); TSKEY cts = getRowPTs(pInput->pPTS, i);
if (pResInfo->numOfRes == 0 || *(TSKEY*)(pInfo->buf + bytes) > cts) { if (pResInfo->numOfRes == 0 || *(TSKEY*)(pInfo->buf + bytes) > cts) {
if (IS_VAR_DATA_TYPE(type)) {
bytes = varDataTLen(data);
pInfo->bytes = bytes;
}
memcpy(pInfo->buf, data, bytes); memcpy(pInfo->buf, data, bytes);
*(TSKEY*)(pInfo->buf + bytes) = cts; *(TSKEY*)(pInfo->buf + bytes) = cts;
pInfo->hasResult = true; pInfo->hasResult = true;
@ -2481,6 +2483,7 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
SInputColumnInfoData* pInput = &pCtx->input; SInputColumnInfoData* pInput = &pCtx->input;
SColumnInfoData* pInputCol = pInput->pData[0]; SColumnInfoData* pInputCol = pInput->pData[0];
int32_t type = pInputCol->info.type;
int32_t bytes = pInputCol->info.bytes; int32_t bytes = pInputCol->info.bytes;
pInfo->bytes = bytes; pInfo->bytes = bytes;
@ -2508,6 +2511,10 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
char* data = colDataGetData(pInputCol, i); char* data = colDataGetData(pInputCol, i);
TSKEY cts = getRowPTs(pInput->pPTS, i); TSKEY cts = getRowPTs(pInput->pPTS, i);
if (pResInfo->numOfRes == 0 || *(TSKEY*)(pInfo->buf + bytes) < cts) { if (pResInfo->numOfRes == 0 || *(TSKEY*)(pInfo->buf + bytes) < cts) {
if (IS_VAR_DATA_TYPE(type)) {
bytes = varDataTLen(data);
pInfo->bytes = bytes;
}
memcpy(pInfo->buf, data, bytes); memcpy(pInfo->buf, data, bytes);
*(TSKEY*)(pInfo->buf + bytes) = cts; *(TSKEY*)(pInfo->buf + bytes) = cts;
// DO_UPDATE_TAG_COLUMNS(pCtx, ts); // DO_UPDATE_TAG_COLUMNS(pCtx, ts);
@ -2527,6 +2534,10 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
char* data = colDataGetData(pInputCol, i); char* data = colDataGetData(pInputCol, i);
TSKEY cts = getRowPTs(pInput->pPTS, i); TSKEY cts = getRowPTs(pInput->pPTS, i);
if (pResInfo->numOfRes == 0 || *(TSKEY*)(pInfo->buf + bytes) < cts) { if (pResInfo->numOfRes == 0 || *(TSKEY*)(pInfo->buf + bytes) < cts) {
if (IS_VAR_DATA_TYPE(type)) {
bytes = varDataTLen(data);
pInfo->bytes = bytes;
}
memcpy(pInfo->buf, data, bytes); memcpy(pInfo->buf, data, bytes);
*(TSKEY*)(pInfo->buf + bytes) = cts; *(TSKEY*)(pInfo->buf + bytes) = cts;
pInfo->hasResult = true; pInfo->hasResult = true;

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

@ -104,10 +104,14 @@ static bool osdMayBeOptimized(SLogicNode* pNode) {
return false; return false;
} }
if (NULL == pNode->pParent || (QUERY_NODE_LOGIC_PLAN_WINDOW != nodeType(pNode->pParent) && if (NULL == pNode->pParent || (QUERY_NODE_LOGIC_PLAN_WINDOW != nodeType(pNode->pParent) &&
QUERY_NODE_LOGIC_PLAN_AGG != nodeType(pNode->pParent))) { QUERY_NODE_LOGIC_PLAN_AGG != nodeType(pNode->pParent) &&
QUERY_NODE_LOGIC_PLAN_PARTITION != nodeType(pNode->pParent))) {
return false; return false;
} }
if (QUERY_NODE_LOGIC_PLAN_WINDOW == nodeType(pNode->pParent)) { if (QUERY_NODE_LOGIC_PLAN_WINDOW == nodeType(pNode->pParent) ||
(QUERY_NODE_LOGIC_PLAN_PARTITION == nodeType(pNode->pParent) &&
pNode->pParent->pParent &&
QUERY_NODE_LOGIC_PLAN_WINDOW == nodeType(pNode->pParent->pParent)) ) {
return true; return true;
} }
return !osdHaveNormalCol(((SAggLogicNode*)pNode->pParent)->pGroupKeys); return !osdHaveNormalCol(((SAggLogicNode*)pNode->pParent)->pGroupKeys);
@ -217,16 +221,22 @@ static int32_t osdGetDataRequired(SNodeList* pFuncs) {
} }
static void setScanWindowInfo(SScanLogicNode* pScan) { static void setScanWindowInfo(SScanLogicNode* pScan) {
if (QUERY_NODE_LOGIC_PLAN_WINDOW == nodeType(pScan->node.pParent)) { SLogicNode* pParent = pScan->node.pParent;
pScan->interval = ((SWindowLogicNode*)pScan->node.pParent)->interval; if (QUERY_NODE_LOGIC_PLAN_PARTITION == nodeType(pParent) &&
pScan->offset = ((SWindowLogicNode*)pScan->node.pParent)->offset; pParent->pParent &&
pScan->sliding = ((SWindowLogicNode*)pScan->node.pParent)->sliding; QUERY_NODE_LOGIC_PLAN_WINDOW == nodeType(pParent->pParent)) {
pScan->intervalUnit = ((SWindowLogicNode*)pScan->node.pParent)->intervalUnit; pParent = pParent->pParent;
pScan->slidingUnit = ((SWindowLogicNode*)pScan->node.pParent)->slidingUnit; }
pScan->triggerType = ((SWindowLogicNode*)pScan->node.pParent)->triggerType; if (QUERY_NODE_LOGIC_PLAN_WINDOW == nodeType(pParent)) {
pScan->watermark = ((SWindowLogicNode*)pScan->node.pParent)->watermark; pScan->interval = ((SWindowLogicNode*)pParent)->interval;
pScan->tsColId = ((SColumnNode*)((SWindowLogicNode*)pScan->node.pParent)->pTspk)->colId; pScan->offset = ((SWindowLogicNode*)pParent)->offset;
pScan->filesFactor = ((SWindowLogicNode*)pScan->node.pParent)->filesFactor; pScan->sliding = ((SWindowLogicNode*)pParent)->sliding;
pScan->intervalUnit = ((SWindowLogicNode*)pParent)->intervalUnit;
pScan->slidingUnit = ((SWindowLogicNode*)pParent)->slidingUnit;
pScan->triggerType = ((SWindowLogicNode*)pParent)->triggerType;
pScan->watermark = ((SWindowLogicNode*)pParent)->watermark;
pScan->tsColId = ((SColumnNode*)((SWindowLogicNode*)pParent)->pTspk)->colId;
pScan->filesFactor = ((SWindowLogicNode*)pParent)->filesFactor;
} }
} }

7
source/libs/stream/inc/streamInc.h
View File

@ -23,6 +23,13 @@
extern "C" { extern "C" {
#endif #endif
typedef struct {
int8_t inited;
void* timer;
} SStreamGlobalEnv;
static SStreamGlobalEnv streamEnv;
int32_t streamExec(SStreamTask* pTask, SMsgCb* pMsgCb); int32_t streamExec(SStreamTask* pTask, SMsgCb* pMsgCb);
int32_t streamDispatch(SStreamTask* pTask, SMsgCb* pMsgCb); int32_t streamDispatch(SStreamTask* pTask, SMsgCb* pMsgCb);
int32_t streamDispatchReqToData(const SStreamDispatchReq* pReq, SStreamDataBlock* pData); int32_t streamDispatchReqToData(const SStreamDispatchReq* pReq, SStreamDataBlock* pData);

68
source/libs/stream/src/stream.c
View File

@ -14,8 +14,74 @@
*/ */
#include "streamInc.h" #include "streamInc.h"
#include "ttimer.h"
int32_t streamTriggerByWrite(SStreamTask* pTask, int32_t vgId, SMsgCb* pMsgCb) { int32_t streamInit() {
int8_t old;
while (1) {
old = atomic_val_compare_exchange_8(&streamEnv.inited, 0, 2);
if (old != 2) break;
}
if (old == 0) {
streamEnv.timer = taosTmrInit(10000, 100, 10000, "STREAM");
if (streamEnv.timer == NULL) {
atomic_store_8(&streamEnv.inited, 0);
return -1;
}
atomic_store_8(&streamEnv.inited, 1);
}
return 0;
}
void streamCleanUp() {
int8_t old;
while (1) {
old = atomic_val_compare_exchange_8(&streamEnv.inited, 1, 2);
if (old != 2) break;
}
if (old == 1) {
taosTmrCleanUp(streamEnv.timer);
atomic_store_8(&streamEnv.inited, 0);
}
}
void streamTriggerByTimer(void* param, void* tmrId) {
SStreamTask* pTask = (void*)param;
if (atomic_load_8(&pTask->triggerStatus) == TASK_TRIGGER_STATUS__ACTIVE) {
SStreamTrigger* trigger = taosAllocateQitem(sizeof(SStreamTrigger), DEF_QITEM);
if (trigger == NULL) return;
trigger->type = STREAM_INPUT__TRIGGER;
trigger->pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
if (trigger->pBlock == NULL) {
taosFreeQitem(trigger);
return;
}
trigger->pBlock->info.type = STREAM_GET_ALL;
atomic_store_8(&pTask->triggerStatus, TASK_TRIGGER_STATUS__IN_ACTIVE);
streamTaskInput(pTask, (SStreamQueueItem*)trigger);
streamLaunchByWrite(pTask, pTask->nodeId, pTask->pMsgCb);
}
taosTmrReset(streamTriggerByTimer, (int32_t)pTask->triggerParam, pTask, streamEnv.timer, &pTask->timer);
}
int32_t streamSetupTrigger(SStreamTask* pTask) {
if (pTask->triggerParam != 0) {
if (streamInit() < 0) {
return -1;
}
pTask->timer = taosTmrStart(streamTriggerByTimer, (int32_t)pTask->triggerParam, pTask, streamEnv.timer);
pTask->triggerStatus = TASK_TRIGGER_STATUS__IN_ACTIVE;
}
return 0;
}
int32_t streamLaunchByWrite(SStreamTask* pTask, int32_t vgId, SMsgCb* pMsgCb) {
int8_t execStatus = atomic_load_8(&pTask->status); int8_t execStatus = atomic_load_8(&pTask->status);
if (execStatus == TASK_STATUS__IDLE || execStatus == TASK_STATUS__CLOSING) { if (execStatus == TASK_STATUS__IDLE || execStatus == TASK_STATUS__CLOSING) {
SStreamTaskRunReq* pRunReq = rpcMallocCont(sizeof(SStreamTaskRunReq)); SStreamTaskRunReq* pRunReq = rpcMallocCont(sizeof(SStreamTaskRunReq));

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

@ -20,15 +20,17 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, void* data, SArray* pRes)
void* exec = pTask->exec.executor; void* exec = pTask->exec.executor;
// set input // set input
if (pTask->inputType == STREAM_INPUT__DATA_SUBMIT) { SStreamQueueItem* pItem = (SStreamQueueItem*)data;
if (pItem->type == STREAM_INPUT__TRIGGER) {
SStreamTrigger* pTrigger = (SStreamTrigger*)data;
qSetMultiStreamInput(exec, pTrigger->pBlock, 1, STREAM_DATA_TYPE_SSDATA_BLOCK, false);
} else if (pItem->type == STREAM_INPUT__DATA_SUBMIT) {
SStreamDataSubmit* pSubmit = (SStreamDataSubmit*)data; SStreamDataSubmit* pSubmit = (SStreamDataSubmit*)data;
ASSERT(pSubmit->type == STREAM_INPUT__DATA_SUBMIT); ASSERT(pTask->inputType == STREAM_INPUT__DATA_SUBMIT);
qSetStreamInput(exec, pSubmit->data, STREAM_DATA_TYPE_SUBMIT_BLOCK, false); qSetStreamInput(exec, pSubmit->data, STREAM_DATA_TYPE_SUBMIT_BLOCK, false);
} else if (pTask->inputType == STREAM_INPUT__DATA_BLOCK) { } else if (pItem->type == STREAM_INPUT__DATA_BLOCK) {
SStreamDataBlock* pBlock = (SStreamDataBlock*)data; SStreamDataBlock* pBlock = (SStreamDataBlock*)data;
ASSERT(pBlock->type == STREAM_INPUT__DATA_BLOCK); ASSERT(pTask->inputType == STREAM_INPUT__DATA_BLOCK);
SArray* blocks = pBlock->blocks; SArray* blocks = pBlock->blocks;
qSetMultiStreamInput(exec, blocks->pData, blocks->size, STREAM_DATA_TYPE_SSDATA_BLOCK, false); qSetMultiStreamInput(exec, blocks->pData, blocks->size, STREAM_DATA_TYPE_SSDATA_BLOCK, false);
} }

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

@ -72,6 +72,7 @@ int32_t tEncodeSStreamTask(SEncoder* pEncoder, const SStreamTask* pTask) {
if (tSerializeSUseDbRspImp(pEncoder, &pTask->shuffleDispatcher.dbInfo) < 0) return -1; if (tSerializeSUseDbRspImp(pEncoder, &pTask->shuffleDispatcher.dbInfo) < 0) return -1;
/*if (tEncodeI8(pEncoder, pTask->shuffleDispatcher.hashMethod) < 0) return -1;*/ /*if (tEncodeI8(pEncoder, pTask->shuffleDispatcher.hashMethod) < 0) return -1;*/
} }
if (tEncodeI64(pEncoder, pTask->triggerParam) < 0) return -1;
/*tEndEncode(pEncoder);*/ /*tEndEncode(pEncoder);*/
return pEncoder->pos; return pEncoder->pos;
@ -121,6 +122,7 @@ int32_t tDecodeSStreamTask(SDecoder* pDecoder, SStreamTask* pTask) {
/*if (tDecodeI8(pDecoder, &pTask->shuffleDispatcher.hashMethod) < 0) return -1;*/ /*if (tDecodeI8(pDecoder, &pTask->shuffleDispatcher.hashMethod) < 0) return -1;*/
if (tDeserializeSUseDbRspImp(pDecoder, &pTask->shuffleDispatcher.dbInfo) < 0) return -1; if (tDeserializeSUseDbRspImp(pDecoder, &pTask->shuffleDispatcher.dbInfo) < 0) return -1;
} }
if (tDecodeI64(pDecoder, &pTask->triggerParam) < 0) return -1;
/*tEndDecode(pDecoder);*/ /*tEndDecode(pDecoder);*/
return 0; return 0;

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

@ -149,12 +149,14 @@ void syncStop(int64_t rid) {
int32_t syncSetStandby(int64_t rid) { int32_t syncSetStandby(int64_t rid) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return TAOS_SYNC_OTHER_ERROR; terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1;
} }
if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) { if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
taosReleaseRef(tsNodeRefId, pSyncNode->rid); taosReleaseRef(tsNodeRefId, pSyncNode->rid);
return TAOS_SYNC_OTHER_ERROR; terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1;
} }
// state change // state change
@ -177,7 +179,8 @@ int32_t syncSetStandby(int64_t rid) {
int32_t syncReconfigBuild(int64_t rid, const SSyncCfg* pNewCfg, SRpcMsg* pRpcMsg) { int32_t syncReconfigBuild(int64_t rid, const SSyncCfg* pNewCfg, SRpcMsg* pRpcMsg) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return TAOS_SYNC_OTHER_ERROR; terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1;
} }
ASSERT(rid == pSyncNode->rid); ASSERT(rid == pSyncNode->rid);
@ -201,7 +204,8 @@ int32_t syncReconfigBuild(int64_t rid, const SSyncCfg* pNewCfg, SRpcMsg* pRpcMsg
if (!IamInNew) { if (!IamInNew) {
taosReleaseRef(tsNodeRefId, pSyncNode->rid); taosReleaseRef(tsNodeRefId, pSyncNode->rid);
return TAOS_SYNC_NOT_IN_NEW_CONFIG; terrno = TSDB_CODE_SYN_NOT_IN_NEW_CONFIG;
return -1;
} }
char* newconfig = syncCfg2Str((SSyncCfg*)pNewCfg); char* newconfig = syncCfg2Str((SSyncCfg*)pNewCfg);
@ -219,7 +223,8 @@ int32_t syncReconfigBuild(int64_t rid, const SSyncCfg* pNewCfg, SRpcMsg* pRpcMsg
int32_t syncReconfig(int64_t rid, const SSyncCfg* pNewCfg) { int32_t syncReconfig(int64_t rid, const SSyncCfg* pNewCfg) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return TAOS_SYNC_OTHER_ERROR; terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1;
} }
ASSERT(rid == pSyncNode->rid); ASSERT(rid == pSyncNode->rid);
@ -246,7 +251,8 @@ int32_t syncReconfig(int64_t rid, const SSyncCfg* pNewCfg) {
if (!IamInNew) { if (!IamInNew) {
sError("sync reconfig error, not in new config"); sError("sync reconfig error, not in new config");
taosReleaseRef(tsNodeRefId, pSyncNode->rid); taosReleaseRef(tsNodeRefId, pSyncNode->rid);
return TAOS_SYNC_NOT_IN_NEW_CONFIG; terrno = TSDB_CODE_SYN_NOT_IN_NEW_CONFIG;
return -1;
} }
char* newconfig = syncCfg2Str((SSyncCfg*)pNewCfg); char* newconfig = syncCfg2Str((SSyncCfg*)pNewCfg);
@ -272,13 +278,15 @@ int32_t syncReconfig(int64_t rid, const SSyncCfg* pNewCfg) {
int32_t syncLeaderTransfer(int64_t rid) { int32_t syncLeaderTransfer(int64_t rid) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return TAOS_SYNC_OTHER_ERROR; terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1;
} }
ASSERT(rid == pSyncNode->rid); ASSERT(rid == pSyncNode->rid);
if (pSyncNode->peersNum == 0) { if (pSyncNode->peersNum == 0) {
taosReleaseRef(tsNodeRefId, pSyncNode->rid); taosReleaseRef(tsNodeRefId, pSyncNode->rid);
return TAOS_SYNC_OTHER_ERROR; terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1;
} }
SNodeInfo newLeader = (pSyncNode->peersNodeInfo)[0]; SNodeInfo newLeader = (pSyncNode->peersNodeInfo)[0];
@ -291,7 +299,8 @@ int32_t syncLeaderTransfer(int64_t rid) {
int32_t syncLeaderTransferTo(int64_t rid, SNodeInfo newLeader) { int32_t syncLeaderTransferTo(int64_t rid, SNodeInfo newLeader) {
SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return TAOS_SYNC_OTHER_ERROR; terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1;
} }
ASSERT(rid == pSyncNode->rid); ASSERT(rid == pSyncNode->rid);
int32_t ret = 0; int32_t ret = 0;
@ -299,7 +308,8 @@ int32_t syncLeaderTransferTo(int64_t rid, SNodeInfo newLeader) {
if (pSyncNode->replicaNum == 1) { if (pSyncNode->replicaNum == 1) {
sError("only one replica, cannot drop leader"); sError("only one replica, cannot drop leader");
taosReleaseRef(tsNodeRefId, pSyncNode->rid); taosReleaseRef(tsNodeRefId, pSyncNode->rid);
return TAOS_SYNC_ONLY_ONE_REPLICA; terrno = TSDB_CODE_SYN_ONE_REPLICA;
return -1;
} }
SyncLeaderTransfer* pMsg = syncLeaderTransferBuild(pSyncNode->vgId); SyncLeaderTransfer* pMsg = syncLeaderTransferBuild(pSyncNode->vgId);
@ -538,11 +548,12 @@ void setHeartbeatTimerMS(int64_t rid, int32_t hbTimerMS) {
} }
int32_t syncPropose(int64_t rid, const SRpcMsg* pMsg, bool isWeak) { int32_t syncPropose(int64_t rid, const SRpcMsg* pMsg, bool isWeak) {
int32_t ret = TAOS_SYNC_PROPOSE_SUCCESS; int32_t ret = 0;
SSyncNode* pSyncNode = taosAcquireRef(tsNodeRefId, rid); SSyncNode* pSyncNode = taosAcquireRef(tsNodeRefId, rid);
if (pSyncNode == NULL) { if (pSyncNode == NULL) {
return TAOS_SYNC_OTHER_ERROR; terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1;
} }
assert(rid == pSyncNode->rid); assert(rid == pSyncNode->rid);
sDebug("vgId:%d sync event propose msgType:%s", pSyncNode->vgId, TMSG_INFO(pMsg->msgType)); sDebug("vgId:%d sync event propose msgType:%s", pSyncNode->vgId, TMSG_INFO(pMsg->msgType));
@ -553,7 +564,7 @@ int32_t syncPropose(int64_t rid, const SRpcMsg* pMsg, bool isWeak) {
} }
int32_t syncNodePropose(SSyncNode* pSyncNode, const SRpcMsg* pMsg, bool isWeak) { int32_t syncNodePropose(SSyncNode* pSyncNode, const SRpcMsg* pMsg, bool isWeak) {
int32_t ret = TAOS_SYNC_PROPOSE_SUCCESS; int32_t ret = 0;
sDebug("vgId:%d sync event propose msgType:%s", pSyncNode->vgId, TMSG_INFO(pMsg->msgType)); sDebug("vgId:%d sync event propose msgType:%s", pSyncNode->vgId, TMSG_INFO(pMsg->msgType));
if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) { if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
@ -567,14 +578,17 @@ int32_t syncNodePropose(SSyncNode* pSyncNode, const SRpcMsg* pMsg, bool isWeak)
syncClientRequest2RpcMsg(pSyncMsg, &rpcMsg); syncClientRequest2RpcMsg(pSyncMsg, &rpcMsg);
if (pSyncNode->FpEqMsg != NULL && (*pSyncNode->FpEqMsg)(pSyncNode->msgcb, &rpcMsg) == 0) { if (pSyncNode->FpEqMsg != NULL && (*pSyncNode->FpEqMsg)(pSyncNode->msgcb, &rpcMsg) == 0) {
ret = TAOS_SYNC_PROPOSE_SUCCESS; ret = 0;
} else { } else {
ret = -1;
terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
sError("syncPropose pSyncNode->FpEqMsg is NULL"); sError("syncPropose pSyncNode->FpEqMsg is NULL");
} }
syncClientRequestDestroy(pSyncMsg); syncClientRequestDestroy(pSyncMsg);
} else { } else {
ret = -1;
terrno = TSDB_CODE_SYN_NOT_LEADER;
sError("syncPropose not leader, %s", syncUtilState2String(pSyncNode->state)); sError("syncPropose not leader, %s", syncUtilState2String(pSyncNode->state));
ret = TAOS_SYNC_PROPOSE_NOT_LEADER;
} }
return ret; return ret;
@ -945,9 +959,13 @@ int32_t syncNodePingAll(SSyncNode* pSyncNode) {
// timer control -------------- // timer control --------------
int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode) { int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode) {
int32_t ret = 0; int32_t ret = 0;
taosTmrReset(pSyncNode->FpPingTimerCB, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager, if (syncEnvIsStart()) {
&pSyncNode->pPingTimer); taosTmrReset(pSyncNode->FpPingTimerCB, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager,
atomic_store_64(&pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser); &pSyncNode->pPingTimer);
atomic_store_64(&pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser);
} else {
sError("sync env is stop, syncNodeStartPingTimer");
}
return ret; return ret;
} }
@ -961,10 +979,14 @@ int32_t syncNodeStopPingTimer(SSyncNode* pSyncNode) {
int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms) { int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms) {
int32_t ret = 0; int32_t ret = 0;
pSyncNode->electTimerMS = ms; if (syncEnvIsStart()) {
taosTmrReset(pSyncNode->FpElectTimerCB, pSyncNode->electTimerMS, pSyncNode, gSyncEnv->pTimerManager, pSyncNode->electTimerMS = ms;
&pSyncNode->pElectTimer); taosTmrReset(pSyncNode->FpElectTimerCB, pSyncNode->electTimerMS, pSyncNode, gSyncEnv->pTimerManager,
atomic_store_64(&pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser); &pSyncNode->pElectTimer);
atomic_store_64(&pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser);
} else {
sError("sync env is stop, syncNodeStartElectTimer");
}
return ret; return ret;
} }
@ -998,9 +1020,13 @@ int32_t syncNodeResetElectTimer(SSyncNode* pSyncNode) {
int32_t syncNodeStartHeartbeatTimer(SSyncNode* pSyncNode) { int32_t syncNodeStartHeartbeatTimer(SSyncNode* pSyncNode) {
int32_t ret = 0; int32_t ret = 0;
taosTmrReset(pSyncNode->FpHeartbeatTimerCB, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager, if (syncEnvIsStart()) {
&pSyncNode->pHeartbeatTimer); taosTmrReset(pSyncNode->FpHeartbeatTimerCB, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager,
atomic_store_64(&pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser); &pSyncNode->pHeartbeatTimer);
atomic_store_64(&pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser);
} else {
sError("sync env is stop, syncNodeStartHeartbeatTimer");
}
return ret; return ret;
} }
@ -1720,14 +1746,25 @@ static void syncNodeEqPingTimer(void* param, void* tmrId) {
syncTimeout2RpcMsg(pSyncMsg, &rpcMsg); syncTimeout2RpcMsg(pSyncMsg, &rpcMsg);
syncRpcMsgLog2((char*)"==syncNodeEqPingTimer==", &rpcMsg); syncRpcMsgLog2((char*)"==syncNodeEqPingTimer==", &rpcMsg);
if (pSyncNode->FpEqMsg != NULL) { if (pSyncNode->FpEqMsg != NULL) {
pSyncNode->FpEqMsg(pSyncNode->msgcb, &rpcMsg); int32_t code = pSyncNode->FpEqMsg(pSyncNode->msgcb, &rpcMsg);
if (code != 0) {
sError("vgId:%d sync enqueue ping msg error, code:%d", pSyncNode->vgId, code);
rpcFreeCont(rpcMsg.pCont);
syncTimeoutDestroy(pSyncMsg);
return;
}
} else { } else {
sTrace("syncNodeEqPingTimer pSyncNode->FpEqMsg is NULL"); sTrace("syncNodeEqPingTimer pSyncNode->FpEqMsg is NULL");
} }
syncTimeoutDestroy(pSyncMsg); syncTimeoutDestroy(pSyncMsg);
taosTmrReset(syncNodeEqPingTimer, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager, if (syncEnvIsStart()) {
&pSyncNode->pPingTimer); taosTmrReset(syncNodeEqPingTimer, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager,
&pSyncNode->pPingTimer);
} else {
sError("sync env is stop, syncNodeEqPingTimer");
}
} else { } else {
sTrace("==syncNodeEqPingTimer== pingTimerLogicClock:%" PRIu64 ", pingTimerLogicClockUser:%" PRIu64 "", sTrace("==syncNodeEqPingTimer== pingTimerLogicClock:%" PRIu64 ", pingTimerLogicClockUser:%" PRIu64 "",
pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser); pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser);
@ -1743,16 +1780,26 @@ static void syncNodeEqElectTimer(void* param, void* tmrId) {
syncTimeout2RpcMsg(pSyncMsg, &rpcMsg); syncTimeout2RpcMsg(pSyncMsg, &rpcMsg);
syncRpcMsgLog2((char*)"==syncNodeEqElectTimer==", &rpcMsg); syncRpcMsgLog2((char*)"==syncNodeEqElectTimer==", &rpcMsg);
if (pSyncNode->FpEqMsg != NULL) { if (pSyncNode->FpEqMsg != NULL) {
pSyncNode->FpEqMsg(pSyncNode->msgcb, &rpcMsg); int32_t code = pSyncNode->FpEqMsg(pSyncNode->msgcb, &rpcMsg);
if (code != 0) {
sError("vgId:%d sync enqueue elect msg error, code:%d", pSyncNode->vgId, code);
rpcFreeCont(rpcMsg.pCont);
syncTimeoutDestroy(pSyncMsg);
return;
}
} else { } else {
sTrace("syncNodeEqElectTimer pSyncNode->FpEqMsg is NULL"); sTrace("syncNodeEqElectTimer FpEqMsg is NULL");
} }
syncTimeoutDestroy(pSyncMsg); syncTimeoutDestroy(pSyncMsg);
// reset timer ms // reset timer ms
pSyncNode->electTimerMS = syncUtilElectRandomMS(pSyncNode->electBaseLine, 2 * pSyncNode->electBaseLine); if (syncEnvIsStart()) {
taosTmrReset(syncNodeEqElectTimer, pSyncNode->electTimerMS, pSyncNode, gSyncEnv->pTimerManager, pSyncNode->electTimerMS = syncUtilElectRandomMS(pSyncNode->electBaseLine, 2 * pSyncNode->electBaseLine);
&pSyncNode->pElectTimer); taosTmrReset(syncNodeEqElectTimer, pSyncNode->electTimerMS, pSyncNode, gSyncEnv->pTimerManager,
&pSyncNode->pElectTimer);
} else {
sError("sync env is stop, syncNodeEqElectTimer");
}
} else { } else {
sTrace("==syncNodeEqElectTimer== electTimerLogicClock:%" PRIu64 ", electTimerLogicClockUser:%" PRIu64 "", sTrace("==syncNodeEqElectTimer== electTimerLogicClock:%" PRIu64 ", electTimerLogicClockUser:%" PRIu64 "",
pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser); pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser);
@ -1774,19 +1821,19 @@ static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {
if (code != 0) { if (code != 0) {
sError("vgId:%d sync enqueue timer msg error, code:%d", pSyncNode->vgId, code); sError("vgId:%d sync enqueue timer msg error, code:%d", pSyncNode->vgId, code);
rpcFreeCont(rpcMsg.pCont); rpcFreeCont(rpcMsg.pCont);
syncTimeoutDestroy(pSyncMsg);
return; return;
} }
} else { } else {
sTrace("syncNodeEqHeartbeatTimer pSyncNode->FpEqMsg is NULL"); sError("syncNodeEqHeartbeatTimer FpEqMsg is NULL");
} }
syncTimeoutDestroy(pSyncMsg); syncTimeoutDestroy(pSyncMsg);
if (gSyncEnv != NULL) { if (syncEnvIsStart()) {
taosTmrReset(syncNodeEqHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager, taosTmrReset(syncNodeEqHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager,
&pSyncNode->pHeartbeatTimer); &pSyncNode->pHeartbeatTimer);
} else { } else {
sError("sync env is already stop"); sError("sync env is stop, syncNodeEqHeartbeatTimer");
} }
} else { } else {
sTrace("==syncNodeEqHeartbeatTimer== heartbeatTimerLogicClock:%" PRIu64 ", heartbeatTimerLogicClockUser:%" PRIu64 sTrace("==syncNodeEqHeartbeatTimer== heartbeatTimerLogicClock:%" PRIu64 ", heartbeatTimerLogicClockUser:%" PRIu64

2
source/libs/sync/test/syncConfigChangeSnapshotTest.cpp
View File

@ -338,7 +338,7 @@ int main(int argc, char** argv) {
if (alreadySend < writeRecordNum) { if (alreadySend < writeRecordNum) {
SRpcMsg* pRpcMsg = createRpcMsg(alreadySend, writeRecordNum, myIndex); SRpcMsg* pRpcMsg = createRpcMsg(alreadySend, writeRecordNum, myIndex);
int32_t ret = syncPropose(rid, pRpcMsg, false); int32_t ret = syncPropose(rid, pRpcMsg, false);
if (ret == TAOS_SYNC_PROPOSE_NOT_LEADER) { if (ret == -1 && terrno == TSDB_CODE_SYN_NOT_LEADER) {
sTrace("%s value%d write not leader", s, alreadySend); sTrace("%s value%d write not leader", s, alreadySend);
} else { } else {
assert(ret == 0); assert(ret == 0);

2
source/libs/sync/test/syncConfigChangeTest.cpp
View File

@ -251,7 +251,7 @@ int main(int argc, char** argv) {
if (alreadySend < writeRecordNum) { if (alreadySend < writeRecordNum) {
SRpcMsg* pRpcMsg = createRpcMsg(alreadySend, writeRecordNum, myIndex); SRpcMsg* pRpcMsg = createRpcMsg(alreadySend, writeRecordNum, myIndex);
int32_t ret = syncPropose(rid, pRpcMsg, false); int32_t ret = syncPropose(rid, pRpcMsg, false);
if (ret == TAOS_SYNC_PROPOSE_NOT_LEADER) { if (ret == -1 && terrno == TSDB_CODE_SYN_NOT_LEADER) {
sTrace("%s value%d write not leader", s, alreadySend); sTrace("%s value%d write not leader", s, alreadySend);
} else { } else {
assert(ret == 0); assert(ret == 0);

2
source/libs/sync/test/syncReplicateTest.cpp
View File

@ -188,7 +188,7 @@ int main(int argc, char** argv) {
if (alreadySend < writeRecordNum) { if (alreadySend < writeRecordNum) {
SRpcMsg* pRpcMsg = createRpcMsg(alreadySend, writeRecordNum, myIndex); SRpcMsg* pRpcMsg = createRpcMsg(alreadySend, writeRecordNum, myIndex);
int32_t ret = syncPropose(rid, pRpcMsg, false); int32_t ret = syncPropose(rid, pRpcMsg, false);
if (ret == TAOS_SYNC_PROPOSE_NOT_LEADER) { if (ret == -1 && terrno == TSDB_CODE_SYN_NOT_LEADER) {
sTrace("%s value%d write not leader", s, alreadySend); sTrace("%s value%d write not leader", s, alreadySend);
} else { } else {
assert(ret == 0); assert(ret == 0);

2
source/libs/sync/test/syncTestTool.cpp
View File

@ -391,7 +391,7 @@ int main(int argc, char** argv) {
if (alreadySend < writeRecordNum) { if (alreadySend < writeRecordNum) {
SRpcMsg* pRpcMsg = createRpcMsg(alreadySend, writeRecordNum, myIndex); SRpcMsg* pRpcMsg = createRpcMsg(alreadySend, writeRecordNum, myIndex);
int32_t ret = syncPropose(rid, pRpcMsg, false); int32_t ret = syncPropose(rid, pRpcMsg, false);
if (ret == TAOS_SYNC_PROPOSE_NOT_LEADER) { if (ret == -1 && terrno == TSDB_CODE_SYN_NOT_LEADER) {
sTrace("%s value%d write not leader, leaderTransferWait:%d", simpleStr, alreadySend, leaderTransferWait); sTrace("%s value%d write not leader, leaderTransferWait:%d", simpleStr, alreadySend, leaderTransferWait);
} else { } else {
assert(ret == 0); assert(ret == 0);

6
source/util/src/tarray.c
View File

@ -174,7 +174,11 @@ void taosArrayRemoveDuplicate(SArray* pArray, __compar_fn_t comparFn, void (*fp)
} }
void* taosArrayAddAll(SArray* pArray, const SArray* pInput) { void* taosArrayAddAll(SArray* pArray, const SArray* pInput) {
return taosArrayAddBatch(pArray, pInput->pData, (int32_t)taosArrayGetSize(pInput)); if (pInput) {
return taosArrayAddBatch(pArray, pInput->pData, (int32_t)taosArrayGetSize(pInput));
} else {
return NULL;
}
} }
void* taosArrayPop(SArray* pArray) { void* taosArrayPop(SArray* pArray) {

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

@ -413,6 +413,8 @@ TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INVALID_MSGLEN, "Invalid msg length")
TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INVALID_MSGTYPE, "Invalid msg type") TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INVALID_MSGTYPE, "Invalid msg type")
TAOS_DEFINE_ERROR(TSDB_CODE_SYN_NOT_LEADER, "Sync not leader") TAOS_DEFINE_ERROR(TSDB_CODE_SYN_NOT_LEADER, "Sync not leader")
TAOS_DEFINE_ERROR(TSDB_CODE_SYN_ONE_REPLICA, "Sync one replica")
TAOS_DEFINE_ERROR(TSDB_CODE_SYN_NOT_IN_NEW_CONFIG, "Sync not in new config")
TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INTERNAL_ERROR, "Sync internal error") TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INTERNAL_ERROR, "Sync internal error")
// wal // wal

65
tests/system-test/2-query/json_tag.py
View File

@ -243,9 +243,8 @@ class TDTestCase:
tdSql.checkRows(2) tdSql.checkRows(2)
tdSql.query("select * from jsons1 where jtag->'tag2'!='beijing'") tdSql.query("select * from jsons1 where jtag->'tag2'!='beijing'")
tdSql.checkRows(5) tdSql.checkRows(5)
#open tdSql.query("select * from jsons1 where jtag->'tag2'=''")
#tdSql.query("select * from jsons1 where jtag->'tag2'=''") tdSql.checkRows(2)
#tdSql.checkRows(2)
# #
# # where json value is int # # where json value is int
tdSql.query("select * from jsons1 where jtag->'tag1'=5") tdSql.query("select * from jsons1 where jtag->'tag1'=5")
@ -253,11 +252,10 @@ class TDTestCase:
tdSql.checkData(0, 1, 2) tdSql.checkData(0, 1, 2)
tdSql.query("select * from jsons1 where jtag->'tag1'=10") tdSql.query("select * from jsons1 where jtag->'tag1'=10")
tdSql.checkRows(0) tdSql.checkRows(0)
# open tdSql.query("select * from jsons1 where jtag->'tag1'<54")
#tdSql.query("select * from jsons1 where jtag->'tag1'<54") tdSql.checkRows(4)
#tdSql.checkRows(3) tdSql.query("select * from jsons1 where jtag->'tag1'<=11")
#tdSql.query("select * from jsons1 where jtag->'tag1'<=11") tdSql.checkRows(4)
#tdSql.checkRows(3)
tdSql.query("select * from jsons1 where jtag->'tag1'>4") tdSql.query("select * from jsons1 where jtag->'tag1'>4")
tdSql.checkRows(2) tdSql.checkRows(2)
tdSql.query("select * from jsons1 where jtag->'tag1'>=5") tdSql.query("select * from jsons1 where jtag->'tag1'>=5")
@ -270,31 +268,28 @@ class TDTestCase:
# # where json value is double # # where json value is double
tdSql.query("select * from jsons1 where jtag->'tag1'=1.232") tdSql.query("select * from jsons1 where jtag->'tag1'=1.232")
tdSql.checkRows(1) tdSql.checkRows(1)
# open tdSql.query("select * from jsons1 where jtag->'tag1'<1.232")
#tdSql.query("select * from jsons1 where jtag->'tag1'<1.232") tdSql.checkRows(1)
#tdSql.checkRows(0) tdSql.query("select * from jsons1 where jtag->'tag1'<=1.232")
#tdSql.query("select * from jsons1 where jtag->'tag1'<=1.232") tdSql.checkRows(2)
#tdSql.checkRows(1)
tdSql.query("select * from jsons1 where jtag->'tag1'>1.23") tdSql.query("select * from jsons1 where jtag->'tag1'>1.23")
tdSql.checkRows(3) tdSql.checkRows(3)
tdSql.query("select * from jsons1 where jtag->'tag1'>=1.232") tdSql.query("select * from jsons1 where jtag->'tag1'>=1.232")
tdSql.checkRows(3) tdSql.checkRows(3)
# open tdSql.query("select * from jsons1 where jtag->'tag1'!=1.232")
#tdSql.query("select * from jsons1 where jtag->'tag1'!=1.232") tdSql.checkRows(6)
#tdSql.checkRows(2)
tdSql.query("select * from jsons1 where jtag->'tag1'!=3.232") tdSql.query("select * from jsons1 where jtag->'tag1'!=3.232")
tdSql.checkRows(7) tdSql.checkRows(7)
#tdSql.error("select * from jsons1 where jtag->'tag1'/0=3") #tdSql.error("select * from jsons1 where jtag->'tag1'/0=3")
#tdSql.error("select * from jsons1 where jtag->'tag1'/5=1") #tdSql.error("select * from jsons1 where jtag->'tag1'/5=1")
# #
# # where json value is bool # # where json value is bool
#tdSql.query("select * from jsons1 where jtag->'tag1'=true") tdSql.query("select * from jsons1 where jtag->'tag1'=true")
# open tdSql.checkRows(0)
#tdSql.checkRows(0)
#tdSql.query("select * from jsons1 where jtag->'tag1'=false") #tdSql.query("select * from jsons1 where jtag->'tag1'=false")
#tdSql.checkRows(1) #tdSql.checkRows(1)
#tdSql.query("select * from jsons1 where jtag->'tag1'!=false") tdSql.query("select * from jsons1 where jtag->'tag1'!=false")
#tdSql.checkRows(0) tdSql.checkRows(3)
#tdSql.error("select * from jsons1 where jtag->'tag1'>false") #tdSql.error("select * from jsons1 where jtag->'tag1'>false")
# #
# # where json value is null # # where json value is null
@ -303,18 +298,17 @@ class TDTestCase:
#tdSql.checkRows(1) #tdSql.checkRows(1)
# #
# # where json key is null # # where json key is null
# open tdSql.query("select * from jsons1 where jtag->'tag_no_exist'=3")
#tdSql.query("select * from jsons1 where jtag->'tag_no_exist'=3") tdSql.checkRows(0)
#tdSql.checkRows(0)
# #
# # where json value is not exist # # where json value is not exist
#tdSql.query("select * from jsons1 where jtag->'tag1' is null") tdSql.query("select * from jsons1 where jtag->'tag1' is null")
#tdSql.checkData(0, 0, 'jsons1_9') tdSql.checkData(0, 0, 'jsons1_9')
#tdSql.checkRows(1) tdSql.checkRows(2)
#tdSql.query("select * from jsons1 where jtag->'tag4' is null") tdSql.query("select * from jsons1 where jtag->'tag4' is null")
#tdSql.checkRows(9) tdSql.checkRows(9)
#tdSql.query("select * from jsons1 where jtag->'tag3' is not null") tdSql.query("select * from jsons1 where jtag->'tag3' is not null")
#tdSql.checkRows(4) tdSql.checkRows(3)
# #
# # test contains # # test contains
tdSql.query("select * from jsons1 where jtag contains 'tag1'") tdSql.query("select * from jsons1 where jtag contains 'tag1'")
@ -344,10 +338,10 @@ class TDTestCase:
# #
# #
# # test with between and # # test with between and
#tdSql.query("select * from jsons1 where jtag->'tag1' between 1 and 30") tdSql.query("select * from jsons1 where jtag->'tag1' between 1 and 30")
#tdSql.checkRows(3) tdSql.checkRows(3)
#tdSql.query("select * from jsons1 where jtag->'tag1' between 'femail' and 'beijing'") tdSql.query("select * from jsons1 where jtag->'tag1' between 'femail' and 'beijing'")
#tdSql.checkRows(2) tdSql.checkRows(2)
# #
# # test with tbname/normal column # # test with tbname/normal column
tdSql.query("select * from jsons1 where tbname = 'jsons1_1'") tdSql.query("select * from jsons1 where tbname = 'jsons1_1'")
@ -362,6 +356,7 @@ class TDTestCase:
# #
# # test where condition like # # test where condition like
# open # open
# syntax error
#tdSql.query("select *,tbname from jsons1 where jtag->'tag2' like 'bei%'") #tdSql.query("select *,tbname from jsons1 where jtag->'tag2' like 'bei%'")
#tdSql.checkRows(2) #tdSql.checkRows(2)
#tdSql.query("select *,tbname from jsons1 where jtag->'tag1' like 'fe%' and jtag->'tag2' is not null") #tdSql.query("select *,tbname from jsons1 where jtag->'tag1' like 'fe%' and jtag->'tag2' is not null")

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

@ -134,7 +134,7 @@ class TDTestCase:
parameterDict['cfg'] = cfgPath parameterDict['cfg'] = cfgPath
prepareEnvThread = threading.Thread(target=self.prepareEnv, kwargs=parameterDict) prepareEnvThread = threading.Thread(target=self.prepareEnv, kwargs=parameterDict)
prepareEnvThread.start() prepareEnvThread.start()
time.sleep(2) prepareEnvThread.join()
# wait stb ready # wait stb ready
while 1: while 1:
@ -245,6 +245,7 @@ class TDTestCase:
prepareEnvThread = threading.Thread(target=self.prepareEnv, kwargs=parameterDict) prepareEnvThread = threading.Thread(target=self.prepareEnv, kwargs=parameterDict)
prepareEnvThread.start() prepareEnvThread.start()
prepareEnvThread.join()
# wait db ready # wait db ready
while 1: while 1:
@ -371,6 +372,7 @@ class TDTestCase:
prepareEnvThread = threading.Thread(target=self.prepareEnv, kwargs=parameterDict) prepareEnvThread = threading.Thread(target=self.prepareEnv, kwargs=parameterDict)
prepareEnvThread.start() prepareEnvThread.start()
prepareEnvThread.join()
# wait db ready # wait db ready
while 1: while 1:

5
tests/system-test/test-all.bat
View File

@ -61,7 +61,8 @@ goto :eof
set tt=%1 set tt=%1
set tt=%tt:.= % set tt=%tt:.= %
set tt=%tt::= % set tt=%tt::= %
set index=1 set tt=%tt: 0= %
set /a index=1
for %%a in (%tt%) do ( for %%a in (%tt%) do (
if !index! EQU 1 ( if !index! EQU 1 (
set /a hh=%%a set /a hh=%%a
@ -75,5 +76,5 @@ for %%a in (%tt%) do (
) )
set /a index=index+1 set /a index=index+1
) )
set /a _timeTemp=(%hh%*60+%mm%)*60+%ss% || echo hh:%hh% mm:%mm% ss:%ss% set /a _timeTemp=(%hh%*60+%mm%)*60+%ss%
goto :eof goto :eof

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

@ -21,6 +21,7 @@ import base64
import json import json
import platform import platform
import socket import socket
import threading
from distutils.log import warn as printf from distutils.log import warn as printf
from fabric2 import Connection from fabric2 import Connection
sys.path.append("../pytest") sys.path.append("../pytest")
@ -30,6 +31,13 @@ from util.cases import *
import taos import taos
def checkRunTimeError():
import win32gui
while 1:
time.sleep(1)
hwnd = win32gui.FindWindow(None, "Microsoft Visual C++ Runtime Library")
if hwnd:
os.system("TASKKILL /F /IM taosd.exe")
if __name__ == "__main__": if __name__ == "__main__":
@ -42,9 +50,6 @@ if __name__ == "__main__":
logSql = True logSql = True
stop = 0 stop = 0
restart = False restart = False
windows = 0
if platform.system().lower() == 'windows':
windows = 1
updateCfgDict = {} updateCfgDict = {}
execCmd = "" execCmd = ""
opts, args = getopt.gnu_getopt(sys.argv[1:], 'f:p:m:l:scghrd:k:e:', [ opts, args = getopt.gnu_getopt(sys.argv[1:], 'f:p:m:l:scghrd:k:e:', [
@ -159,7 +164,9 @@ if __name__ == "__main__":
host = masterIp host = masterIp
tdLog.info("Procedures for tdengine deployed in %s" % (host)) tdLog.info("Procedures for tdengine deployed in %s" % (host))
if windows: if platform.system().lower() == 'windows':
if (masterIp == "" and not fileName[0:12] == "0-others\\udf"):
threading.Thread(target=checkRunTimeError,daemon=True).start()
tdCases.logSql(logSql) tdCases.logSql(logSql)
tdLog.info("Procedures for testing self-deployment") tdLog.info("Procedures for testing self-deployment")
tdDnodes.init(deployPath, masterIp) tdDnodes.init(deployPath, masterIp)