diff --git a/examples/c/stream_demo.c b/examples/c/stream_demo.c index 5a141867e7b56693eece323a8e039e7afe94b1dd..6d341c61c76c85df04bf9cc97bd6666c18c83fe8 100644 --- a/examples/c/stream_demo.c +++ b/examples/c/stream_demo.c @@ -88,9 +88,9 @@ int32_t create_stream() { /*const char* sql = "select min(k), max(k), sum(k) as sum_of_k from st1";*/ /*const char* sql = "select sum(k) from tu1 interval(10m)";*/ /*pRes = tmq_create_stream(pConn, "stream1", "out1", sql);*/ - pRes = taos_query(pConn, - "create stream stream1 trigger at_once into abc2.outstb as select _wstartts, sum(k) from st1 " - "partition by tbname interval(10m) "); + pRes = taos_query( + pConn, + "create stream stream1 trigger at_once into abc1.outstb as select _wstartts, sum(k) from st1 interval(10m) "); if (taos_errno(pRes) != 0) { printf("failed to create stream stream1, reason:%s\n", taos_errstr(pRes)); return -1; @@ -107,11 +107,4 @@ int main(int argc, char* argv[]) { code = init_env(); } create_stream(); -#if 0 - tmq_t* tmq = build_consumer(); - tmq_list_t* topic_list = build_topic_list(); - /*perf_loop(tmq, topic_list);*/ - /*basic_consume_loop(tmq, topic_list);*/ - sync_consume_loop(tmq, topic_list); -#endif } diff --git a/include/common/tmsgcb.h b/include/common/tmsgcb.h index e99377f9b4b27871506d1739520060b8caa51417..b56f7552666dd23650f9911dc23e03b9ae2fa707 100644 --- a/include/common/tmsgcb.h +++ b/include/common/tmsgcb.h @@ -34,7 +34,6 @@ typedef enum { WRITE_QUEUE, APPLY_QUEUE, SYNC_QUEUE, - MERGE_QUEUE, QUEUE_MAX, } EQueueType; diff --git a/include/dnode/snode/snode.h b/include/dnode/snode/snode.h index 611bff49f116c3a10363c607474ebb90eb275186..3d0ef2e052ef20978156b76b1dd7d552391b4a8b 100644 --- a/include/dnode/snode/snode.h +++ b/include/dnode/snode/snode.h @@ -16,8 +16,8 @@ #ifndef _TD_SNODE_H_ #define _TD_SNODE_H_ -#include "tmsgcb.h" #include "tmsg.h" +#include "tmsgcb.h" #include "trpc.h" #ifdef __cplusplus @@ -68,8 +68,8 @@ int32_t sndGetLoad(SSnode *pSnode, SSnodeLoad *pLoad); * @param pMsg The request message * @param pRsp The response message */ -void sndProcessUMsg(SSnode *pSnode, SRpcMsg *pMsg); -void sndProcessSMsg(SSnode *pSnode, SRpcMsg *pMsg); +int32_t sndProcessUMsg(SSnode *pSnode, SRpcMsg *pMsg); +int32_t sndProcessSMsg(SSnode *pSnode, SRpcMsg *pMsg); #ifdef __cplusplus } diff --git a/include/libs/planner/planner.h b/include/libs/planner/planner.h index 8ed95b60101472b347b10b285fcacaa3123cce57..c4f71e57a8174c62cf331e4afec35604786282a0 100644 --- a/include/libs/planner/planner.h +++ b/include/libs/planner/planner.h @@ -36,7 +36,6 @@ typedef struct SPlanContext { int64_t watermark; char* pMsg; int32_t msgLen; - // double filesFactor; } SPlanContext; // Create the physical plan for the query, according to the AST. diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h index 2c9d66a828e658cd286d2e5bf4edcc28d7ea37a9..2b3a1f265089bf3f10caee10458d035ac2f90169 100644 --- a/include/libs/stream/tstream.h +++ b/include/libs/stream/tstream.h @@ -152,7 +152,7 @@ void* streamDataBlockDecode(const void* buf, SStreamDataBlock* pInput); typedef struct { char* qmsg; // followings are not applicable to encoder and decoder - void* inputHandle; + // void* inputHandle; void* executor; } STaskExec; @@ -240,12 +240,13 @@ struct SStreamTask { int8_t inputType; int8_t status; - int8_t sourceType; int8_t execType; int8_t sinkType; int8_t dispatchType; int16_t dispatchMsgType; + int8_t dataScan; + // node info int32_t childId; int32_t nodeId; diff --git a/source/dnode/mgmt/mgmt_snode/src/smHandle.c b/source/dnode/mgmt/mgmt_snode/src/smHandle.c index 66ab627e3245678fe1b04ab9c8d2b8876022361a..52a69f95b47da1238e7c4b05c39ed53e867cf4ad 100644 --- a/source/dnode/mgmt/mgmt_snode/src/smHandle.c +++ b/source/dnode/mgmt/mgmt_snode/src/smHandle.c @@ -95,9 +95,12 @@ SArray *smGetMsgHandles() { if (dmSetMgmtHandle(pArray, TDMT_MON_SM_INFO, smPutNodeMsgToMonitorQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DEPLOY, smPutNodeMsgToMgmtQueue, 1) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TASK_DROP, smPutNodeMsgToMgmtQueue, 1) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RUN, smPutNodeMsgToMgmtQueue, 1) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DISPATCH, smPutNodeMsgToMgmtQueue, 1) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DISPATCH_RSP, smPutNodeMsgToMgmtQueue, 1) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RECOVER, smPutNodeMsgToMgmtQueue, 1) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RECOVER_RSP, smPutNodeMsgToMgmtQueue, 1) == NULL) goto _OVER; code = 0; _OVER: diff --git a/source/dnode/mgmt/mgmt_snode/src/smWorker.c b/source/dnode/mgmt/mgmt_snode/src/smWorker.c index 34a205232e69f057d879188fe4ed98df5b378e3a..8d93ddd66c7fb8aa27b550ebaa497739ae662545 100644 --- a/source/dnode/mgmt/mgmt_snode/src/smWorker.c +++ b/source/dnode/mgmt/mgmt_snode/src/smWorker.c @@ -55,7 +55,9 @@ static void smProcessUniqueQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t num taosGetQitem(qall, (void **)&pMsg); dTrace("msg:%p, get from snode-unique queue", pMsg); - sndProcessUMsg(pMgmt->pSnode, pMsg); + if (sndProcessUMsg(pMgmt->pSnode, pMsg) < 0) { + ASSERT(0); + } dTrace("msg:%p, is freed", pMsg); rpcFreeCont(pMsg->pCont); @@ -67,7 +69,9 @@ static void smProcessSharedQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) { SSnodeMgmt *pMgmt = pInfo->ahandle; dTrace("msg:%p, get from snode-shared queue", pMsg); - sndProcessSMsg(pMgmt->pSnode, pMsg); + if (sndProcessSMsg(pMgmt->pSnode, pMsg) < 0) { + ASSERT(0); + } dTrace("msg:%p, is freed", pMsg); rpcFreeCont(pMsg->pCont); diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c index 09fdd4b087f739dc29e86484a45f48f1af13d412..95dd5732c6a7ad1de2c0c8d3f753445163a6c3e1 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c @@ -170,10 +170,6 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp dTrace("vgId:%d, msg:%p put into vnode-sync queue", pVnode->vgId, pMsg); taosWriteQitem(pVnode->pSyncQ, pMsg); break; - case MERGE_QUEUE: - dTrace("vgId:%d, msg:%p put into vnode-merge queue", pVnode->vgId, pMsg); - taosWriteQitem(pVnode->pMergeQ, pMsg); - break; case APPLY_QUEUE: dTrace("vgId:%d, msg:%p put into vnode-apply queue", pVnode->vgId, pMsg); taosWriteQitem(pVnode->pApplyQ, pMsg); @@ -196,8 +192,6 @@ int32_t vmPutMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsg int32_t vmPutMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, FETCH_QUEUE); } -int32_t vmPutMsgToMergeQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, MERGE_QUEUE); } - int32_t vmPutMsgToMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { dTrace("msg:%p, put into vnode-mgmt queue", pMsg); taosWriteQitem(pMgmt->mgmtWorker.queue, pMsg); @@ -243,9 +237,6 @@ int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) { case FETCH_QUEUE: size = taosQueueItemSize(pVnode->pFetchQ); break; - case MERGE_QUEUE: - size = taosQueueItemSize(pVnode->pMergeQ); - break; default: break; } diff --git a/source/dnode/mnode/impl/src/mndScheduler.c b/source/dnode/mnode/impl/src/mndScheduler.c index 39bb6798aa869277bd6e510829d3049e0bd02b87..3ff0c39bc31a19f47aace0bac90c5aac25707342 100644 --- a/source/dnode/mnode/impl/src/mndScheduler.c +++ b/source/dnode/mnode/impl/src/mndScheduler.c @@ -63,9 +63,8 @@ int32_t mndConvertRsmaTask(char** pDst, int32_t* pDstLen, const char* ast, int64 .topicQuery = false, .streamQuery = true, .rSmaQuery = true, - .triggerType = STREAM_TRIGGER_AT_ONCE, + .triggerType = triggerType, .watermark = watermark, - /*.filesFactor = filesFactor,*/ }; if (qCreateQueryPlan(&cxt, &pPlan, NULL) < 0) { @@ -270,7 +269,6 @@ int32_t mndAddShuffleSinkTasksToStream(SMnode* pMnode, STrans* pTrans, SStreamOb pTask->epSet = mndGetVgroupEpset(pMnode, pVgroup); // source - pTask->sourceType = TASK_SOURCE__MERGE; pTask->inputType = TASK_INPUT_TYPE__DATA_BLOCK; // exec @@ -316,7 +314,6 @@ int32_t mndAddFixedSinkTaskToStream(SMnode* pMnode, STrans* pTrans, SStreamObj* #endif pTask->epSet = mndGetVgroupEpset(pMnode, &pStream->fixedSinkVg); // source - pTask->sourceType = TASK_SOURCE__MERGE; pTask->inputType = TASK_INPUT_TYPE__DATA_BLOCK; // exec @@ -427,6 +424,8 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) { SStreamTask* pTask = tNewSStreamTask(pStream->uid); mndAddTaskToTaskSet(taskSourceLevel, pTask); + pTask->dataScan = 1; + // input pTask->inputType = TASK_INPUT_TYPE__SUMBIT_BLOCK; @@ -470,6 +469,8 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) { SStreamTask* pTask = tNewSStreamTask(pStream->uid); mndAddTaskToTaskSet(taskOneLevel, pTask); + pTask->dataScan = 1; + // input pTask->inputType = TASK_INPUT_TYPE__SUMBIT_BLOCK; diff --git a/source/dnode/snode/inc/sndInt.h b/source/dnode/snode/inc/sndInt.h index 2802537dcd31323a703cd6a4d2222bb8fd9d3d68..8916e2a31cd79c3486c00841979e14fcaf25b2df 100644 --- a/source/dnode/snode/inc/sndInt.h +++ b/source/dnode/snode/inc/sndInt.h @@ -56,7 +56,6 @@ SStreamTask* sndMetaGetTask(SStreamMeta* pMeta, int32_t taskId); int32_t sndMetaRemoveTask(SStreamMeta* pMeta, int32_t taskId); int32_t sndDropTaskOfStream(SStreamMeta* pMeta, int64_t streamId); - int32_t sndStopTaskOfStream(SStreamMeta* pMeta, int64_t streamId); int32_t sndResumeTaskOfStream(SStreamMeta* pMeta, int64_t streamId); diff --git a/source/dnode/snode/src/snode.c b/source/dnode/snode/src/snode.c index cbbe071c5fa386e178273303ae7d2b35bf1b7ad2..8ef48ccbf9324888cad8b227fb7fa434843078aa 100644 --- a/source/dnode/snode/src/snode.c +++ b/source/dnode/snode/src/snode.c @@ -76,45 +76,158 @@ int32_t sndMetaRemoveTask(SStreamMeta *pMeta, int32_t taskId) { return taosHashRemove(pMeta->pHash, &taskId, sizeof(int32_t)); } -static int32_t sndProcessTaskExecReq(SSnode *pSnode, SRpcMsg *pMsg) { - /*SStreamExecMsgHead *pHead = pMsg->pCont;*/ - /*int32_t taskId = pHead->streamTaskId;*/ - /*SStreamTask *pTask = sndMetaGetTask(pSnode->pMeta, taskId);*/ - /*if (pTask == NULL) {*/ - /*return -1;*/ - /*}*/ +static int32_t sndProcessTaskDeployReq(SSnode *pNode, SRpcMsg *pMsg) { + SStreamMeta *pMeta = pNode->pMeta; + char *msg = pMsg->pCont; + int32_t msgLen = pMsg->contLen; + + SStreamTask *pTask = taosMemoryCalloc(1, sizeof(SStreamTask)); + if (pTask == NULL) { + return -1; + } + SDecoder decoder; + tDecoderInit(&decoder, (uint8_t *)msg, msgLen); + if (tDecodeSStreamTask(&decoder, pTask) < 0) { + ASSERT(0); + } + tDecoderClear(&decoder); + + pTask->status = TASK_STATUS__IDLE; + + pTask->inputQueue = streamQueueOpen(); + pTask->outputQueue = streamQueueOpen(); + pTask->inputStatus = TASK_INPUT_STATUS__NORMAL; + pTask->outputStatus = TASK_INPUT_STATUS__NORMAL; + + if (pTask->inputQueue == NULL || pTask->outputQueue == NULL) goto FAIL; + + pTask->pMsgCb = &pNode->msgCb; + + ASSERT(pTask->execType != TASK_EXEC__NONE); + + SReadHandle handle = { + .pMsgCb = &pNode->msgCb, + }; + + /*pTask->exec.inputHandle = NULL;*/ + pTask->exec.executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, &handle); + ASSERT(pTask->exec.executor); + + streamSetupTrigger(pTask); + + qInfo("deploy stream: stream id %ld task id %d child id %d on snode", pTask->streamId, pTask->taskId, pTask->childId); + + return 0; + +FAIL: + if (pTask->inputQueue) streamQueueClose(pTask->inputQueue); + if (pTask->outputQueue) streamQueueClose(pTask->outputQueue); + return -1; +} + +static int32_t sndProcessTaskRunReq(SSnode *pNode, SRpcMsg *pMsg) { + SStreamMeta *pMeta = pNode->pMeta; + SStreamTaskRunReq *pReq = pMsg->pCont; + int32_t taskId = pReq->taskId; + SStreamTask *pTask = *(SStreamTask **)taosHashGet(pMeta->pHash, &taskId, sizeof(int32_t)); + streamTaskProcessRunReq(pTask, &pNode->msgCb); return 0; } -void sndProcessUMsg(SSnode *pSnode, SRpcMsg *pMsg) { +static int32_t sndProcessTaskDispatchReq(SSnode *pNode, SRpcMsg *pMsg) { + SStreamMeta *pMeta = pNode->pMeta; + + char *msgStr = pMsg->pCont; + char *msgBody = POINTER_SHIFT(msgStr, sizeof(SMsgHead)); + int32_t msgLen = pMsg->contLen - sizeof(SMsgHead); + + SStreamDispatchReq req; + SDecoder decoder; + tDecoderInit(&decoder, msgBody, msgLen); + tDecodeStreamDispatchReq(&decoder, &req); + int32_t taskId = req.taskId; + SStreamTask *pTask = *(SStreamTask **)taosHashGet(pMeta->pHash, &taskId, sizeof(int32_t)); + SRpcMsg rsp = { + .info = pMsg->info, + .code = 0, + }; + streamProcessDispatchReq(pTask, &pNode->msgCb, &req, &rsp); + return 0; +} + +static int32_t sndProcessTaskRecoverReq(SSnode *pNode, SRpcMsg *pMsg) { + SStreamMeta *pMeta = pNode->pMeta; + + SStreamTaskRecoverReq *pReq = pMsg->pCont; + int32_t taskId = pReq->taskId; + SStreamTask *pTask = *(SStreamTask **)taosHashGet(pMeta->pHash, &taskId, sizeof(int32_t)); + streamProcessRecoverReq(pTask, &pNode->msgCb, pReq, pMsg); + return 0; +} + +static int32_t sndProcessTaskDispatchRsp(SSnode *pNode, SRpcMsg *pMsg) { + SStreamMeta *pMeta = pNode->pMeta; + + SStreamDispatchRsp *pRsp = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)); + int32_t taskId = pRsp->taskId; + SStreamTask *pTask = *(SStreamTask **)taosHashGet(pMeta->pHash, &taskId, sizeof(int32_t)); + streamProcessDispatchRsp(pTask, &pNode->msgCb, pRsp); + return 0; +} + +static int32_t sndProcessTaskRecoverRsp(SSnode *pNode, SRpcMsg *pMsg) { + SStreamMeta *pMeta = pNode->pMeta; + + SStreamTaskRecoverRsp *pRsp = pMsg->pCont; + int32_t taskId = pRsp->taskId; + SStreamTask *pTask = *(SStreamTask **)taosHashGet(pMeta->pHash, &taskId, sizeof(int32_t)); + streamProcessRecoverRsp(pTask, pRsp); + return 0; +} + +static int32_t sndProcessTaskDropReq(SSnode *pNode, SRpcMsg *pMsg) { + SStreamMeta *pMeta = pNode->pMeta; + + char *msg = pMsg->pCont; + int32_t msgLen = pMsg->contLen; + SVDropStreamTaskReq *pReq = (SVDropStreamTaskReq *)msg; + int32_t code = taosHashRemove(pMeta->pHash, &pReq->taskId, sizeof(int32_t)); + ASSERT(code == 0); + if (code == 0) { + // sendrsp + } + return code; +} + +int32_t sndProcessUMsg(SSnode *pSnode, SRpcMsg *pMsg) { // stream deploy // stream stop/resume // operator exec - if (pMsg->msgType == TDMT_STREAM_TASK_DEPLOY) { - void *msg = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)); - SStreamTask *pTask = taosMemoryMalloc(sizeof(SStreamTask)); - if (pTask == NULL) { + switch (pMsg->msgType) { + case TDMT_STREAM_TASK_DEPLOY: + return sndProcessTaskDeployReq(pSnode, pMsg); + case TDMT_VND_STREAM_TASK_DROP: + return sndProcessTaskDropReq(pSnode, pMsg); + default: ASSERT(0); - return; - } - SDecoder decoder; - tDecoderInit(&decoder, msg, pMsg->contLen - sizeof(SMsgHead)); - tDecodeSStreamTask(&decoder, pTask); - tDecoderClear(&decoder); - - sndMetaDeployTask(pSnode->pMeta, pTask); - /*} else if (pMsg->msgType == TDMT_SND_TASK_EXEC) {*/ - /*sndProcessTaskExecReq(pSnode, pMsg);*/ - } else { - ASSERT(0); } + return 0; } -void sndProcessSMsg(SSnode *pSnode, SRpcMsg *pMsg) { - // operator exec - /*if (pMsg->msgType == TDMT_SND_TASK_EXEC) {*/ - /*sndProcessTaskExecReq(pSnode, pMsg);*/ - /*} else {*/ - ASSERT(0); - /*}*/ +int32_t sndProcessSMsg(SSnode *pSnode, SRpcMsg *pMsg) { + switch (pMsg->msgType) { + case TDMT_STREAM_TASK_RUN: + return sndProcessTaskRunReq(pSnode, pMsg); + case TDMT_STREAM_TASK_DISPATCH: + return sndProcessTaskDispatchReq(pSnode, pMsg); + case TDMT_STREAM_TASK_RECOVER: + return sndProcessTaskRecoverReq(pSnode, pMsg); + case TDMT_STREAM_TASK_DISPATCH_RSP: + return sndProcessTaskDispatchRsp(pSnode, pMsg); + case TDMT_STREAM_TASK_RECOVER_RSP: + return sndProcessTaskRecoverRsp(pSnode, pMsg); + default: + ASSERT(0); + } + return 0; } diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index 70b09ec701af09888f37434367b72bcfe3c46938..06a119b076af7062ef6da9b7d02073d503818102 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -125,10 +125,10 @@ int32_t tqProcessOffsetCommitReq(STQ* pTq, char* msg, int32_t msgLen) { if (offset.type == TMQ_OFFSET__SNAPSHOT) { tqDebug("receive offset commit msg to %s on vg %d, offset(type:snapshot) uid: %ld, ts: %ld", offset.subKey, - pTq->pVnode->config.vgId, offset.uid, offset.ts); + TD_VID(pTq->pVnode), offset.uid, offset.ts); } else if (offset.type == TMQ_OFFSET__LOG) { tqDebug("receive offset commit msg to %s on vg %d, offset(type:log) version: %ld", offset.subKey, - pTq->pVnode->config.vgId, offset.version); + TD_VID(pTq->pVnode), offset.version); } else { ASSERT(0); } @@ -159,7 +159,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) { if (pOffset != NULL) { ASSERT(pOffset->type == TMQ_OFFSET__LOG); tqDebug("consumer %ld, restore offset of %s on vg %d, offset(type:log) version: %ld", consumerId, pReq->subKey, - pTq->pVnode->config.vgId, pOffset->version); + TD_VID(pTq->pVnode), pOffset->version); fetchOffset = pOffset->version + 1; } else { if (pReq->currentOffset == TMQ_CONF__RESET_OFFSET__EARLIEAST) { @@ -167,13 +167,13 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) { } else if (pReq->currentOffset == TMQ_CONF__RESET_OFFSET__LATEST) { fetchOffset = walGetCommittedVer(pTq->pWal); } else if (pReq->currentOffset == TMQ_CONF__RESET_OFFSET__NONE) { - tqError("tmq poll: no offset committed for consumer %ld in vg %d, subkey %s", consumerId, - pTq->pVnode->config.vgId, pReq->subKey); + tqError("tmq poll: no offset committed for consumer %ld in vg %d, subkey %s", consumerId, TD_VID(pTq->pVnode), + pReq->subKey); terrno = TSDB_CODE_TQ_NO_COMMITTED_OFFSET; return -1; } tqDebug("consumer %ld, restore offset of %s on vg %d failed, config is %ld, set to %ld", consumerId, pReq->subKey, - pTq->pVnode->config.vgId, pReq->currentOffset, fetchOffset); + TD_VID(pTq->pVnode), pReq->currentOffset, fetchOffset); } } @@ -183,14 +183,14 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) { STqHandle* pHandle = taosHashGet(pTq->handles, pReq->subKey, strlen(pReq->subKey)); /*ASSERT(pHandle);*/ if (pHandle == NULL) { - tqError("tmq poll: no consumer handle for consumer %ld in vg %d, subkey %s", consumerId, pTq->pVnode->config.vgId, + tqError("tmq poll: no consumer handle for consumer %ld in vg %d, subkey %s", consumerId, TD_VID(pTq->pVnode), pReq->subKey); return -1; } if (pHandle->consumerId != consumerId) { tqError("tmq poll: consumer handle mismatch for consumer %ld in vg %d, subkey %s, handle consumer id %ld", - consumerId, pTq->pVnode->config.vgId, pReq->subKey, pHandle->consumerId); + consumerId, TD_VID(pTq->pVnode), pReq->subKey, pHandle->consumerId); return -1; } @@ -304,7 +304,6 @@ int32_t tqProcessVgDeleteReq(STQ* pTq, char* msg, int32_t msgLen) { return 0; } -// TODO: persist meta into tdb int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) { SMqRebVgReq req = {0}; tDecodeSMqRebVgReq(msg, &req); @@ -346,10 +345,10 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) { pHandle->execHandle.execTb.suid = req.suid; SArray* tbUidList = taosArrayInit(0, sizeof(int64_t)); tsdbGetCtbIdList(pTq->pVnode->pMeta, req.suid, tbUidList); - tqDebug("vg %d, tq try get suid: %ld", pTq->pVnode->config.vgId, req.suid); + tqDebug("vg %d, tq try get suid: %ld", TD_VID(pTq->pVnode), req.suid); for (int32_t i = 0; i < taosArrayGetSize(tbUidList); i++) { int64_t tbUid = *(int64_t*)taosArrayGet(tbUidList, i); - tqDebug("vg %d, idx %d, uid: %ld", pTq->pVnode->config.vgId, i, tbUid); + tqDebug("vg %d, idx %d, uid: %ld", TD_VID(pTq->pVnode), i, tbUid); } for (int32_t i = 0; i < 5; i++) { tqReadHandleSetTbUidList(pHandle->execHandle.pExecReader[i], tbUidList); @@ -400,16 +399,21 @@ int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) { // exec if (pTask->execType != TASK_EXEC__NONE) { // expand runners - STqReadHandle* pStreamReader = tqInitSubmitMsgScanner(pTq->pVnode->pMeta); - SReadHandle handle = { - .reader = pStreamReader, - .meta = pTq->pVnode->pMeta, - .pMsgCb = &pTq->pVnode->msgCb, - .vnode = pTq->pVnode, - }; - pTask->exec.inputHandle = pStreamReader; - pTask->exec.executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, &handle); - ASSERT(pTask->exec.executor); + if (pTask->dataScan) { + STqReadHandle* pStreamReader = tqInitSubmitMsgScanner(pTq->pVnode->pMeta); + SReadHandle handle = { + .reader = pStreamReader, + .meta = pTq->pVnode->pMeta, + .pMsgCb = &pTq->pVnode->msgCb, + .vnode = pTq->pVnode, + }; + /*pTask->exec.inputHandle = pStreamReader;*/ + pTask->exec.executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, &handle); + ASSERT(pTask->exec.executor); + } else { + pTask->exec.executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, NULL); + ASSERT(pTask->exec.executor); + } } // sink @@ -431,7 +435,7 @@ int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) { 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, TD_VID(pTq->pVnode)); taosHashPut(pTq->pStreamTasks, &pTask->taskId, sizeof(int32_t), &pTask, sizeof(void*)); @@ -464,7 +468,7 @@ int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* pReq) { continue; } - if (streamLaunchByWrite(pTask, pTq->pVnode->config.vgId, &pTq->pVnode->msgCb) < 0) { + if (streamLaunchByWrite(pTask, TD_VID(pTq->pVnode), &pTq->pVnode->msgCb) < 0) { continue; } } else { @@ -534,9 +538,9 @@ int32_t tqProcessTaskRecoverRsp(STQ* pTq, SRpcMsg* pMsg) { int32_t tqProcessTaskDropReq(STQ* pTq, char* msg, int32_t msgLen) { SVDropStreamTaskReq* pReq = (SVDropStreamTaskReq*)msg; int32_t code = taosHashRemove(pTq->pStreamTasks, &pReq->taskId, sizeof(int32_t)); + ASSERT(code == 0); if (code == 0) { // sendrsp } - ASSERT(code == 0); return code; } diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c index b1d076e8f57e1970fc42e0834032d374e8ed149e..c99b3c1058d8f1e87312261ccdb1f39c3c526828 100644 --- a/source/libs/executor/src/executor.c +++ b/source/libs/executor/src/executor.c @@ -99,7 +99,7 @@ int32_t qSetMultiStreamInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numO } qTaskInfo_t qCreateStreamExecTaskInfo(void* msg, void* streamReadHandle) { - if (msg == NULL || streamReadHandle == NULL) { + if (msg == NULL) { return NULL; } diff --git a/source/libs/executor/src/executorMain.c b/source/libs/executor/src/executorMain.c index ff281dacbdbae5d1cdf600fc6fb44ad2a4d2bdec..a9e1e03178b0a28780ecfdd696d44e244d2648a1 100644 --- a/source/libs/executor/src/executorMain.c +++ b/source/libs/executor/src/executorMain.c @@ -13,10 +13,10 @@ * along with this program. If not, see . */ -#include "os.h" -#include "tref.h" #include "dataSinkMgt.h" +#include "os.h" #include "tmsg.h" +#include "tref.h" #include "tudf.h" #include "executor.h" @@ -24,15 +24,13 @@ #include "query.h" static TdThreadOnce initPoolOnce = PTHREAD_ONCE_INIT; -int32_t exchangeObjRefPool = -1; +int32_t exchangeObjRefPool = -1; -static void initRefPool() { - exchangeObjRefPool = taosOpenRef(1024, doDestroyExchangeOperatorInfo); -} +static void initRefPool() { exchangeObjRefPool = taosOpenRef(1024, doDestroyExchangeOperatorInfo); } int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId, SSubplan* pSubplan, qTaskInfo_t* pTaskInfo, DataSinkHandle* handle, const char* sql, EOPTR_EXEC_MODEL model) { - assert(readHandle != NULL && pSubplan != NULL); + assert(pSubplan != NULL); SExecTaskInfo** pTask = (SExecTaskInfo**)pTaskInfo; taosThreadOnce(&initPoolOnce, initRefPool); @@ -47,57 +45,57 @@ int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId, if (code != TSDB_CODE_SUCCESS) { goto _error; } - + if (handle) { void* pSinkParam = NULL; code = createDataSinkParam(pSubplan->pDataSink, &pSinkParam, pTaskInfo); if (code != TSDB_CODE_SUCCESS) { goto _error; } - + code = dsCreateDataSinker(pSubplan->pDataSink, handle, pSinkParam); } - _error: +_error: // if failed to add ref for all tables in this query, abort current query return code; } #ifdef TEST_IMPL // wait moment -int waitMoment(SQInfo* pQInfo){ - if(pQInfo->sql) { - int ms = 0; +int waitMoment(SQInfo* pQInfo) { + if (pQInfo->sql) { + int ms = 0; char* pcnt = strstr(pQInfo->sql, " count(*)"); - if(pcnt) return 0; - + if (pcnt) return 0; + char* pos = strstr(pQInfo->sql, " t_"); - if(pos){ + if (pos) { pos += 3; ms = atoi(pos); - while(*pos >= '0' && *pos <= '9'){ - pos ++; + while (*pos >= '0' && *pos <= '9') { + pos++; } char unit_char = *pos; - if(unit_char == 'h'){ - ms *= 3600*1000; - } else if(unit_char == 'm'){ - ms *= 60*1000; - } else if(unit_char == 's'){ + if (unit_char == 'h') { + ms *= 3600 * 1000; + } else if (unit_char == 'm') { + ms *= 60 * 1000; + } else if (unit_char == 's') { ms *= 1000; } } - if(ms == 0) return 0; + if (ms == 0) return 0; printf("test wait sleep %dms. sql=%s ...\n", ms, pQInfo->sql); - - if(ms < 1000) { + + if (ms < 1000) { taosMsleep(ms); } else { int used_ms = 0; - while(used_ms < ms) { + while (used_ms < ms) { taosMsleep(1000); used_ms += 1000; - if(isTaskKilled(pQInfo)){ + if (isTaskKilled(pQInfo)) { printf("test check query is canceled, sleep break.%s\n", pQInfo->sql); break; } @@ -108,15 +106,14 @@ int waitMoment(SQInfo* pQInfo){ } #endif -int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t *useconds) { +int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t* useconds) { SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo; int64_t threadId = taosGetSelfPthreadId(); *pRes = NULL; int64_t curOwner = 0; if ((curOwner = atomic_val_compare_exchange_64(&pTaskInfo->owner, 0, threadId)) != 0) { - qError("%s-%p execTask is now executed by thread:%p", GET_TASKID(pTaskInfo), pTaskInfo, - (void*)curOwner); + qError("%s-%p execTask is now executed by thread:%p", GET_TASKID(pTaskInfo), pTaskInfo, (void*)curOwner); pTaskInfo->code = TSDB_CODE_QRY_IN_EXEC; return pTaskInfo->code; } @@ -152,18 +149,18 @@ int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t *useconds) { cleanUpUdfs(); - int32_t current = (*pRes != NULL)? (*pRes)->info.rows:0; + int32_t current = (*pRes != NULL) ? (*pRes)->info.rows : 0; uint64_t total = pTaskInfo->pRoot->resultInfo.totalRows; qDebug("%s task suspended, %d rows returned, total:%" PRId64 " rows, in sinkNode:%d, elapsed:%.2f ms", - GET_TASKID(pTaskInfo), current, total, 0, el/1000.0); + GET_TASKID(pTaskInfo), current, total, 0, el / 1000.0); atomic_store_64(&pTaskInfo->owner, 0); return pTaskInfo->code; } int32_t qKillTask(qTaskInfo_t qinfo) { - SExecTaskInfo *pTaskInfo = (SExecTaskInfo *)qinfo; + SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)qinfo; if (pTaskInfo == NULL) { return TSDB_CODE_QRY_INVALID_QHANDLE; @@ -182,7 +179,7 @@ int32_t qKillTask(qTaskInfo_t qinfo) { } int32_t qAsyncKillTask(qTaskInfo_t qinfo) { - SExecTaskInfo *pTaskInfo = (SExecTaskInfo *)qinfo; + SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)qinfo; if (pTaskInfo == NULL) { return TSDB_CODE_QRY_INVALID_QHANDLE; @@ -195,7 +192,7 @@ int32_t qAsyncKillTask(qTaskInfo_t qinfo) { } int32_t qIsTaskCompleted(qTaskInfo_t qinfo) { - SExecTaskInfo *pTaskInfo = (SExecTaskInfo *)qinfo; + SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)qinfo; if (pTaskInfo == NULL) { return TSDB_CODE_QRY_INVALID_QHANDLE; @@ -205,18 +202,18 @@ int32_t qIsTaskCompleted(qTaskInfo_t qinfo) { } void qDestroyTask(qTaskInfo_t qTaskHandle) { - SExecTaskInfo* pTaskInfo = (SExecTaskInfo*) qTaskHandle; - qDebug("%s execTask completed, numOfRows:%"PRId64, GET_TASKID(pTaskInfo), pTaskInfo->pRoot->resultInfo.totalRows); + SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)qTaskHandle; + qDebug("%s execTask completed, numOfRows:%" PRId64, GET_TASKID(pTaskInfo), pTaskInfo->pRoot->resultInfo.totalRows); - queryCostStatis(pTaskInfo); // print the query cost summary + queryCostStatis(pTaskInfo); // print the query cost summary doDestroyTask(pTaskInfo); } -int32_t qGetExplainExecInfo(qTaskInfo_t tinfo, int32_t *resNum, SExplainExecInfo **pRes) { - SExecTaskInfo *pTaskInfo = (SExecTaskInfo *)tinfo; - int32_t capacity = 0; +int32_t qGetExplainExecInfo(qTaskInfo_t tinfo, int32_t* resNum, SExplainExecInfo** pRes) { + SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo; + int32_t capacity = 0; - return getOperatorExplainExecInfo(pTaskInfo->pRoot, pRes, &capacity, resNum); + return getOperatorExplainExecInfo(pTaskInfo->pRoot, pRes, &capacity, resNum); } int32_t qSerializeTaskStatus(qTaskInfo_t tinfo, char** pOutput, int32_t* len) { diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 54b5961585580f1d289b50b37852d787f2639d79..5c038ed7098d79f2de2e5d36b19724aa916f096c 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -141,8 +141,8 @@ static int32_t doCopyToSDataBlock(SExecTaskInfo* taskInfo, SSDataBlock* pBlock, SqlFunctionCtx* pCtx, int32_t numOfExprs); static void initCtxOutputBuffer(SqlFunctionCtx* pCtx, int32_t size); -static void doSetTableGroupOutputBuf(SOperatorInfo* pOperator, SAggOperatorInfo* pAggInfo, int32_t numOfOutput, uint64_t groupId); - +static void doSetTableGroupOutputBuf(SOperatorInfo* pOperator, SAggOperatorInfo* pAggInfo, int32_t numOfOutput, + uint64_t groupId); // setup the output buffer for each operator static bool hasNull(SColumn* pColumn, SColumnDataAgg* pStatis) { @@ -1230,8 +1230,9 @@ void initResultRow(SResultRow* pResultRow) { * offset[0] offset[1] offset[2] */ // TODO refactor: some function move away -void setFunctionResultOutput(SOperatorInfo *pOperator, SOptrBasicInfo *pInfo, SAggSupporter* pSup, int32_t stage, int32_t numOfExprs) { - SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; +void setFunctionResultOutput(SOperatorInfo* pOperator, SOptrBasicInfo* pInfo, SAggSupporter* pSup, int32_t stage, + int32_t numOfExprs) { + SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SqlFunctionCtx* pCtx = pOperator->exprSupp.pCtx; int32_t* rowEntryInfoOffset = pOperator->exprSupp.rowEntryInfoOffset; @@ -1380,9 +1381,10 @@ void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const int8_t* rowR } } -void doSetTableGroupOutputBuf(SOperatorInfo* pOperator, SAggOperatorInfo* pAggInfo, int32_t numOfOutput, uint64_t groupId) { +void doSetTableGroupOutputBuf(SOperatorInfo* pOperator, SAggOperatorInfo* pAggInfo, int32_t numOfOutput, + uint64_t groupId) { // for simple group by query without interval, all the tables belong to one group result. - SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; + SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SResultRowInfo* pResultRowInfo = &pAggInfo->binfo.resultRowInfo; SqlFunctionCtx* pCtx = pOperator->exprSupp.pCtx; int32_t* rowEntryInfoOffset = pOperator->exprSupp.rowEntryInfoOffset; @@ -2040,8 +2042,8 @@ static int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInf } int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadInfo, int32_t numOfRows, char* pData, - int32_t compLen, int32_t numOfOutput, int64_t startTs, uint64_t* total, - SArray* pColList) { + int32_t compLen, int32_t numOfOutput, int64_t startTs, uint64_t* total, + SArray* pColList) { if (pColList == NULL) { // data from other sources blockCompressDecode(pRes, numOfOutput, numOfRows, pData); pRes->info.rows = numOfRows; @@ -2165,8 +2167,9 @@ static SSDataBlock* concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SEx } SRetrieveTableRsp* pTableRsp = pDataInfo->pRsp; - code = extractDataBlockFromFetchRsp(pExchangeInfo->pResult, pLoadInfo, pTableRsp->numOfRows, pTableRsp->data, - pTableRsp->compLen, pTableRsp->numOfCols, startTs, &pDataInfo->totalRows, NULL); + code = + extractDataBlockFromFetchRsp(pExchangeInfo->pResult, pLoadInfo, pTableRsp->numOfRows, pTableRsp->data, + pTableRsp->compLen, pTableRsp->numOfCols, startTs, &pDataInfo->totalRows, NULL); if (code != 0) { taosMemoryFreeClear(pDataInfo->pRsp); goto _error; @@ -2281,7 +2284,7 @@ static SSDataBlock* seqLoadRemoteData(SOperatorInfo* pOperator) { SRetrieveTableRsp* pTableRsp = pDataInfo->pRsp; int32_t code = extractDataBlockFromFetchRsp(pExchangeInfo->pResult, pLoadInfo, pTableRsp->numOfRows, pTableRsp->data, - pTableRsp->compLen, pTableRsp->numOfCols, startTs, &pDataInfo->totalRows, NULL); + pTableRsp->compLen, pTableRsp->numOfCols, startTs, &pDataInfo->totalRows, NULL); if (pRsp->completed == 1) { qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64 " numOfRows:%d, rowsOfSource:%" PRIu64 @@ -2673,8 +2676,8 @@ static SSDataBlock* doSortedMerge(SOperatorInfo* pOperator) { } int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize; - pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, SORT_MULTISOURCE_MERGE, pInfo->bufPageSize, - numOfBufPage, pInfo->binfo.pRes, "GET_TASKID(pTaskInfo)"); + pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, SORT_MULTISOURCE_MERGE, pInfo->bufPageSize, numOfBufPage, + pInfo->binfo.pRes, "GET_TASKID(pTaskInfo)"); tsortSetFetchRawDataFp(pInfo->pSortHandle, loadNextDataBlock, NULL, NULL); @@ -3130,7 +3133,7 @@ static SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) { SProjectOperatorInfo* pProjectInfo = pOperator->info; SOptrBasicInfo* pInfo = &pProjectInfo->binfo; - SExprSupp* pSup = &pOperator->exprSupp; + SExprSupp* pSup = &pOperator->exprSupp; SSDataBlock* pRes = pInfo->pRes; blockDataCleanup(pRes); @@ -3403,8 +3406,8 @@ void cleanupAggSup(SAggSupporter* pAggSup) { destroyDiskbasedBuf(pAggSup->pResultBuf); } -int32_t initAggInfo(SExprSupp* pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, - size_t keyBufSize, const char* pkey) { +int32_t initAggInfo(SExprSupp* pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, size_t keyBufSize, + const char* pkey) { initExprSupp(pSup, pExprInfo, numOfCols); doInitAggInfoSup(pAggSup, pSup->pCtx, numOfCols, keyBufSize, pkey); for (int32_t i = 0; i < numOfCols; ++i) { @@ -3457,12 +3460,12 @@ SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo* initBasicInfo(&pInfo->binfo, pResultBlock); initExprSupp(&pInfo->scalarExprSup, pScalarExprInfo, numOfScalarExpr); - pInfo->groupId = INT32_MIN; - pOperator->name = "TableAggregate"; + pInfo->groupId = INT32_MIN; + pOperator->name = "TableAggregate"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_HASH_AGG; - pOperator->blocking = true; - pOperator->status = OP_NOT_OPENED; - pOperator->info = pInfo; + pOperator->blocking = true; + pOperator->status = OP_NOT_OPENED; + pOperator->info = pInfo; pOperator->pTaskInfo = pTaskInfo; pOperator->fpSet = createOperatorFpSet(doOpenAggregateOptr, getAggregateResult, NULL, NULL, destroyAggOperatorInfo, @@ -3578,25 +3581,26 @@ static SArray* setRowTsColumnOutputInfo(SqlFunctionCtx* pCtx, int32_t numOfCols) return pList; } -SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhysiNode* pProjPhyNode, SExecTaskInfo* pTaskInfo) { +SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhysiNode* pProjPhyNode, + SExecTaskInfo* pTaskInfo) { SProjectOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SProjectOperatorInfo)); SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); if (pInfo == NULL || pOperator == NULL) { goto _error; } - int32_t numOfCols = 0; + int32_t numOfCols = 0; SExprInfo* pExprInfo = createExprInfo(pProjPhyNode->pProjections, NULL, &numOfCols); SSDataBlock* pResBlock = createResDataBlock(pProjPhyNode->node.pOutputDataBlockDesc); SLimit limit = {.limit = pProjPhyNode->limit, .offset = pProjPhyNode->offset}; SLimit slimit = {.limit = pProjPhyNode->slimit, .offset = pProjPhyNode->soffset}; - pInfo->limit = limit; - pInfo->slimit = slimit; - pInfo->curOffset = limit.offset; - pInfo->curSOffset = slimit.offset; - pInfo->binfo.pRes = pResBlock; + pInfo->limit = limit; + pInfo->slimit = slimit; + pInfo->curOffset = limit.offset; + pInfo->curSOffset = slimit.offset; + pInfo->binfo.pRes = pResBlock; pInfo->pFilterNode = pProjPhyNode->node.pConditions; int32_t numOfRows = 4096; @@ -3614,12 +3618,12 @@ SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhys setFunctionResultOutput(pOperator, &pInfo->binfo, &pInfo->aggSup, MAIN_SCAN, numOfCols); pInfo->pPseudoColInfo = setRowTsColumnOutputInfo(pOperator->exprSupp.pCtx, numOfCols); - pOperator->name = "ProjectOperator"; + pOperator->name = "ProjectOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_PROJECT; - pOperator->blocking = false; - pOperator->status = OP_NOT_OPENED; - pOperator->info = pInfo; - pOperator->pTaskInfo = pTaskInfo; + pOperator->blocking = false; + pOperator->status = OP_NOT_OPENED; + pOperator->info = pInfo; + pOperator->pTaskInfo = pTaskInfo; pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doProjectOperation, NULL, NULL, destroyProjectOperatorInfo, NULL, NULL, NULL); @@ -3639,7 +3643,7 @@ _error: static SSDataBlock* doApplyIndefinitFunction(SOperatorInfo* pOperator) { SIndefOperatorInfo* pIndefInfo = pOperator->info; SOptrBasicInfo* pInfo = &pIndefInfo->binfo; - SExprSupp* pSup = &pOperator->exprSupp; + SExprSupp* pSup = &pOperator->exprSupp; SSDataBlock* pRes = pInfo->pRes; blockDataCleanup(pRes); @@ -3677,7 +3681,7 @@ static SSDataBlock* doApplyIndefinitFunction(SOperatorInfo* pOperator) { SExprSupp* pScalarSup = &pIndefInfo->scalarSup; if (pScalarSup->pExprInfo != NULL) { code = projectApplyFunctions(pScalarSup->pExprInfo, pBlock, pBlock, pScalarSup->pCtx, pScalarSup->numOfExprs, - pIndefInfo->pPseudoColInfo); + pIndefInfo->pPseudoColInfo); if (code != TSDB_CODE_SUCCESS) { longjmp(pTaskInfo->env, code); } @@ -3686,8 +3690,8 @@ static SSDataBlock* doApplyIndefinitFunction(SOperatorInfo* pOperator) { setInputDataBlock(pOperator, pSup->pCtx, pBlock, order, scanFlag, false); blockDataEnsureCapacity(pInfo->pRes, pInfo->pRes->info.rows + pBlock->info.rows); - code = projectApplyFunctions(pOperator->exprSupp.pExprInfo, pInfo->pRes, pBlock, pSup->pCtx, pOperator->exprSupp.numOfExprs, - pIndefInfo->pPseudoColInfo); + code = projectApplyFunctions(pOperator->exprSupp.pExprInfo, pInfo->pRes, pBlock, pSup->pCtx, + pOperator->exprSupp.numOfExprs, pIndefInfo->pPseudoColInfo); if (code != TSDB_CODE_SUCCESS) { longjmp(pTaskInfo->env, code); } @@ -3745,14 +3749,14 @@ SOperatorInfo* createIndefinitOutputOperatorInfo(SOperatorInfo* downstream, SPhy pInfo->binfo.pRes = pResBlock; pInfo->pPseudoColInfo = setRowTsColumnOutputInfo(pSup->pCtx, numOfExpr); - pOperator->name = "IndefinitOperator"; + pOperator->name = "IndefinitOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_PROJECT; - pOperator->blocking = false; - pOperator->status = OP_NOT_OPENED; - pOperator->info = pInfo; + pOperator->blocking = false; + pOperator->status = OP_NOT_OPENED; + pOperator->info = pInfo; pOperator->exprSupp.pExprInfo = pExprInfo; pOperator->exprSupp.numOfExprs = numOfExpr; - pOperator->pTaskInfo = pTaskInfo; + pOperator->pTaskInfo = pTaskInfo; pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doApplyIndefinitFunction, NULL, NULL, destroyIndefinitOperatorInfo, NULL, NULL, NULL); @@ -3827,11 +3831,11 @@ SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode* goto _error; } - int32_t num = 0; - SSDataBlock* pResBlock = createResDataBlock(pPhyFillNode->node.pOutputDataBlockDesc); - SExprInfo* pExprInfo = createExprInfo(pPhyFillNode->pTargets, NULL, &num); - SInterval* pInterval = &((SIntervalAggOperatorInfo*)downstream->info)->interval; - int32_t type = convertFillType(pPhyFillNode->mode); + int32_t num = 0; + SSDataBlock* pResBlock = createResDataBlock(pPhyFillNode->node.pOutputDataBlockDesc); + SExprInfo* pExprInfo = createExprInfo(pPhyFillNode->pTargets, NULL, &num); + SInterval* pInterval = &((SIntervalAggOperatorInfo*)downstream->info)->interval; + int32_t type = convertFillType(pPhyFillNode->mode); SResultInfo* pResultInfo = &pOperator->resultInfo; initResultSizeInfo(pOperator, 4096); @@ -3842,16 +3846,16 @@ SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode* goto _error; } - pInfo->pRes = pResBlock; + pInfo->pRes = pResBlock; pInfo->multigroupResult = multigroupResult; - pOperator->name = "FillOperator"; - pOperator->blocking = false; - pOperator->status = OP_NOT_OPENED; + pOperator->name = "FillOperator"; + pOperator->blocking = false; + pOperator->status = OP_NOT_OPENED; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_FILL; - pOperator->exprSupp.pExprInfo = pExprInfo; - pOperator->exprSupp.numOfExprs = num; - pOperator->info = pInfo; - pOperator->pTaskInfo = pTaskInfo; + pOperator->exprSupp.pExprInfo = pExprInfo; + pOperator->exprSupp.numOfExprs = num; + pOperator->info = pInfo; + pOperator->pTaskInfo = pTaskInfo; pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doFill, NULL, NULL, destroySFillOperatorInfo, NULL, NULL, NULL); @@ -4050,10 +4054,14 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo STimeWindowAggSupp twSup = { .waterMark = pTableScanNode->watermark, .calTrigger = pTableScanNode->triggerType, .maxTs = INT64_MIN}; tsdbReaderT pDataReader = NULL; - if (pHandle->vnode) { - pDataReader = doCreateDataReader(pTableScanNode, pHandle, pTableListInfo, (uint64_t)queryId, taskId, pTagCond); - } else { - getTableList(pHandle->meta, pScanPhyNode, pTableListInfo, pTagCond); + + if (pHandle) { + if (pHandle->vnode) { + pDataReader = + doCreateDataReader(pTableScanNode, pHandle, pTableListInfo, (uint64_t)queryId, taskId, pTagCond); + } else { + getTableList(pHandle->meta, pScanPhyNode, pTableListInfo, pTagCond); + } } if (pDataReader == NULL && terrno != 0) { @@ -4310,7 +4318,7 @@ SArray* extractColumnInfo(SNodeList* pNodeList) { SColumn c = {0}; c.slotId = pNode->slotId; c.colId = pNode->slotId; - c.type = pValNode->node.type; + c.type = pValNode->node.type; c.bytes = pValNode->node.resType.bytes; c.scale = pValNode->node.resType.scale; c.precision = pValNode->node.resType.precision; @@ -4413,7 +4421,7 @@ int32_t decodeOperator(SOperatorInfo* ops, const char* result, int32_t length) { ASSERT(length == *(int32_t*)result); const char* data = result + sizeof(int32_t); - code = ops->fpSet.decodeResultRow(ops, (char*) data); + code = ops->fpSet.decodeResultRow(ops, (char*)data); if (code != TDB_CODE_SUCCESS) { return code; } diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 426d45976f0f6cff629f2dd3bbaa939ab10f93f4..c47e9dd4aee34289f975ae7d3d3d32a6536be52f 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -527,7 +527,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, goto _error; } - //taosSsleep(20); + // taosSsleep(20); SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc; @@ -740,9 +740,8 @@ static bool prepareDataScan(SStreamBlockScanInfo* pInfo) { SStreamAggSupporter* pAggSup = pInfo->sessionSup.pStreamAggSup; int64_t gap = pInfo->sessionSup.gap; int32_t winIndex = 0; - SResultWindowInfo* pCurWin = - getSessionTimeWindow(pAggSup, tsCols[pInfo->updateResIndex], INT64_MIN, - pSDB->info.groupId, gap, &winIndex); + SResultWindowInfo* pCurWin = + getSessionTimeWindow(pAggSup, tsCols[pInfo->updateResIndex], INT64_MIN, pSDB->info.groupId, gap, &winIndex); win = pCurWin->win; pInfo->updateResIndex += updateSessionWindowInfo(pCurWin, tsCols, NULL, pSDB->info.rows, pInfo->updateResIndex, gap, NULL); @@ -823,23 +822,23 @@ static SSDataBlock* doDataScan(SStreamBlockScanInfo* pInfo) { 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); + /* 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) { @@ -854,7 +853,7 @@ static void setUpdateData(SStreamBlockScanInfo* pInfo, SSDataBlock* pBlock, SSDa 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++) { + 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) { @@ -1084,9 +1083,6 @@ SOperatorInfo* createStreamScanOperatorInfo(void* pDataReader, SReadHandle* pHan SScanPhysiNode* pScanPhyNode = &pTableScanNode->scan; SDataBlockDescNode* pDescNode = pScanPhyNode->node.pOutputDataBlockDesc; - SOperatorInfo* pTableScanDummy = createTableScanOperatorInfo(pTableScanNode, pDataReader, pHandle, pTaskInfo); - - STableScanInfo* pSTInfo = (STableScanInfo*)pTableScanDummy->info; int32_t numOfCols = 0; pInfo->pColMatchInfo = extractColMatchInfo(pScanPhyNode->pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID); @@ -1103,16 +1099,6 @@ SOperatorInfo* createStreamScanOperatorInfo(void* pDataReader, SReadHandle* pHan } } - // set the extract column id to streamHandle - tqReadHandleSetColIdList((STqReadHandle*)pHandle->reader, pColIds); - SArray* tableIdList = extractTableIdList(&pTaskInfo->tableqinfoList); - int32_t code = tqReadHandleSetTbUidList(pHandle->reader, tableIdList); - if (code != 0) { - taosArrayDestroy(tableIdList); - goto _error; - } - taosArrayDestroy(tableIdList); - pInfo->pBlockLists = taosArrayInit(4, POINTER_BYTES); if (pInfo->pBlockLists == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; @@ -1124,10 +1110,31 @@ SOperatorInfo* createStreamScanOperatorInfo(void* pDataReader, SReadHandle* pHan goto _error; } - if (pSTInfo->interval.interval > 0 && pDataReader) { - pInfo->pUpdateInfo = updateInfoInitP(&pSTInfo->interval, pTwSup->waterMark); - } else { - pInfo->pUpdateInfo = NULL; + if (pHandle) { + SOperatorInfo* pTableScanDummy = createTableScanOperatorInfo(pTableScanNode, pDataReader, pHandle, pTaskInfo); + STableScanInfo* pSTInfo = (STableScanInfo*)pTableScanDummy->info; + if (pSTInfo->interval.interval > 0) { + pInfo->pUpdateInfo = updateInfoInitP(&pSTInfo->interval, pTwSup->waterMark); + } else { + pInfo->pUpdateInfo = NULL; + } + pInfo->pOperatorDumy = pTableScanDummy; + pInfo->interval = pSTInfo->interval; + + pInfo->readHandle = *pHandle; + ASSERT(pHandle->reader); + pInfo->streamBlockReader = pHandle->reader; + pInfo->tableUid = pScanPhyNode->uid; + + // set the extract column id to streamHandle + tqReadHandleSetColIdList((STqReadHandle*)pHandle->reader, pColIds); + SArray* tableIdList = extractTableIdList(&pTaskInfo->tableqinfoList); + int32_t code = tqReadHandleSetTbUidList(pHandle->reader, tableIdList); + if (code != 0) { + taosArrayDestroy(tableIdList); + goto _error; + } + taosArrayDestroy(tableIdList); } // create the pseduo columns info @@ -1135,19 +1142,14 @@ SOperatorInfo* createStreamScanOperatorInfo(void* pDataReader, SReadHandle* pHan pInfo->pPseudoExpr = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pInfo->numOfPseudoExpr); } - pInfo->readHandle = *pHandle; - pInfo->tableUid = pScanPhyNode->uid; - pInfo->streamBlockReader = pHandle->reader; pInfo->pRes = createResDataBlock(pDescNode); pInfo->pUpdateRes = createResDataBlock(pDescNode); pInfo->pCondition = pScanPhyNode->node.pConditions; pInfo->pDataReader = pDataReader; pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE; - pInfo->pOperatorDumy = pTableScanDummy; - pInfo->interval = pSTInfo->interval; pInfo->sessionSup = (SessionWindowSupporter){.pStreamAggSup = NULL, .gap = -1}; pInfo->groupId = 0; - + pOperator->name = "StreamBlockScanOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN; pOperator->blocking = false; @@ -1426,15 +1428,15 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) { // table comment pColInfoData = taosArrayGet(p->pDataBlock, 8); - if(pInfo->pCur->mr.me.ctbEntry.commentLen > 0) { + if (pInfo->pCur->mr.me.ctbEntry.commentLen > 0) { char comment[TSDB_TB_COMMENT_LEN + VARSTR_HEADER_SIZE] = {0}; STR_TO_VARSTR(comment, pInfo->pCur->mr.me.ctbEntry.comment); colDataAppend(pColInfoData, numOfRows, comment, false); - }else if(pInfo->pCur->mr.me.ctbEntry.commentLen == 0) { + } else if (pInfo->pCur->mr.me.ctbEntry.commentLen == 0) { char comment[VARSTR_HEADER_SIZE + VARSTR_HEADER_SIZE] = {0}; STR_TO_VARSTR(comment, ""); colDataAppend(pColInfoData, numOfRows, comment, false); - }else{ + } else { colDataAppendNULL(pColInfoData, numOfRows); } @@ -1462,15 +1464,15 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) { // table comment pColInfoData = taosArrayGet(p->pDataBlock, 8); - if(pInfo->pCur->mr.me.ntbEntry.commentLen > 0) { + if (pInfo->pCur->mr.me.ntbEntry.commentLen > 0) { char comment[TSDB_TB_COMMENT_LEN + VARSTR_HEADER_SIZE] = {0}; STR_TO_VARSTR(comment, pInfo->pCur->mr.me.ntbEntry.comment); colDataAppend(pColInfoData, numOfRows, comment, false); - }else if(pInfo->pCur->mr.me.ntbEntry.commentLen == 0) { + } else if (pInfo->pCur->mr.me.ntbEntry.commentLen == 0) { char comment[VARSTR_HEADER_SIZE + VARSTR_HEADER_SIZE] = {0}; STR_TO_VARSTR(comment, ""); colDataAppend(pColInfoData, numOfRows, comment, false); - }else{ + } else { colDataAppendNULL(pColInfoData, numOfRows); } @@ -1570,7 +1572,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) { } extractDataBlockFromFetchRsp(pInfo->pRes, &pInfo->loadInfo, pRsp->numOfRows, pRsp->data, pRsp->compLen, - pOperator->exprSupp.numOfExprs, startTs, NULL, pInfo->scanCols); + pOperator->exprSupp.numOfExprs, startTs, NULL, pInfo->scanCols); // todo log the filter info doFilterResult(pInfo); @@ -1867,7 +1869,8 @@ SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysi int32_t num = 0; int32_t numOfExprs = 0; SExprInfo* pExprInfo = createExprInfo(pPhyNode->pScanPseudoCols, NULL, &numOfExprs); - SArray* colList = extractColMatchInfo(pPhyNode->pScanPseudoCols, pDescNode, &num, COL_MATCH_FROM_COL_ID); + SArray* colList = extractColMatchInfo(pPhyNode->pScanPseudoCols, pDescNode, &num, COL_MATCH_FROM_COL_ID); + initExprSupp(&pOperator->exprSupp, pExprInfo, numOfExprs); @@ -1878,8 +1881,9 @@ SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysi pInfo->curPos = 0; pInfo->pFilterNode = pPhyNode->node.pConditions; - pOperator->name = "TagScanOperator"; + pOperator->name = "TagScanOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN; + pOperator->blocking = false; pOperator->status = OP_NOT_OPENED; pOperator->info = pInfo; @@ -1952,8 +1956,7 @@ typedef struct STableMergeScanInfo { int32_t createMultipleDataReaders(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle, STableListInfo* pTableListInfo, SArray* arrayReader, uint64_t queryId, uint64_t taskId, SNode* pTagCond) { - int32_t code = - getTableList(pHandle->meta, &pTableScanNode->scan, pTableListInfo, pTagCond); + int32_t code = getTableList(pHandle->meta, &pTableScanNode->scan, pTableListInfo, pTagCond); if (code != TSDB_CODE_SUCCESS) { goto _error; } @@ -1990,7 +1993,7 @@ _error: static int32_t loadDataBlockFromOneTable(SOperatorInfo* pOperator, STableMergeScanInfo* pTableScanInfo, int32_t readerIdx, SSDataBlock* pBlock, uint32_t* status) { - SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; + SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; STableMergeScanInfo* pInfo = pOperator->info; SFileBlockLoadRecorder* pCost = &pTableScanInfo->readRecorder; @@ -2177,9 +2180,8 @@ int32_t doOpenTableMergeScanOperator(SOperatorInfo* pOperator) { int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize; - pInfo->pSortHandle = - tsortCreateSortHandle(pInfo->pSortInfo, SORT_MULTISOURCE_MERGE, pInfo->bufPageSize, - numOfBufPage, pInfo->pSortInputBlock, pTaskInfo->id.str); + pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, SORT_MULTISOURCE_MERGE, pInfo->bufPageSize, numOfBufPage, + pInfo->pSortInputBlock, pTaskInfo->id.str); tsortSetFetchRawDataFp(pInfo->pSortHandle, getTableDataBlock, NULL, NULL); @@ -2243,8 +2245,7 @@ SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) { longjmp(pTaskInfo->env, code); } - SSDataBlock* pBlock = - getSortedTableMergeScanBlockData(pInfo->pSortHandle, pOperator->resultInfo.capacity, pOperator); + SSDataBlock* pBlock = getSortedTableMergeScanBlockData(pInfo->pSortHandle, pOperator->resultInfo.capacity, pOperator); if (pBlock != NULL) { pOperator->resultInfo.totalRows += pBlock->info.rows; @@ -2277,20 +2278,20 @@ void destroyTableMergeScanOperatorInfo(void* param, int32_t numOfOutput) { typedef struct STableMergeScanExecInfo { SFileBlockLoadRecorder blockRecorder; - SSortExecInfo sortExecInfo; + SSortExecInfo sortExecInfo; } STableMergeScanExecInfo; int32_t getTableMergeScanExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) { ASSERT(pOptr != NULL); // TODO: merge these two info into one struct STableMergeScanExecInfo* execInfo = taosMemoryCalloc(1, sizeof(STableMergeScanExecInfo)); - STableMergeScanInfo* pInfo = pOptr->info; + STableMergeScanInfo* pInfo = pOptr->info; execInfo->blockRecorder = pInfo->readRecorder; execInfo->sortExecInfo = tsortGetSortExecInfo(pInfo->pSortHandle); *pOptrExplain = execInfo; *len = sizeof(STableMergeScanExecInfo); - + return TSDB_CODE_SUCCESS; } @@ -2305,8 +2306,7 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc; int32_t numOfCols = 0; - SArray* pColList = - extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID); + SArray* pColList = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID); int32_t code = initQueryTableDataCond(&pInfo->cond, pTableScanNode); if (code != TSDB_CODE_SUCCESS) { @@ -2320,16 +2320,16 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN pInfo->scanInfo = (SScanInfo){.numOfAsc = pTableScanNode->scanSeq[0], .numOfDesc = pTableScanNode->scanSeq[1]}; - pInfo->readHandle = *readHandle; - pInfo->interval = extractIntervalInfo(pTableScanNode); + pInfo->readHandle = *readHandle; + pInfo->interval = extractIntervalInfo(pTableScanNode); pInfo->sample.sampleRatio = pTableScanNode->ratio; - pInfo->sample.seed = taosGetTimestampSec(); - pInfo->dataBlockLoadFlag = pTableScanNode->dataRequired; - pInfo->pFilterNode = pTableScanNode->scan.node.pConditions; - pInfo->dataReaders = dataReaders; - pInfo->scanFlag = MAIN_SCAN; - pInfo->pColMatchInfo = pColList; - pInfo->curTWinIdx = 0; + pInfo->sample.seed = taosGetTimestampSec(); + pInfo->dataBlockLoadFlag = pTableScanNode->dataRequired; + pInfo->pFilterNode = pTableScanNode->scan.node.pConditions; + pInfo->dataReaders = dataReaders; + pInfo->scanFlag = MAIN_SCAN; + pInfo->pColMatchInfo = pColList; + pInfo->curTWinIdx = 0; pInfo->pResBlock = createResDataBlock(pDescNode); @@ -2347,22 +2347,22 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN pInfo->pSortInputBlock = createOneDataBlock(pInfo->pResBlock, false); int32_t rowSize = pInfo->pResBlock->info.rowSize; - pInfo->bufPageSize = getProperSortPageSize(rowSize); + pInfo->bufPageSize = getProperSortPageSize(rowSize); // todo the total available buffer should be determined by total capacity of buffer of this task. // the additional one is reserved for merge result - pInfo->sortBufSize = pInfo->bufPageSize * (taosArrayGetSize(dataReaders) + 1); - pInfo->hasGroupId = false; + pInfo->sortBufSize = pInfo->bufPageSize * (taosArrayGetSize(dataReaders) + 1); + pInfo->hasGroupId = false; pInfo->prefetchedTuple = NULL; - pOperator->name = "TableMergeScanOperator"; + pOperator->name = "TableMergeScanOperator"; // TODO : change it pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN; - pOperator->blocking = false; - pOperator->status = OP_NOT_OPENED; - pOperator->info = pInfo; - pOperator->exprSupp.numOfExprs = numOfCols; - pOperator->pTaskInfo = pTaskInfo; + pOperator->blocking = false; + pOperator->status = OP_NOT_OPENED; + pOperator->info = pInfo; + pOperator->exprSupp.numOfExprs = numOfCols; + pOperator->pTaskInfo = pTaskInfo; initResultSizeInfo(pOperator, 1024); pOperator->fpSet = diff --git a/source/libs/stream/src/streamTask.c b/source/libs/stream/src/streamTask.c index a35e7679a121c9799a61a7973cbb1b3fc3898a5a..b5a63d937ae6be238406839921ba8a514568001a 100644 --- a/source/libs/stream/src/streamTask.c +++ b/source/libs/stream/src/streamTask.c @@ -36,11 +36,11 @@ int32_t tEncodeSStreamTask(SEncoder* pEncoder, const SStreamTask* pTask) { if (tEncodeI32(pEncoder, pTask->taskId) < 0) return -1; if (tEncodeI8(pEncoder, pTask->inputType) < 0) return -1; if (tEncodeI8(pEncoder, pTask->status) < 0) return -1; - if (tEncodeI8(pEncoder, pTask->sourceType) < 0) return -1; if (tEncodeI8(pEncoder, pTask->execType) < 0) return -1; if (tEncodeI8(pEncoder, pTask->sinkType) < 0) return -1; if (tEncodeI8(pEncoder, pTask->dispatchType) < 0) return -1; if (tEncodeI16(pEncoder, pTask->dispatchMsgType) < 0) return -1; + if (tEncodeI8(pEncoder, pTask->dataScan) < 0) return -1; if (tEncodeI32(pEncoder, pTask->childId) < 0) return -1; if (tEncodeI32(pEncoder, pTask->nodeId) < 0) return -1; @@ -84,11 +84,11 @@ int32_t tDecodeSStreamTask(SDecoder* pDecoder, SStreamTask* pTask) { if (tDecodeI32(pDecoder, &pTask->taskId) < 0) return -1; if (tDecodeI8(pDecoder, &pTask->inputType) < 0) return -1; if (tDecodeI8(pDecoder, &pTask->status) < 0) return -1; - if (tDecodeI8(pDecoder, &pTask->sourceType) < 0) return -1; if (tDecodeI8(pDecoder, &pTask->execType) < 0) return -1; if (tDecodeI8(pDecoder, &pTask->sinkType) < 0) return -1; if (tDecodeI8(pDecoder, &pTask->dispatchType) < 0) return -1; if (tDecodeI16(pDecoder, &pTask->dispatchMsgType) < 0) return -1; + if (tDecodeI8(pDecoder, &pTask->dataScan) < 0) return -1; if (tDecodeI32(pDecoder, &pTask->childId) < 0) return -1; if (tDecodeI32(pDecoder, &pTask->nodeId) < 0) return -1;