From 609b21647bb771bbb4487c18fc6f68103de33e9e Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Mon, 25 Apr 2022 15:40:25 +0800 Subject: [PATCH] fix: msg with schema --- example/src/tmq.c | 7 +- include/common/tmsg.h | 110 ++++++--------------- source/client/inc/clientInt.h | 4 + source/client/src/clientImpl.c | 29 +++--- source/client/src/tmq.c | 6 +- source/dnode/mnode/impl/src/mndScheduler.c | 91 ++++++++--------- source/dnode/mnode/impl/src/mndTopic.c | 7 +- source/dnode/vnode/src/tq/tq.c | 30 ++++-- source/dnode/vnode/src/tq/tqRead.c | 66 +++++++++---- 9 files changed, 178 insertions(+), 172 deletions(-) diff --git a/example/src/tmq.c b/example/src/tmq.c index 32390f71c9..56f210081b 100644 --- a/example/src/tmq.c +++ b/example/src/tmq.c @@ -103,8 +103,8 @@ int32_t create_topic() { } taos_free_result(pRes); - /*pRes = taos_query(pConn, "create topic topic_ctb_column as abc1");*/ - pRes = taos_query(pConn, "create topic topic_ctb_column as select ts, c1, c2, c3 from ct1"); + pRes = taos_query(pConn, "create topic topic_ctb_column as abc1"); + /*pRes = taos_query(pConn, "create topic topic_ctb_column as select ts, c1, c2, c3 from ct1");*/ if (taos_errno(pRes) != 0) { printf("failed to create topic topic_ctb_column, reason:%s\n", taos_errstr(pRes)); return -1; @@ -162,9 +162,10 @@ tmq_t* build_consumer() { tmq_conf_set(conf, "group.id", "tg2"); tmq_conf_set(conf, "td.connect.user", "root"); tmq_conf_set(conf, "td.connect.pass", "taosdata"); - tmq_conf_set(conf, "td.connect.db", "abc1"); + /*tmq_conf_set(conf, "td.connect.db", "abc1");*/ tmq_conf_set_offset_commit_cb(conf, tmq_commit_cb_print); tmq_t* tmq = tmq_consumer_new(conf, NULL, 0); + assert(tmq); return tmq; } diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 5a2dd502c1..a07101f404 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -2052,80 +2052,6 @@ static FORCE_INLINE void* tDecodeSMqSetCVgReq(void* buf, SMqSetCVgReq* pReq) { return buf; } -typedef struct { - int64_t leftForVer; - int32_t vgId; - int32_t epoch; - int64_t consumerId; - char topicName[TSDB_TOPIC_FNAME_LEN]; -} SMqCancelConnReq; - -static FORCE_INLINE int32_t tEncodeSMqCancelConnReq(void** buf, const SMqCancelConnReq* pReq) { - int32_t tlen = 0; - tlen += taosEncodeFixedI64(buf, pReq->leftForVer); - tlen += taosEncodeFixedI32(buf, pReq->vgId); - tlen += taosEncodeFixedI32(buf, pReq->epoch); - tlen += taosEncodeFixedI64(buf, pReq->consumerId); - tlen += taosEncodeString(buf, pReq->topicName); - return tlen; -} - -static FORCE_INLINE void* tDecodeSMqCancelConnReq(void* buf, SMqCancelConnReq* pReq) { - buf = taosDecodeFixedI64(buf, &pReq->leftForVer); - buf = taosDecodeFixedI32(buf, &pReq->vgId); - buf = taosDecodeFixedI32(buf, &pReq->epoch); - buf = taosDecodeFixedI64(buf, &pReq->consumerId); - buf = taosDecodeStringTo(buf, pReq->topicName); - return buf; -} - -typedef struct { - int8_t reserved; -} SMqCancelConnRsp; - -typedef struct { - int64_t leftForVer; - int32_t vgId; - int64_t oldConsumerId; - int64_t newConsumerId; - char* topic; -} SMqMVRebReq; - -static FORCE_INLINE int32_t tEncodeSMqMVRebReq(void** buf, const SMqMVRebReq* pReq) { - int32_t tlen = 0; - tlen += taosEncodeFixedI64(buf, pReq->leftForVer); - tlen += taosEncodeFixedI32(buf, pReq->vgId); - tlen += taosEncodeFixedI64(buf, pReq->oldConsumerId); - tlen += taosEncodeFixedI64(buf, pReq->newConsumerId); - tlen += taosEncodeString(buf, pReq->topic); - return tlen; -} - -static FORCE_INLINE void* tDecodeSMqMVRebReq(void* buf, SMqMVRebReq* pReq) { - buf = taosDecodeFixedI64(buf, &pReq->leftForVer); - buf = taosDecodeFixedI32(buf, &pReq->vgId); - buf = taosDecodeFixedI64(buf, &pReq->oldConsumerId); - buf = taosDecodeFixedI64(buf, &pReq->newConsumerId); - buf = taosDecodeString(buf, &pReq->topic); - return buf; -} - -typedef struct { - SMsgHead header; - int32_t vgId; - int64_t consumerId; - char topicName[TSDB_TOPIC_FNAME_LEN]; - char cgroup[TSDB_CGROUP_LEN]; -} SMqSetCVgRsp; - -typedef struct { - SMsgHead header; - int32_t vgId; - int64_t consumerId; - char topicName[TSDB_TOPIC_FNAME_LEN]; - char cgroup[TSDB_CGROUP_LEN]; -} SMqMVRebRsp; - typedef struct { int32_t vgId; int64_t offset; @@ -2152,6 +2078,24 @@ typedef struct { SSchema* pSchema; } SSchemaWrapper; +static FORCE_INLINE SSchemaWrapper* tCloneSSchemaWrapper(const SSchemaWrapper* pSchemaWrapper) { + SSchemaWrapper* pSW = (SSchemaWrapper*)taosMemoryMalloc(sizeof(SSchemaWrapper)); + if (pSW == NULL) return pSW; + pSW->nCols = pSchemaWrapper->nCols; + pSW->pSchema = (SSchema*)taosMemoryCalloc(pSW->nCols, sizeof(SSchema)); + if (pSW->pSchema == NULL) { + taosMemoryFree(pSW); + return NULL; + } + memcpy(pSW->pSchema, pSchemaWrapper->pSchema, pSW->nCols * sizeof(SSchema)); + return pSW; +} + +static FORCE_INLINE void tDeleteSSchemaWrapper(SSchemaWrapper* pSchemaWrapper) { + taosMemoryFree(pSchemaWrapper->pSchema); + taosMemoryFree(pSchemaWrapper); +} + static FORCE_INLINE int32_t taosEncodeSSchema(void** buf, const SSchema* pSchema) { int32_t tlen = 0; tlen += taosEncodeFixedI8(buf, pSchema->type); @@ -2162,13 +2106,13 @@ static FORCE_INLINE int32_t taosEncodeSSchema(void** buf, const SSchema* pSchema return tlen; } -static FORCE_INLINE void* taosDecodeSSchema(void* buf, SSchema* pSchema) { +static FORCE_INLINE void* taosDecodeSSchema(const void* buf, SSchema* pSchema) { buf = taosDecodeFixedI8(buf, &pSchema->type); buf = taosDecodeFixedI8(buf, &pSchema->flags); buf = taosDecodeFixedI32(buf, &pSchema->bytes); buf = taosDecodeFixedI16(buf, &pSchema->colId); buf = taosDecodeStringTo(buf, pSchema->name); - return buf; + return (void*)buf; } static FORCE_INLINE int32_t tEncodeSSchema(SCoder* pEncoder, const SSchema* pSchema) { @@ -2198,7 +2142,7 @@ static FORCE_INLINE int32_t taosEncodeSSchemaWrapper(void** buf, const SSchemaWr return tlen; } -static FORCE_INLINE void* taosDecodeSSchemaWrapper(void* buf, SSchemaWrapper* pSW) { +static FORCE_INLINE void* taosDecodeSSchemaWrapper(const void* buf, SSchemaWrapper* pSW) { buf = taosDecodeFixedU32(buf, &pSW->nCols); pSW->pSchema = (SSchema*)taosMemoryCalloc(pSW->nCols, sizeof(SSchema)); if (pSW->pSchema == NULL) { @@ -2208,7 +2152,7 @@ static FORCE_INLINE void* taosDecodeSSchemaWrapper(void* buf, SSchemaWrapper* pS for (int32_t i = 0; i < pSW->nCols; i++) { buf = taosDecodeSSchema(buf, &pSW->pSchema[i]); } - return buf; + return (void*)buf; } static FORCE_INLINE int32_t tEncodeSSchemaWrapper(SCoder* pEncoder, const SSchemaWrapper* pSW) { @@ -2615,6 +2559,10 @@ static FORCE_INLINE int32_t tEncodeSMqDataBlkRsp(void** buf, const SMqDataBlkRsp void* data = taosArrayGetP(pRsp->blockData, i); tlen += taosEncodeFixedI32(buf, bLen); tlen += taosEncodeBinary(buf, data, bLen); + if (pRsp->withSchema) { + SSchemaWrapper* pSW = (SSchemaWrapper*)taosArrayGetP(pRsp->blockSchema, i); + tlen += taosEncodeSSchemaWrapper(buf, pSW); + } } } return tlen; @@ -2627,6 +2575,7 @@ static FORCE_INLINE void* tDecodeSMqDataBlkRsp(const void* buf, SMqDataBlkRsp* p buf = taosDecodeFixedI32(buf, &pRsp->blockNum); pRsp->blockData = taosArrayInit(pRsp->blockNum, sizeof(void*)); pRsp->blockDataLen = taosArrayInit(pRsp->blockNum, sizeof(void*)); + pRsp->blockSchema = taosArrayInit(pRsp->blockNum, sizeof(void*)); if (pRsp->blockNum != 0) { buf = taosDecodeFixedI8(buf, &pRsp->withTbName); buf = taosDecodeFixedI8(buf, &pRsp->withSchema); @@ -2639,6 +2588,11 @@ static FORCE_INLINE void* tDecodeSMqDataBlkRsp(const void* buf, SMqDataBlkRsp* p buf = taosDecodeBinary(buf, &data, bLen); taosArrayPush(pRsp->blockDataLen, &bLen); taosArrayPush(pRsp->blockData, &data); + if (pRsp->withSchema) { + SSchemaWrapper* pSW = (SSchemaWrapper*)taosMemoryMalloc(sizeof(SSchemaWrapper)); + buf = taosDecodeSSchemaWrapper(buf, pSW); + taosArrayPush(pRsp->blockSchema, &pSW); + } } } return (void*)buf; diff --git a/source/client/inc/clientInt.h b/source/client/inc/clientInt.h index 814caf330a..c0237a184a 100644 --- a/source/client/inc/clientInt.h +++ b/source/client/inc/clientInt.h @@ -231,6 +231,10 @@ static FORCE_INLINE SReqResultInfo* tmqGetNextResInfo(TAOS_RES* res, bool conver msg->resIter++; if (msg->resIter < msg->rsp.blockNum) { SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)taosArrayGetP(msg->rsp.blockData, msg->resIter); + if (msg->rsp.withSchema) { + SSchemaWrapper* pSW = (SSchemaWrapper*)taosArrayGetP(msg->rsp.blockSchema, msg->resIter); + setResSchemaInfo(&msg->resInfo, pSW->pSchema, pSW->nCols); + } setQueryResultFromRsp(&msg->resInfo, pRetrieve, convertUcs4); return &msg->resInfo; } diff --git a/source/client/src/clientImpl.c b/source/client/src/clientImpl.c index 2ae9a11803..a40edbb65a 100644 --- a/source/client/src/clientImpl.c +++ b/source/client/src/clientImpl.c @@ -226,17 +226,15 @@ int32_t execDdlQuery(SRequestObj* pRequest, SQuery* pQuery) { int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArray* pNodeList) { pRequest->type = pQuery->msgType; - SPlanContext cxt = { - .queryId = pRequest->requestId, - .acctId = pRequest->pTscObj->acctId, - .mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp), - .pAstRoot = pQuery->pRoot, - .showRewrite = pQuery->showRewrite, - .pTransporter = pRequest->pTscObj->pAppInfo->pTransporter, - .pMsg = pRequest->msgBuf, - .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE - }; - int32_t code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &cxt.pCatalog); + SPlanContext cxt = {.queryId = pRequest->requestId, + .acctId = pRequest->pTscObj->acctId, + .mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp), + .pAstRoot = pQuery->pRoot, + .showRewrite = pQuery->showRewrite, + .pTransporter = pRequest->pTscObj->pAppInfo->pTransporter, + .pMsg = pRequest->msgBuf, + .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE}; + int32_t code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &cxt.pCatalog); if (TSDB_CODE_SUCCESS == code) { code = qCreateQueryPlan(&cxt, pPlan, pNodeList); } @@ -247,6 +245,7 @@ void setResSchemaInfo(SReqResultInfo* pResInfo, const SSchema* pSchema, int32_t ASSERT(pSchema != NULL && numOfCols > 0); pResInfo->numOfCols = numOfCols; + // TODO handle memory leak pResInfo->fields = taosMemoryCalloc(numOfCols, sizeof(TAOS_FIELD)); pResInfo->userFields = taosMemoryCalloc(numOfCols, sizeof(TAOS_FIELD)); @@ -282,7 +281,7 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList SQueryResult res = {.code = 0, .numOfRows = 0, .msgSize = ERROR_MSG_BUF_DEFAULT_SIZE, .msg = pRequest->msgBuf}; int32_t code = schedulerExecJob(pTransporter, pNodeList, pDag, &pRequest->body.queryJob, pRequest->sqlstr, - pRequest->metric.start, &res); + pRequest->metric.start, &res); if (code != TSDB_CODE_SUCCESS) { if (pRequest->body.queryJob != 0) { schedulerFreeJob(pRequest->body.queryJob); @@ -840,12 +839,12 @@ int32_t setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32 return code; } - char* p = (char*) pResultInfo->pData; + char* p = (char*)pResultInfo->pData; - int32_t dataLen = *(int32_t*) p; + int32_t dataLen = *(int32_t*)p; p += sizeof(int32_t); - uint64_t groupId = *(uint64_t*) p; + uint64_t groupId = *(uint64_t*)p; p += sizeof(uint64_t); int32_t* colLength = (int32_t*)p; diff --git a/source/client/src/tmq.c b/source/client/src/tmq.c index d576f28cb4..988406c84d 100644 --- a/source/client/src/tmq.c +++ b/source/client/src/tmq.c @@ -376,7 +376,7 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) { ASSERT(user); ASSERT(pass); - ASSERT(conf->db); + /*ASSERT(conf->db);*/ ASSERT(conf->groupId[0]); pTmq->pTscObj = taos_connect_internal(conf->ip, user, pass, NULL, conf->db, conf->port, CONN_TYPE__TMQ); @@ -1118,7 +1118,9 @@ SMqRspObj* tmqBuildRspFromWrapper(SMqPollRspWrapper* pWrapper) { pRspObj->resInfo.totalRows = 0; pRspObj->resInfo.precision = TSDB_TIME_PRECISION_MILLI; - setResSchemaInfo(&pRspObj->resInfo, pWrapper->topicHandle->schema.pSchema, pWrapper->topicHandle->schema.nCols); + if (!pWrapper->msg.withSchema) { + setResSchemaInfo(&pRspObj->resInfo, pWrapper->topicHandle->schema.pSchema, pWrapper->topicHandle->schema.nCols); + } taosFreeQitem(pWrapper); return pRspObj; diff --git a/source/dnode/mnode/impl/src/mndScheduler.c b/source/dnode/mnode/impl/src/mndScheduler.c index 73583058f1..3dff65866c 100644 --- a/source/dnode/mnode/impl/src/mndScheduler.c +++ b/source/dnode/mnode/impl/src/mndScheduler.c @@ -476,33 +476,37 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) { int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscribeObj* pSub) { SSdb* pSdb = pMnode->pSdb; SVgObj* pVgroup = NULL; - SQueryPlan* pPlan = qStringToQueryPlan(pTopic->physicalPlan); - if (pPlan == NULL) { - terrno = TSDB_CODE_QRY_INVALID_INPUT; - return -1; + SQueryPlan* pPlan = NULL; + SSubplan* plan = NULL; + if (pTopic->subType == TOPIC_SUB_TYPE__TABLE) { + pPlan = qStringToQueryPlan(pTopic->physicalPlan); + if (pPlan == NULL) { + terrno = TSDB_CODE_QRY_INVALID_INPUT; + return -1; + } + + ASSERT(pSub->vgNum == -1); + + pSub->vgNum = 0; + + int32_t levelNum = LIST_LENGTH(pPlan->pSubplans); + if (levelNum != 1) { + qDestroyQueryPlan(pPlan); + terrno = TSDB_CODE_MND_UNSUPPORTED_TOPIC; + return -1; + } + + SNodeListNode* inner = nodesListGetNode(pPlan->pSubplans, 0); + + int32_t opNum = LIST_LENGTH(inner->pNodeList); + if (opNum != 1) { + qDestroyQueryPlan(pPlan); + terrno = TSDB_CODE_MND_UNSUPPORTED_TOPIC; + return -1; + } + plan = nodesListGetNode(inner->pNodeList, 0); } - ASSERT(pSub->vgNum == -1); - - pSub->vgNum = 0; - - int32_t levelNum = LIST_LENGTH(pPlan->pSubplans); - if (levelNum != 1) { - qDestroyQueryPlan(pPlan); - terrno = TSDB_CODE_MND_UNSUPPORTED_TOPIC; - return -1; - } - - SNodeListNode* inner = nodesListGetNode(pPlan->pSubplans, 0); - - int32_t opNum = LIST_LENGTH(inner->pNodeList); - if (opNum != 1) { - qDestroyQueryPlan(pPlan); - terrno = TSDB_CODE_MND_UNSUPPORTED_TOPIC; - return -1; - } - SSubplan* plan = nodesListGetNode(inner->pNodeList, 0); - int64_t unexistKey = -1; SMqConsumerEpInSub* pEpInSub = taosHashGet(pSub->consumerHash, &unexistKey, sizeof(int64_t)); ASSERT(pEpInSub); @@ -519,38 +523,35 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib } pSub->vgNum++; - plan->execNode.nodeId = pVgroup->vgId; - plan->execNode.epSet = mndGetVgroupEpset(pMnode, pVgroup); SMqVgEp* pVgEp = taosMemoryMalloc(sizeof(SMqVgEp)); - pVgEp->epSet = plan->execNode.epSet; - pVgEp->vgId = plan->execNode.nodeId; - -#if 0 - SMqConsumerEp consumerEp = {0}; - consumerEp.status = 0; - consumerEp.consumerId = -1; - consumerEp.epSet = plan->execNode.epSet; - consumerEp.vgId = plan->execNode.nodeId; -#endif + pVgEp->epSet = mndGetVgroupEpset(pMnode, pVgroup); + pVgEp->vgId = pVgroup->vgId; + taosArrayPush(pEpInSub->vgs, &pVgEp); mDebug("init subscribption %s, assign vg: %d", pSub->key, pVgEp->vgId); - int32_t msgLen; - if (qSubPlanToString(plan, &pVgEp->qmsg, &msgLen) < 0) { - sdbRelease(pSdb, pVgroup); - qDestroyQueryPlan(pPlan); - terrno = TSDB_CODE_QRY_INVALID_INPUT; - return -1; + if (pTopic->subType == TOPIC_SUB_TYPE__TABLE) { + int32_t msgLen; + + plan->execNode.epSet = pVgEp->epSet; + plan->execNode.nodeId = pVgEp->vgId; + + if (qSubPlanToString(plan, &pVgEp->qmsg, &msgLen) < 0) { + sdbRelease(pSdb, pVgroup); + qDestroyQueryPlan(pPlan); + terrno = TSDB_CODE_QRY_INVALID_INPUT; + return -1; + } + } else { + pVgEp->qmsg = strdup(""); } - taosArrayPush(pEpInSub->vgs, &pVgEp); ASSERT(taosHashGetSize(pSub->consumerHash) == 1); /*taosArrayPush(pSub->unassignedVg, &consumerEp);*/ } - ASSERT(pEpInSub->vgs->size > 0); pEpInSub = taosHashGet(pSub->consumerHash, &unexistKey, sizeof(int64_t)); ASSERT(pEpInSub->vgs->size > 0); diff --git a/source/dnode/mnode/impl/src/mndTopic.c b/source/dnode/mnode/impl/src/mndTopic.c index faddc3a1c5..e7a37658dc 100644 --- a/source/dnode/mnode/impl/src/mndTopic.c +++ b/source/dnode/mnode/impl/src/mndTopic.c @@ -282,10 +282,10 @@ static int32_t mndCreateTopic(SMnode *pMnode, SNodeMsg *pReq, SCMCreateTopicReq topicObj.version = 1; topicObj.sql = strdup(pCreate->sql); topicObj.sqlLen = strlen(pCreate->sql) + 1; - topicObj.ast = strdup(pCreate->ast); - topicObj.astLen = strlen(pCreate->ast) + 1; if (pCreate->ast && pCreate->ast[0]) { + topicObj.ast = strdup(pCreate->ast); + topicObj.astLen = strlen(pCreate->ast) + 1; topicObj.subType = TOPIC_SUB_TYPE__TABLE; topicObj.withTbName = 0; topicObj.withSchema = 0; @@ -314,6 +314,9 @@ static int32_t mndCreateTopic(SMnode *pMnode, SNodeMsg *pReq, SCMCreateTopicReq return -1; } } else { + topicObj.ast = strdup(""); + topicObj.astLen = 1; + topicObj.physicalPlan = strdup(""); topicObj.subType = TOPIC_SUB_TYPE__DB; topicObj.withTbName = 1; topicObj.withSchema = 1; diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index 434c5bc5bb..fdb43a761b 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -411,8 +411,10 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) { SMqDataBlkRsp rsp = {0}; rsp.reqOffset = pReq->currentOffset; + rsp.withSchema = pExec->withSchema; rsp.blockData = taosArrayInit(0, sizeof(void*)); rsp.blockDataLen = taosArrayInit(0, sizeof(int32_t)); + rsp.blockSchema = taosArrayInit(0, sizeof(void*)); while (1) { consumerEpoch = atomic_load_32(&pExec->epoch); @@ -512,6 +514,12 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) { ASSERT(actualLen <= dataStrLen); taosArrayPush(rsp.blockDataLen, &actualLen); taosArrayPush(rsp.blockData, &buf); + + if (pExec->withSchema) { + SSchemaWrapper* pSW = tCloneSSchemaWrapper(pExec->pExecReader[workerId]->pSchemaWrapper); + taosArrayPush(rsp.blockSchema, &pSW); + } + rsp.blockNum++; } // db subscribe @@ -540,6 +548,10 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) { ASSERT(actualLen <= dataStrLen); taosArrayPush(rsp.blockDataLen, &actualLen); taosArrayPush(rsp.blockData, &buf); + + SSchemaWrapper* pSW = tCloneSSchemaWrapper(pExec->pExecReader[workerId]->pSchemaWrapper); + taosArrayPush(rsp.blockSchema, &pSW); + rsp.blockNum++; } } else { @@ -586,6 +598,8 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) { // TODO destroy taosArrayDestroy(rsp.blockData); taosArrayDestroy(rsp.blockDataLen); + taosArrayDestroyP(rsp.blockSchema, (FDelete)tDeleteSSchemaWrapper); + return 0; } @@ -827,12 +841,16 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) { pExec->pWalReader = walOpenReadHandle(pTq->pVnode->pWal); for (int32_t i = 0; i < 5; i++) { pExec->pExecReader[i] = tqInitSubmitMsgScanner(pTq->pVnode->pMeta); - SReadHandle handle = { - .reader = pExec->pExecReader[i], - .meta = pTq->pVnode->pMeta, - }; - pExec->task[i] = qCreateStreamExecTaskInfo(pExec->qmsg, &handle); - ASSERT(pExec->task[i]); + if (pExec->subType == TOPIC_SUB_TYPE__TABLE) { + SReadHandle handle = { + .reader = pExec->pExecReader[i], + .meta = pTq->pVnode->pMeta, + }; + pExec->task[i] = qCreateStreamExecTaskInfo(pExec->qmsg, &handle); + ASSERT(pExec->task[i]); + } else { + pExec->task[i] = NULL; + } } taosHashPut(pTq->execs, req.subKey, strlen(req.subKey), pExec, sizeof(STqExec)); return 0; diff --git a/source/dnode/vnode/src/tq/tqRead.c b/source/dnode/vnode/src/tq/tqRead.c index eb45577e0a..ef7e346867 100644 --- a/source/dnode/vnode/src/tq/tqRead.c +++ b/source/dnode/vnode/src/tq/tqRead.c @@ -65,7 +65,9 @@ bool tqNextDataBlock(STqReadHandle* pHandle) { /*pHandle->pBlock->uid = htobe64(pHandle->pBlock->uid);*/ /*if (pHandle->tbUid == pHandle->pBlock->uid) {*/ - ASSERT(pHandle->tbIdHash); + if (pHandle->tbIdHash == NULL) { + return true; + } void* ret = taosHashGet(pHandle->tbIdHash, &pHandle->pBlock->uid, sizeof(int64_t)); if (ret != NULL) { /*printf("retrieve one tb %ld\n", pHandle->pBlock->uid);*/ @@ -107,26 +109,15 @@ int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* p *pNumOfRows = pHandle->pBlock->numOfRows; int32_t colNumNeed = taosArrayGetSize(pHandle->pColIdList); - if (colNumNeed > pSchemaWrapper->nCols) { - colNumNeed = pSchemaWrapper->nCols; - } + if (colNumNeed == 0) { + *ppCols = taosArrayInit(pSchemaWrapper->nCols, sizeof(SColumnInfoData)); + if (*ppCols == NULL) { + return -1; + } - *ppCols = taosArrayInit(colNumNeed, sizeof(SColumnInfoData)); - if (*ppCols == NULL) { - return -1; - } - - int32_t colMeta = 0; - int32_t colNeed = 0; - while (colMeta < pSchemaWrapper->nCols && colNeed < colNumNeed) { - SSchema* pColSchema = &pSchemaWrapper->pSchema[colMeta]; - col_id_t colIdSchema = pColSchema->colId; - col_id_t colIdNeed = *(col_id_t*)taosArrayGet(pHandle->pColIdList, colNeed); - if (colIdSchema < colIdNeed) { - colMeta++; - } else if (colIdSchema > colIdNeed) { - colNeed++; - } else { + int32_t colMeta = 0; + while (colMeta < pSchemaWrapper->nCols) { + SSchema* pColSchema = &pSchemaWrapper->pSchema[colMeta]; SColumnInfoData colInfo = {0}; colInfo.info.bytes = pColSchema->bytes; colInfo.info.colId = pColSchema->colId; @@ -137,7 +128,40 @@ int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* p } taosArrayPush(*ppCols, &colInfo); colMeta++; - colNeed++; + } + } else { + if (colNumNeed > pSchemaWrapper->nCols) { + colNumNeed = pSchemaWrapper->nCols; + } + + *ppCols = taosArrayInit(colNumNeed, sizeof(SColumnInfoData)); + if (*ppCols == NULL) { + return -1; + } + + int32_t colMeta = 0; + int32_t colNeed = 0; + while (colMeta < pSchemaWrapper->nCols && colNeed < colNumNeed) { + SSchema* pColSchema = &pSchemaWrapper->pSchema[colMeta]; + col_id_t colIdSchema = pColSchema->colId; + col_id_t colIdNeed = *(col_id_t*)taosArrayGet(pHandle->pColIdList, colNeed); + if (colIdSchema < colIdNeed) { + colMeta++; + } else if (colIdSchema > colIdNeed) { + colNeed++; + } else { + SColumnInfoData colInfo = {0}; + colInfo.info.bytes = pColSchema->bytes; + colInfo.info.colId = pColSchema->colId; + colInfo.info.type = pColSchema->type; + + if (colInfoDataEnsureCapacity(&colInfo, 0, *pNumOfRows) < 0) { + goto FAIL; + } + taosArrayPush(*ppCols, &colInfo); + colMeta++; + colNeed++; + } } }