From 258d50306fa5295fa21bb9434bdcdfcd7f8d0e71 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Tue, 2 Aug 2022 15:43:45 +0800 Subject: [PATCH] refactor: do some internal refactor. --- source/libs/executor/inc/executorimpl.h | 4 +- source/libs/executor/src/cachescanoperator.c | 2 +- source/libs/executor/src/executorimpl.c | 5 +- source/libs/executor/src/timewindowoperator.c | 124 +++++++++--------- 4 files changed, 63 insertions(+), 72 deletions(-) diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index e89459e555..d24156bc60 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -554,8 +554,6 @@ typedef struct SMergeAlignedIntervalAggOperatorInfo { uint64_t groupId; // current groupId int64_t curTs; // current ts SSDataBlock* prefetchedBlock; - bool inputBlocksFinished; - SNode* pCondition; } SMergeAlignedIntervalAggOperatorInfo; @@ -906,7 +904,7 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* SOperatorInfo* createMergeIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId, - SExecTaskInfo* pTaskInfo); + bool mergeResultBlock, SExecTaskInfo* pTaskInfo); SOperatorInfo* createMergeAlignedIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId, diff --git a/source/libs/executor/src/cachescanoperator.c b/source/libs/executor/src/cachescanoperator.c index 56a5e253d8..94e4384b30 100644 --- a/source/libs/executor/src/cachescanoperator.c +++ b/source/libs/executor/src/cachescanoperator.c @@ -50,7 +50,7 @@ SOperatorInfo* createLastrowScanOperator(SLastRowScanPhysiNode* pScanNode, SRead STableListInfo* pTableList = &pTaskInfo->tableqinfoList; - initResultSizeInfo(&pOperator->resultInfo, 1024); + initResultSizeInfo(&pOperator->resultInfo, 4096); blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity); pInfo->pUidList = taosArrayInit(4, sizeof(int64_t)); diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index a7b843d9c2..794e751dd2 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -3481,10 +3481,9 @@ SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo* goto _error; } - int32_t numOfRows = 1024; size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES; - initResultSizeInfo(&pOperator->resultInfo, numOfRows); + initResultSizeInfo(&pOperator->resultInfo, 4096); int32_t code = initAggInfo(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str); if (code != TSDB_CODE_SUCCESS) { goto _error; @@ -4147,7 +4146,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo .precision = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->node.resType.precision}; int32_t tsSlotId = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId; - pOptr = createMergeIntervalOperatorInfo(ops[0], pExprInfo, num, pResBlock, &interval, tsSlotId, pTaskInfo); + pOptr = createMergeIntervalOperatorInfo(ops[0], pExprInfo, num, pResBlock, &interval, tsSlotId, pIntervalPhyNode->window.mergeDataBlock, pTaskInfo); } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL == type) { int32_t children = 0; pOptr = createStreamFinalIntervalOperatorInfo(ops[0], pPhyNode, pTaskInfo, children); diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c index c447d6b43b..812428920a 100644 --- a/source/libs/executor/src/timewindowoperator.c +++ b/source/libs/executor/src/timewindowoperator.c @@ -4622,20 +4622,25 @@ static void doMergeAlignedIntervalAggImpl(SOperatorInfo* pOperatorInfo, SResultR int32_t numOfOutput = pSup->numOfExprs; int64_t* tsCols = extractTsCol(pBlock, iaInfo); uint64_t tableGroupId = pBlock->info.groupId; - TSKEY currTs = getStartTsKey(&pBlock->info.window, tsCols); SResultRow* pResult = NULL; + TSKEY ts = getStartTsKey(&pBlock->info.window, tsCols); + // there is an result exists if (miaInfo->curTs != INT64_MIN) { ASSERT(taosHashGetSize(iaInfo->aggSup.pResultRowHashTable) == 1); - if (currTs != miaInfo->curTs) { + + if (ts != miaInfo->curTs) { outputMergeAlignedIntervalResult(pOperatorInfo, tableGroupId, pResultBlock, miaInfo->curTs); - miaInfo->curTs = INT64_MIN; + miaInfo->curTs = ts; } + } else { + miaInfo->curTs = ts; + ASSERT(taosHashGetSize(iaInfo->aggSup.pResultRowHashTable) == 0); } STimeWindow win = {0}; - win.skey = currTs; + win.skey = miaInfo->curTs; win.ekey = taosTimeAdd(win.skey, iaInfo->interval.interval, iaInfo->interval.intervalUnit, iaInfo->interval.precision) - 1; @@ -4646,12 +4651,11 @@ static void doMergeAlignedIntervalAggImpl(SOperatorInfo* pOperatorInfo, SResultR longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY); } - miaInfo->curTs = win.skey; int32_t currPos = startPos; STimeWindow currWin = win; while (++currPos < pBlock->info.rows) { - if (tsCols[currPos] == currTs) { + if (tsCols[currPos] == miaInfo->curTs) { continue; } @@ -4659,11 +4663,10 @@ static void doMergeAlignedIntervalAggImpl(SOperatorInfo* pOperatorInfo, SResultR doApplyFunctions(pTaskInfo, pSup->pCtx, &currWin, &iaInfo->twAggSup.timeWindowData, startPos, currPos - startPos, tsCols, pBlock->info.rows, numOfOutput, iaInfo->inputOrder); - outputMergeAlignedIntervalResult(pOperatorInfo, tableGroupId, pResultBlock, currTs); - miaInfo->curTs = INT64_MIN; + outputMergeAlignedIntervalResult(pOperatorInfo, tableGroupId, pResultBlock, miaInfo->curTs); + miaInfo->curTs = tsCols[currPos]; - currTs = tsCols[currPos]; - currWin.skey = currTs; + currWin.skey = miaInfo->curTs; currWin.ekey = taosTimeAdd(currWin.skey, iaInfo->interval.interval, iaInfo->interval.intervalUnit, iaInfo->interval.precision) - 1; @@ -4680,14 +4683,6 @@ static void doMergeAlignedIntervalAggImpl(SOperatorInfo* pOperatorInfo, SResultR updateTimeWindowInfo(&iaInfo->twAggSup.timeWindowData, &currWin, true); doApplyFunctions(pTaskInfo, pSup->pCtx, &currWin, &iaInfo->twAggSup.timeWindowData, startPos, currPos - startPos, tsCols, pBlock->info.rows, numOfOutput, iaInfo->inputOrder); - - if (currPos >= pBlock->info.rows) { - // we need to see next block if exists - } else { - ASSERT(0); - outputMergeAlignedIntervalResult(pOperatorInfo, tableGroupId, pResultBlock, currTs); - miaInfo->curTs = INT64_MIN; - } } static SSDataBlock* doMergeAlignedIntervalAgg(SOperatorInfo* pOperator) { @@ -4703,59 +4698,56 @@ static SSDataBlock* doMergeAlignedIntervalAgg(SOperatorInfo* pOperator) { SSDataBlock* pRes = iaInfo->binfo.pRes; blockDataCleanup(pRes); - blockDataEnsureCapacity(pRes, pOperator->resultInfo.capacity); - if (!miaInfo->inputBlocksFinished) { - SOperatorInfo* downstream = pOperator->pDownstream[0]; - int32_t scanFlag = MAIN_SCAN; + SOperatorInfo* downstream = pOperator->pDownstream[0]; + int32_t scanFlag = MAIN_SCAN; - while (1) { - SSDataBlock* pBlock = NULL; - if (miaInfo->prefetchedBlock == NULL) { - pBlock = downstream->fpSet.getNextFn(downstream); - } else { - pBlock = miaInfo->prefetchedBlock; - miaInfo->groupId = pBlock->info.groupId; - miaInfo->prefetchedBlock = NULL; - } + while (1) { + SSDataBlock* pBlock = NULL; + if (miaInfo->prefetchedBlock == NULL) { + pBlock = downstream->fpSet.getNextFn(downstream); + } else { + pBlock = miaInfo->prefetchedBlock; + miaInfo->prefetchedBlock = NULL; - if (pBlock == NULL) { - // close last unfinalized time window - if (miaInfo->curTs != INT64_MIN) { - ASSERT(taosHashGetSize(iaInfo->aggSup.pResultRowHashTable) == 1); - outputMergeAlignedIntervalResult(pOperator, miaInfo->groupId, pRes, miaInfo->curTs); - miaInfo->curTs = INT64_MIN; - } - - doSetOperatorCompleted(pOperator); - miaInfo->inputBlocksFinished = true; - break; - } - - if (!miaInfo->hasGroupId) { - miaInfo->hasGroupId = true; - miaInfo->groupId = pBlock->info.groupId; - } else if (miaInfo->groupId != pBlock->info.groupId) { - // if there are unclosed time window, close it firstly. - ASSERT(miaInfo->curTs != INT64_MIN); - outputMergeAlignedIntervalResult(pOperator, miaInfo->groupId, pRes, miaInfo->curTs); - miaInfo->prefetchedBlock = pBlock; - miaInfo->curTs = INT64_MIN; - break; - } - - getTableScanInfo(pOperator, &iaInfo->inputOrder, &scanFlag); - setInputDataBlock(pOperator, pSup->pCtx, pBlock, iaInfo->inputOrder, scanFlag, true); - doMergeAlignedIntervalAggImpl(pOperator, &iaInfo->binfo.resultRowInfo, pBlock, scanFlag, pRes); - doFilter(miaInfo->pCondition, pRes, NULL); - if (pRes->info.rows >= pOperator->resultInfo.capacity) { - break; - } + miaInfo->groupId = pBlock->info.groupId; } - pRes->info.groupId = miaInfo->groupId; + if (pBlock == NULL) { + // close last unfinalized time window + if (miaInfo->curTs != INT64_MIN) { + ASSERT(taosHashGetSize(iaInfo->aggSup.pResultRowHashTable) == 1); + outputMergeAlignedIntervalResult(pOperator, miaInfo->groupId, pRes, miaInfo->curTs); + miaInfo->curTs = INT64_MIN; + } + + doSetOperatorCompleted(pOperator); + break; + } + + if (!miaInfo->hasGroupId) { + miaInfo->hasGroupId = true; + miaInfo->groupId = pBlock->info.groupId; + } else if (miaInfo->groupId != pBlock->info.groupId) { + // if there are unclosed time window, close it firstly. + ASSERT(miaInfo->curTs != INT64_MIN); + outputMergeAlignedIntervalResult(pOperator, miaInfo->groupId, pRes, miaInfo->curTs); + miaInfo->prefetchedBlock = pBlock; + miaInfo->curTs = INT64_MIN; + break; + } + + getTableScanInfo(pOperator, &iaInfo->inputOrder, &scanFlag); + setInputDataBlock(pOperator, pSup->pCtx, pBlock, iaInfo->inputOrder, scanFlag, true); + doMergeAlignedIntervalAggImpl(pOperator, &iaInfo->binfo.resultRowInfo, pBlock, scanFlag, pRes); + + doFilter(miaInfo->pCondition, pRes, NULL); + if (pRes->info.rows >= pOperator->resultInfo.capacity) { + break; + } } + pRes->info.groupId = miaInfo->groupId; miaInfo->hasGroupId = false; size_t rows = pRes->info.rows; @@ -4809,6 +4801,7 @@ SOperatorInfo* createMergeAlignedIntervalOperatorInfo(SOperatorInfo* downstream, } initResultRowInfo(&iaInfo->binfo.resultRowInfo); + blockDataEnsureCapacity(iaInfo->binfo.pRes, pOperator->resultInfo.capacity); pOperator->name = "TimeMergeAlignedIntervalAggOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL; @@ -5076,7 +5069,7 @@ static SSDataBlock* doMergeIntervalAgg(SOperatorInfo* pOperator) { SOperatorInfo* createMergeIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId, - SExecTaskInfo* pTaskInfo) { + bool mergeBlock, SExecTaskInfo* pTaskInfo) { SMergeIntervalAggOperatorInfo* miaInfo = taosMemoryCalloc(1, sizeof(SMergeIntervalAggOperatorInfo)); SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); if (miaInfo == NULL || pOperator == NULL) { @@ -5090,6 +5083,7 @@ SOperatorInfo* createMergeIntervalOperatorInfo(SOperatorInfo* downstream, SExprI iaInfo->inputOrder = TSDB_ORDER_ASC; iaInfo->interval = *pInterval; iaInfo->execModel = pTaskInfo->execModel; + iaInfo->binfo.mergeResultBlock = mergeBlock; iaInfo->primaryTsIndex = primaryTsSlotId;