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

enh: optimize copy group result

This commit is contained in:
dapan1121 2023-05-16 14:12:48 +08:00
parent 198be9a7a6
commit e7df9b8fa0
2 changed files with 22 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
source/dnode/mgmt/mgmt_mnode/src/mmWorker.c
View File

@ -120,6 +120,11 @@ int32_t mmPutMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
} }
int32_t mmPutMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) { int32_t mmPutMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
if (NULL == pMgmt->pMnode) {
const STraceId *trace = &pMsg->info.traceId;
dGError("msg:%p, stop to pre-process in mnode since mnode is NULL, type:%s", pMsg, TMSG_INFO(pMsg->msgType));
return -1;
}
pMsg->info.node = pMgmt->pMnode; pMsg->info.node = pMgmt->pMnode;
if (mndPreProcessQueryMsg(pMsg) != 0) { if (mndPreProcessQueryMsg(pMsg) != 0) {
const STraceId *trace = &pMsg->info.traceId; const STraceId *trace = &pMsg->info.traceId;

31
source/libs/executor/src/executorInt.c
View File

@ -82,7 +82,7 @@ static void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const SC
static int32_t doSetInputDataBlock(SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t order, int32_t scanFlag, static int32_t doSetInputDataBlock(SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t order, int32_t scanFlag,
bool createDummyCol); bool createDummyCol);
static int32_t doCopyToSDataBlock(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock, SExprSupp* pSup, SDiskbasedBuf* pBuf, static int32_t doCopyToSDataBlock(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock, SExprSupp* pSup, SDiskbasedBuf* pBuf,
SGroupResInfo* pGroupResInfo, int32_t threshold); SGroupResInfo* pGroupResInfo, int32_t threshold, bool ignoreGroup);
SResultRow* getNewResultRow(SDiskbasedBuf* pResultBuf, int32_t* currentPageId, int32_t interBufSize) { SResultRow* getNewResultRow(SDiskbasedBuf* pResultBuf, int32_t* currentPageId, int32_t interBufSize) {
SFilePage* pData = NULL; SFilePage* pData = NULL;
@ -776,7 +776,7 @@ int32_t finalizeResultRows(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPos
} }
int32_t doCopyToSDataBlock(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock, SExprSupp* pSup, SDiskbasedBuf* pBuf, int32_t doCopyToSDataBlock(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock, SExprSupp* pSup, SDiskbasedBuf* pBuf,
SGroupResInfo* pGroupResInfo, int32_t threshold) { SGroupResInfo* pGroupResInfo, int32_t threshold, bool ignoreGroup) {
SExprInfo* pExprInfo = pSup->pExprInfo; SExprInfo* pExprInfo = pSup->pExprInfo;
int32_t numOfExprs = pSup->numOfExprs; int32_t numOfExprs = pSup->numOfExprs;
int32_t* rowEntryOffset = pSup->rowEntryInfoOffset; int32_t* rowEntryOffset = pSup->rowEntryInfoOffset;
@ -803,20 +803,23 @@ int32_t doCopyToSDataBlock(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock, SExprS
continue; continue;
} }
if (pBlock->info.id.groupId == 0) { if (!ignoreGroup) {
pBlock->info.id.groupId = pPos->groupId; if (pBlock->info.id.groupId == 0) {
} else { pBlock->info.id.groupId = pPos->groupId;
// current value belongs to different group, it can't be packed into one datablock } else {
if (pBlock->info.id.groupId != pPos->groupId) { // current value belongs to different group, it can't be packed into one datablock
releaseBufPage(pBuf, page); if (pBlock->info.id.groupId != pPos->groupId) {
break; releaseBufPage(pBuf, page);
break;
}
} }
} }
if (pBlock->info.rows + pRow->numOfRows > pBlock->info.capacity) { if (pBlock->info.rows + pRow->numOfRows > pBlock->info.capacity) {
blockDataEnsureCapacity(pBlock, pBlock->info.rows + pRow->numOfRows); uint32_t newSize = pBlock->info.rows + pRow->numOfRows + (numOfRows - i) > 1 ? 1 : 0;
blockDataEnsureCapacity(pBlock, newSize);
qDebug("datablock capacity not sufficient, expand to required:%" PRId64 ", current capacity:%d, %s", qDebug("datablock capacity not sufficient, expand to required:%" PRId64 ", current capacity:%d, %s",
(pRow->numOfRows + pBlock->info.rows), pBlock->info.capacity, GET_TASKID(pTaskInfo)); newSize, pBlock->info.capacity, GET_TASKID(pTaskInfo));
// todo set the pOperator->resultInfo size // todo set the pOperator->resultInfo size
} }
@ -853,7 +856,7 @@ void doBuildStreamResBlock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGr
// clear the existed group id // clear the existed group id
pBlock->info.id.groupId = 0; pBlock->info.id.groupId = 0;
ASSERT(!pbInfo->mergeResultBlock); ASSERT(!pbInfo->mergeResultBlock);
doCopyToSDataBlock(pTaskInfo, pBlock, &pOperator->exprSupp, pBuf, pGroupResInfo, pOperator->resultInfo.threshold); doCopyToSDataBlock(pTaskInfo, pBlock, &pOperator->exprSupp, pBuf, pGroupResInfo, pOperator->resultInfo.threshold, false);
void* tbname = NULL; void* tbname = NULL;
if (streamStateGetParName(pTaskInfo->streamInfo.pState, pBlock->info.id.groupId, &tbname) < 0) { if (streamStateGetParName(pTaskInfo->streamInfo.pState, pBlock->info.id.groupId, &tbname) < 0) {
@ -880,10 +883,10 @@ void doBuildResultDatablock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SG
// clear the existed group id // clear the existed group id
pBlock->info.id.groupId = 0; pBlock->info.id.groupId = 0;
if (!pbInfo->mergeResultBlock) { if (!pbInfo->mergeResultBlock) {
doCopyToSDataBlock(pTaskInfo, pBlock, &pOperator->exprSupp, pBuf, pGroupResInfo, pOperator->resultInfo.threshold); doCopyToSDataBlock(pTaskInfo, pBlock, &pOperator->exprSupp, pBuf, pGroupResInfo, pOperator->resultInfo.threshold, false);
} else { } else {
while (hasRemainResults(pGroupResInfo)) { while (hasRemainResults(pGroupResInfo)) {
doCopyToSDataBlock(pTaskInfo, pBlock, &pOperator->exprSupp, pBuf, pGroupResInfo, pOperator->resultInfo.threshold); doCopyToSDataBlock(pTaskInfo, pBlock, &pOperator->exprSupp, pBuf, pGroupResInfo, pOperator->resultInfo.threshold, true);
if (pBlock->info.rows >= pOperator->resultInfo.threshold) { if (pBlock->info.rows >= pOperator->resultInfo.threshold) {
break; break;
} }