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

enh: support delete msg for rsma

This commit is contained in:
kailixu 2023-11-09 19:51:01 +08:00
parent 9cf154543f
commit 572eb691fb
8 changed files with 121 additions and 111 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source/dnode/vnode/src/inc/tq.h
View File

@ -160,7 +160,7 @@ int32_t tqResetStreamTaskStatus(STQ* pTq);
int32_t tqStopStreamTasks(STQ* pTq); int32_t tqStopStreamTasks(STQ* pTq);
// tq util // tq util
int32_t extractDelDataBlock(const void* pData, int32_t len, int64_t ver, SStreamRefDataBlock** pRefBlock); int32_t extractDelDataBlock(const void* pData, int32_t len, int64_t ver, void** pRefBlock, int8_t type);
int32_t tqExtractDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest, SRpcMsg* pMsg); int32_t tqExtractDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest, SRpcMsg* pMsg);
int32_t tqDoSendDataRsp(const SRpcHandleInfo* pRpcHandleInfo, const SMqDataRsp* pRsp, int32_t epoch, int64_t consumerId, int32_t tqDoSendDataRsp(const SRpcHandleInfo* pRpcHandleInfo, const SMqDataRsp* pRsp, int32_t epoch, int64_t consumerId,
int32_t type, int64_t sver, int64_t ever); int32_t type, int64_t sver, int64_t ever);

4
source/dnode/vnode/src/inc/vnodeInt.h
View File

@ -286,8 +286,8 @@ int32_t tdProcessTSmaCreate(SSma* pSma, int64_t version, const char* msg);
int32_t tdProcessTSmaInsert(SSma* pSma, int64_t indexUid, const char* msg); int32_t tdProcessTSmaInsert(SSma* pSma, int64_t indexUid, const char* msg);
int32_t tdProcessRSmaCreate(SSma* pSma, SVCreateStbReq* pReq); int32_t tdProcessRSmaCreate(SSma* pSma, SVCreateStbReq* pReq);
int32_t tdProcessRSmaSubmit(SSma* pSma, int64_t version, void* pReq, void* pMsg, int32_t len, int32_t inputType); int32_t tdProcessRSmaSubmit(SSma* pSma, int64_t version, void* pReq, void* pMsg, int32_t len);
int32_t tdProcessRSmaDelete(SSma* pSma, int64_t version, void* pReq, void* pMsg, int32_t len, int32_t inputType); int32_t tdProcessRSmaDelete(SSma* pSma, int64_t version, void* pReq, void* pMsg, int32_t len);
int32_t tdProcessRSmaDrop(SSma* pSma, SVDropStbReq* pReq); int32_t tdProcessRSmaDrop(SSma* pSma, SVDropStbReq* pReq);
int32_t tdFetchTbUidList(SSma* pSma, STbUidStore** ppStore, tb_uid_t suid, tb_uid_t uid); int32_t tdFetchTbUidList(SSma* pSma, STbUidStore** ppStore, tb_uid_t suid, tb_uid_t uid);
int32_t tdUpdateTbUidList(SSma* pSma, STbUidStore* pUidStore, bool isAdd); int32_t tdUpdateTbUidList(SSma* pSma, STbUidStore* pUidStore, bool isAdd);

5
source/dnode/vnode/src/sma/smaCommit.c
View File

@ -217,10 +217,7 @@ static int32_t tdProcessRSmaAsyncCommitImpl(SSma *pSma, SCommitInfo *pInfo) {
int32_t lino = 0; int32_t lino = 0;
SVnode *pVnode = pSma->pVnode; SVnode *pVnode = pSma->pVnode;
SSmaEnv *pSmaEnv = SMA_RSMA_ENV(pSma); if (!SMA_RSMA_ENV(pSma)) goto _exit;
if (!pSmaEnv) {
goto _exit;
}
code = tsdbCommitBegin(VND_RSMA1(pVnode), pInfo); code = tsdbCommitBegin(VND_RSMA1(pVnode), pInfo);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);

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

@ -43,7 +43,7 @@ static int32_t tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int32_t msgSiz
ERsmaExecType type, int8_t level); ERsmaExecType type, int8_t level);
static SRSmaInfo *tdAcquireRSmaInfoBySuid(SSma *pSma, int64_t suid); static SRSmaInfo *tdAcquireRSmaInfoBySuid(SSma *pSma, int64_t suid);
static void tdReleaseRSmaInfo(SSma *pSma, SRSmaInfo *pInfo); static void tdReleaseRSmaInfo(SSma *pSma, SRSmaInfo *pInfo);
static void tdFreeRSmaSubmitItems(SArray *pItems); static void tdFreeRSmaSubmitItems(SArray *pItems, int32_t type);
static int32_t tdRSmaFetchAllResult(SSma *pSma, SRSmaInfo *pInfo); static int32_t tdRSmaFetchAllResult(SSma *pSma, SRSmaInfo *pInfo);
static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSmaInfoItem *pItem, SRSmaInfo *pInfo, static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSmaInfoItem *pItem, SRSmaInfo *pInfo,
int32_t execType, int8_t *streamFlushed); int32_t execType, int8_t *streamFlushed);
@ -723,7 +723,7 @@ _exit:
*/ */
static int32_t tdExecuteRSmaImplAsync(SSma *pSma, int64_t version, const void *pMsg, int32_t len, int32_t inputType, static int32_t tdExecuteRSmaImplAsync(SSma *pSma, int64_t version, const void *pMsg, int32_t len, int32_t inputType,
SRSmaInfo *pInfo, tb_uid_t suid) { SRSmaInfo *pInfo, tb_uid_t suid) {
int32_t size = RSMA_SUBMIT_HEAD_LEN + len; // header(type+len+version) + payload int32_t size = RSMA_SUBMIT_HEAD_LEN + len; // header + payload
void *qItem = taosAllocateQitem(size, DEF_QITEM, 0); void *qItem = taosAllocateQitem(size, DEF_QITEM, 0);
if (!qItem) { if (!qItem) {
@ -940,12 +940,8 @@ static int32_t tdExecuteRSmaAsync(SSma *pSma, int64_t version, const void *pMsg,
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t tdProcessRSmaSubmit(SSma *pSma, int64_t version, void *pReq, void *pMsg, int32_t len, int32_t inputType) { int32_t tdProcessRSmaSubmit(SSma *pSma, int64_t version, void *pReq, void *pMsg, int32_t len) {
SSmaEnv *pEnv = SMA_RSMA_ENV(pSma); if (!SMA_RSMA_ENV(pSma)) return TSDB_CODE_SUCCESS;
if (!pEnv) {
// only applicable when rsma env exists
return TDB_CODE_SUCCESS;
}
if ((terrno = atomic_load_32(&SMA_RSMA_STAT(pSma)->execStat))) { if ((terrno = atomic_load_32(&SMA_RSMA_STAT(pSma)->execStat))) {
smaError("vgId:%d, failed to process rsma submit since invalid exec code: %s", SMA_VID(pSma), terrstr()); smaError("vgId:%d, failed to process rsma submit since invalid exec code: %s", SMA_VID(pSma), terrstr());
@ -954,27 +950,25 @@ int32_t tdProcessRSmaSubmit(SSma *pSma, int64_t version, void *pReq, void *pMsg,
STbUidStore uidStore = {0}; STbUidStore uidStore = {0};
if (inputType == STREAM_INPUT__DATA_SUBMIT) { if (tdFetchSubmitReqSuids(pReq, &uidStore) < 0) {
if (tdFetchSubmitReqSuids(pReq, &uidStore) < 0) { smaError("vgId:%d, failed to process rsma submit fetch suid since: %s", SMA_VID(pSma), terrstr());
smaError("vgId:%d, failed to process rsma submit fetch suid since: %s", SMA_VID(pSma), terrstr()); goto _err;
}
if (uidStore.suid != 0) {
if (tdExecuteRSmaAsync(pSma, version, pMsg, len, STREAM_INPUT__DATA_SUBMIT, uidStore.suid) < 0) {
smaError("vgId:%d, failed to process rsma submit exec 1 since: %s", SMA_VID(pSma), terrstr());
goto _err; goto _err;
} }
if (uidStore.suid != 0) { void *pIter = NULL;
if (tdExecuteRSmaAsync(pSma, version, pMsg, len, inputType, uidStore.suid) < 0) { while ((pIter = taosHashIterate(uidStore.uidHash, pIter))) {
smaError("vgId:%d, failed to process rsma submit exec 1 since: %s", SMA_VID(pSma), terrstr()); tb_uid_t *pTbSuid = (tb_uid_t *)taosHashGetKey(pIter, NULL);
if (tdExecuteRSmaAsync(pSma, version, pMsg, len, STREAM_INPUT__DATA_SUBMIT, *pTbSuid) < 0) {
smaError("vgId:%d, failed to process rsma submit exec 2 since: %s", SMA_VID(pSma), terrstr());
taosHashCancelIterate(uidStore.uidHash, pIter);
goto _err; goto _err;
} }
void *pIter = NULL;
while ((pIter = taosHashIterate(uidStore.uidHash, pIter))) {
tb_uid_t *pTbSuid = (tb_uid_t *)taosHashGetKey(pIter, NULL);
if (tdExecuteRSmaAsync(pSma, version, pMsg, len, inputType, *pTbSuid) < 0) {
smaError("vgId:%d, failed to process rsma submit exec 2 since: %s", SMA_VID(pSma), terrstr());
taosHashCancelIterate(uidStore.uidHash, pIter);
goto _err;
}
}
} }
} }
tdUidStoreDestory(&uidStore); tdUidStoreDestory(&uidStore);
@ -984,19 +978,18 @@ _err:
return terrno; return terrno;
} }
int32_t tdProcessRSmaDelete(SSma *pSma, int64_t version, void *pReq, void *pMsg, int32_t len, int32_t inputType) { int32_t tdProcessRSmaDelete(SSma *pSma, int64_t version, void *pReq, void *pMsg, int32_t len) {
SSmaEnv *pEnv = SMA_RSMA_ENV(pSma); if (!SMA_RSMA_ENV(pSma)) return TSDB_CODE_SUCCESS;
if (!pEnv) {
// only applicable when rsma env exists if ((terrno = atomic_load_32(&SMA_RSMA_STAT(pSma)->execStat))) {
return TSDB_CODE_SUCCESS; smaError("vgId:%d, failed to process rsma delete since invalid exec code: %s", SMA_VID(pSma), terrstr());
goto _err;
} }
if (inputType == STREAM_INPUT__REF_DATA_BLOCK) { SDeleteRes *pDelRes = pReq;
SDeleteRes *pReq = pReq; if (tdExecuteRSmaAsync(pSma, version, pMsg, len, STREAM_INPUT__REF_DATA_BLOCK, pDelRes->suid) < 0) {
if (tdExecuteRSmaAsync(pSma, version, pMsg, len, inputType, pReq->suid) < 0) { smaError("vgId:%d, failed to process rsma submit exec 1 since: %s", SMA_VID(pSma), terrstr());
smaError("vgId:%d, failed to process rsma submit exec 1 since: %s", SMA_VID(pSma), terrstr()); goto _err;
goto _err;
}
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
_err: _err:
@ -1381,10 +1374,20 @@ _end:
tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId); tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId);
} }
static void tdFreeRSmaSubmitItems(SArray *pItems) { static void tdFreeRSmaSubmitItems(SArray *pItems, int32_t type) {
for (int32_t i = 0; i < taosArrayGetSize(pItems); ++i) { int32_t arrSize = taosArrayGetSize(pItems);
SPackedData *packData = taosArrayGet(pItems, i); if (type == STREAM_INPUT__MERGED_SUBMIT) {
taosFreeQitem(POINTER_SHIFT(packData->msgStr, -RSMA_SUBMIT_HEAD_LEN)); for (int32_t i = 0; i < arrSize; ++i) {
SPackedData *packData = TARRAY_GET_ELEM(pItems, i);
taosFreeQitem(POINTER_SHIFT(packData->msgStr, -RSMA_SUBMIT_HEAD_LEN));
}
} else if (type == STREAM_INPUT__REF_DATA_BLOCK) {
for (int32_t i = 0; i < arrSize; ++i) {
SPackedData *packData = TARRAY_GET_ELEM(pItems, i);
blockDataDestroy(packData->pDataBlock);
}
} else {
ASSERTS(0, "unknown type:%d", type);
} }
taosArrayClear(pItems); taosArrayClear(pItems);
} }
@ -1448,34 +1451,37 @@ _err:
return TSDB_CODE_FAILED; return TSDB_CODE_FAILED;
} }
#define RSMA_SUBMIT_MSG_TYPE(msg) (*(int8_t *)(msg))
#define RSMA_SUBMIT_MSG_LEN(msg) (*(int32_t *)POINTER_SHIFT((msg), sizeof(int8_t)))
#define RSMA_SUBMIT_MSG_VER(msg) (*(int64_t *)POINTER_SHIFT((msg), sizeof(int8_t) + sizeof(int32_t)))
#define RSMA_SUBMIT_MSG_BODY(msg) (POINTER_SHIFT((msg), sizeof(int8_t) + sizeof(int32_t) + sizeof(int64_t)))
static int32_t tdRSmaBatchExec(SSma *pSma, SRSmaInfo *pInfo, STaosQall *qall, SArray *pSubmitArr, ERsmaExecType type) { static int32_t tdRSmaBatchExec(SSma *pSma, SRSmaInfo *pInfo, STaosQall *qall, SArray *pSubmitArr, ERsmaExecType type) {
void *msg = NULL;
void *msg = NULL; int8_t resume = 0;
int8_t resume = 0; int32_t nSubmit = 0;
int32_t nSubmit = 0; int32_t nDelete = 0;
int32_t nDelete = 0;
SPackedData packData; SPackedData packData;
taosArrayClear(pSubmitArr); taosArrayClear(pSubmitArr);
// the submitReq/deleteReq msg may exsit alternately in the msg queue, consume them sequentially in batch mode
while (1) { while (1) {
taosGetQitem(qall, (void **)&msg); taosGetQitem(qall, (void **)&msg);
if (msg) { if (msg) {
int8_t inputType = *(int8_t *)msg; int8_t inputType = RSMA_SUBMIT_MSG_TYPE(msg);
msg = POINTER_SHIFT(msg, sizeof(int8_t));
if (inputType == STREAM_INPUT__DATA_SUBMIT) { if (inputType == STREAM_INPUT__DATA_SUBMIT) {
if (nDelete > 0) { if (nDelete > 0) {
resume = 1; resume = 1;
break; break;
} }
_resume_submit: _resume_submit:
packData.msgLen = *(int32_t *)msg; packData.msgLen = RSMA_SUBMIT_MSG_LEN(msg);
packData.ver = *(int64_t *)POINTER_SHIFT(msg, sizeof(int32_t)); packData.ver = RSMA_SUBMIT_MSG_VER(msg);
packData.msgStr = POINTER_SHIFT(msg, sizeof(int32_t) + sizeof(int64_t)); packData.msgStr = RSMA_SUBMIT_MSG_BODY(msg);
if (!taosArrayPush(pSubmitArr, &packData)) { if (!taosArrayPush(pSubmitArr, &packData)) {
taosFreeQitem(msg);
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
goto _err; goto _err;
} }
@ -1486,46 +1492,49 @@ static int32_t tdRSmaBatchExec(SSma *pSma, SRSmaInfo *pInfo, STaosQall *qall, SA
break; break;
} }
_resume_delete: _resume_delete:
++nDelete; extractDelDataBlock(RSMA_SUBMIT_MSG_BODY(msg), RSMA_SUBMIT_MSG_LEN(msg), RSMA_SUBMIT_MSG_VER(msg),
#if 0 &packData.pDataBlock, 1);
if (!taosArrayPush(pSubmitArr, &packData)) { if (!taosArrayPush(pSubmitArr, &packData)) {
terrno = TSDB_CODE_OUT_OF_MEMORY; taosFreeQitem(msg);
tdFreeRSmaSubmitItems(pSubmitArr); terrno = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
#endif
} else {
break;
}
}
if (nSubmit > 0) {
int32_t size = taosArrayGetSize(pSubmitArr);
if (size > 0) {
for (int32_t i = 1; i <= TSDB_RETENTION_L2; ++i) {
if (tdExecuteRSmaImpl(pSma, pSubmitArr->pData, size, STREAM_INPUT__MERGED_SUBMIT, pInfo, type, i) < 0) {
goto _err; goto _err;
} }
taosFreeQitem(msg);
++nDelete;
} else {
ASSERTS(0, "unknown msg type:%d", inputType);
break;
} }
tdFreeRSmaSubmitItems(pSubmitArr);
} }
} else if (nDelete > 0) {
}
if (resume == 0) { if (nSubmit > 0 || nDelete > 0) {
goto _rtn; int32_t size = TARRAY_SIZE(pSubmitArr);
} else if (resume == 1) { if (size > 0) {
nSubmit = 0; int32_t inputType = nSubmit > 0 ? STREAM_INPUT__MERGED_SUBMIT : STREAM_INPUT__REF_DATA_BLOCK;
nDelete = 0; for (int32_t i = 1; i <= TSDB_RETENTION_L2; ++i) {
resume = 0; if (tdExecuteRSmaImpl(pSma, pSubmitArr->pData, size, inputType, pInfo, type, i) < 0) {
taosArrayClear(pSubmitArr); goto _err;
goto _resume_submit; }
} else { }
nSubmit = 0; tdFreeRSmaSubmitItems(pSubmitArr, inputType);
nDelete = 0; }
resume = 0; }
taosArrayClear(pSubmitArr);
goto _resume_delete; if (resume == 0) {
goto _rtn;
} else if (resume == 1) {
nSubmit = 0;
nDelete = 0;
resume = 0;
tdFreeRSmaSubmitItems(pSubmitArr, STREAM_INPUT__REF_DATA_BLOCK);
goto _resume_submit;
} else {
nSubmit = 0;
nDelete = 0;
resume = 0;
tdFreeRSmaSubmitItems(pSubmitArr, STREAM_INPUT__MERGED_SUBMIT);
goto _resume_delete;
}
} }
_rtn: _rtn:
@ -1534,7 +1543,7 @@ _err:
atomic_store_32(&SMA_RSMA_STAT(pSma)->execStat, terrno); atomic_store_32(&SMA_RSMA_STAT(pSma)->execStat, terrno);
smaError("vgId:%d, batch exec for suid:%" PRIi64 " execType:%d size:%d failed since %s", SMA_VID(pSma), pInfo->suid, smaError("vgId:%d, batch exec for suid:%" PRIi64 " execType:%d size:%d failed since %s", SMA_VID(pSma), pInfo->suid,
type, (int32_t)taosArrayGetSize(pSubmitArr), terrstr()); type, (int32_t)taosArrayGetSize(pSubmitArr), terrstr());
tdFreeRSmaSubmitItems(pSubmitArr); tdFreeRSmaSubmitItems(pSubmitArr, nSubmit ? STREAM_INPUT__MERGED_SUBMIT : STREAM_INPUT__REF_DATA_BLOCK);
while (1) { while (1) {
void *msg = NULL; void *msg = NULL;
taosGetQitem(qall, (void **)&msg); taosGetQitem(qall, (void **)&msg);

2
source/dnode/vnode/src/tq/tqRead.c
View File

@ -343,7 +343,7 @@ int32_t extractMsgFromWal(SWalReader* pReader, void** pItem, int64_t maxVer, con
void* pBody = POINTER_SHIFT(pCont->body, sizeof(SMsgHead)); void* pBody = POINTER_SHIFT(pCont->body, sizeof(SMsgHead));
int32_t len = pCont->bodyLen - sizeof(SMsgHead); int32_t len = pCont->bodyLen - sizeof(SMsgHead);
code = extractDelDataBlock(pBody, len, ver, (SStreamRefDataBlock**)pItem); code = extractDelDataBlock(pBody, len, ver, (void**)pItem, 0);
if (code == TSDB_CODE_SUCCESS) { if (code == TSDB_CODE_SUCCESS) {
if (*pItem == NULL) { if (*pItem == NULL) {
tqDebug("s-task:%s empty delete msg, discard it, len:%d, ver:%" PRId64, id, len, ver); tqDebug("s-task:%s empty delete msg, discard it, len:%d, ver:%" PRId64, id, len, ver);

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

@ -399,7 +399,7 @@ int32_t tqDoSendDataRsp(const SRpcHandleInfo* pRpcHandleInfo, const SMqDataRsp*
return 0; return 0;
} }
int32_t extractDelDataBlock(const void* pData, int32_t len, int64_t ver, SStreamRefDataBlock** pRefBlock) { int32_t extractDelDataBlock(const void* pData, int32_t len, int64_t ver, void** pRefBlock, int8_t type) {
SDecoder* pCoder = &(SDecoder){0}; SDecoder* pCoder = &(SDecoder){0};
SDeleteRes* pRes = &(SDeleteRes){0}; SDeleteRes* pRes = &(SDeleteRes){0};
@ -442,14 +442,19 @@ int32_t extractDelDataBlock(const void* pData, int32_t len, int64_t ver, SStream
} }
taosArrayDestroy(pRes->uidList); taosArrayDestroy(pRes->uidList);
*pRefBlock = taosAllocateQitem(sizeof(SStreamRefDataBlock), DEF_QITEM, 0); if (type == 0) {
if (*pRefBlock == NULL) { *pRefBlock = taosAllocateQitem(sizeof(SStreamRefDataBlock), DEF_QITEM, 0);
blockDataCleanup(pDelBlock); if (*pRefBlock == NULL) {
taosMemoryFree(pDelBlock); blockDataCleanup(pDelBlock);
return TSDB_CODE_OUT_OF_MEMORY; taosMemoryFree(pDelBlock);
return TSDB_CODE_OUT_OF_MEMORY;
}
((SStreamRefDataBlock*)(*pRefBlock))->type = STREAM_INPUT__REF_DATA_BLOCK;
((SStreamRefDataBlock*)(*pRefBlock))->pBlock = pDelBlock;
} else {
*pRefBlock = pDelBlock;
} }
(*pRefBlock)->type = STREAM_INPUT__REF_DATA_BLOCK;
(*pRefBlock)->pBlock = pDelBlock;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }

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

@ -380,7 +380,7 @@ static int32_t vnodePreProcessDeleteMsg(SVnode *pVnode, SRpcMsg *pMsg) {
SEncoder *pCoder = &(SEncoder){0}; SEncoder *pCoder = &(SEncoder){0};
SDeleteRes res = {0}; SDeleteRes res = {0};
SReadHandle handle = {.vnode = pVnode, .pMsgCb = &pVnode->msgCb}; SReadHandle handle = {.vnode = pVnode, .pMsgCb = &pVnode->msgCb, .skipRollup = 1};
initStorageAPI(&handle.api); initStorageAPI(&handle.api);
code = qWorkerProcessDeleteMsg(&handle, pVnode->pQuery, pMsg, &res); code = qWorkerProcessDeleteMsg(&handle, pVnode->pQuery, pMsg, &res);
@ -1669,7 +1669,7 @@ _exit:
atomic_add_fetch_64(&pVnode->statis.nBatchInsert, 1); atomic_add_fetch_64(&pVnode->statis.nBatchInsert, 1);
if (code == 0) { if (code == 0) {
atomic_add_fetch_64(&pVnode->statis.nBatchInsertSuccess, 1); atomic_add_fetch_64(&pVnode->statis.nBatchInsertSuccess, 1);
code = tdProcessRSmaSubmit(pVnode->pSma, ver, pSubmitReq, pReq, len, STREAM_INPUT__DATA_SUBMIT); code = tdProcessRSmaSubmit(pVnode->pSma, ver, pSubmitReq, pReq, len);
} }
// clear // clear
@ -1947,7 +1947,7 @@ static int32_t vnodeProcessDeleteReq(SVnode *pVnode, int64_t ver, void *pReq, in
if (code) goto _err; if (code) goto _err;
} }
tdProcessRSmaDelete(pVnode->pSma, ver, pRes, pReq, len, STREAM_INPUT__REF_DATA_BLOCK); tdProcessRSmaDelete(pVnode->pSma, ver, pRes, pReq, len);
tDecoderClear(pCoder); tDecoderClear(pCoder);
taosArrayDestroy(pRes->uidList); taosArrayDestroy(pRes->uidList);

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

@ -78,12 +78,11 @@ static int32_t doSetSMABlock(SOperatorInfo* pOperator, void* input, size_t numOf
taosArrayPush(pInfo->pBlockLists, &tmp); taosArrayPush(pInfo->pBlockLists, &tmp);
pInfo->blockType = STREAM_INPUT__CHECKPOINT; pInfo->blockType = STREAM_INPUT__CHECKPOINT;
} else if (type == STREAM_INPUT__REF_DATA_BLOCK) { } else if (type == STREAM_INPUT__REF_DATA_BLOCK) {
for (int32_t i = 0; i < numOfBlocks; ++i) { for (int32_t i = 0; i < numOfBlocks; ++i) {
SSDataBlock* pDataBlock = &((SSDataBlock*)input)[i]; SPackedData* pReq = POINTER_SHIFT(input, i * sizeof(SPackedData));
SPackedData tmp = {.pDataBlock = pDataBlock}; taosArrayPush(pInfo->pBlockLists, pReq);
taosArrayPush(pInfo->pBlockLists, &tmp);
} }
pInfo->blockType = STREAM_INPUT__REF_DATA_BLOCK; pInfo->blockType = STREAM_INPUT__DATA_BLOCK;
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;