From b9471b8115b61224d4f5c832a0216ac76cb84e25 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Mon, 28 Mar 2022 11:13:59 +0800 Subject: [PATCH 1/8] [TD-14222] add test for encode/decode ResultRow --- source/libs/executor/src/executorimpl.c | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 2417bd352a..b739ffb2b9 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -6363,6 +6363,16 @@ static int32_t doOpenAggregateOptr(SOperatorInfo *pOperator) { // the pDataBlock are always the same one, no need to call this again setInputDataBlock(pOperator, pInfo->pCtx, pBlock, order); doAggregateImpl(pOperator, 0, pInfo->pCtx); + + char *result = NULL; + int32_t length = 0; + pOperator->encodeResultRow(pOperator, &result, &length); + SAggSupporter *pSup = &pAggInfo->aggSup; + taosHashClear(pSup->pResultRowHashTable); + pOperator->decodeResultRow(pOperator, result, length); + if(result){ + free(result); + } } finalizeQueryResult(pInfo->pCtx, pOperator->numOfOutput); @@ -6486,7 +6496,7 @@ static bool aggDecodeResultRow(SOperatorInfo* pOperator, char *result, int32_t l initResultRow(resultRow); - pInfo->resultRowInfo.pPosition[pInfo->resultRowInfo.size++] = (SResultRowPosition) {.pageId = resultRow->pageId, .offset = resultRow->offset}; + //pInfo->resultRowInfo.pPosition[pInfo->resultRowInfo.size++] = (SResultRowPosition) {.pageId = resultRow->pageId, .offset = resultRow->offset}; } if (offset != length){ From c941a2c7109ff651877f362b0ae833236565459f Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Tue, 29 Mar 2022 18:16:40 +0800 Subject: [PATCH 2/8] add test for encode/decode ResultRow in group by and interval --- include/libs/nodes/nodes.h | 3 +- source/libs/executor/src/executorimpl.c | 76 ++++++++++++++++++++++--- 2 files changed, 71 insertions(+), 8 deletions(-) diff --git a/include/libs/nodes/nodes.h b/include/libs/nodes/nodes.h index 411d34063c..9698f0e13e 100644 --- a/include/libs/nodes/nodes.h +++ b/include/libs/nodes/nodes.h @@ -141,7 +141,8 @@ typedef enum ENodeType { QUERY_NODE_PHYSICAL_PLAN_DISPATCH, QUERY_NODE_PHYSICAL_PLAN_INSERT, QUERY_NODE_PHYSICAL_SUBPLAN, - QUERY_NODE_PHYSICAL_PLAN + QUERY_NODE_PHYSICAL_PLAN, + QUERY_NODE_PHYSICAL_PLAN_GROUPBY } ENodeType; /** diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 5d6495b52a..5ec5d8cd70 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -6346,7 +6346,7 @@ static int32_t doOpenAggregateOptr(SOperatorInfo *pOperator) { taosHashClear(pSup->pResultRowHashTable); pOperator->decodeResultRow(pOperator, result, length); if(result){ - free(result); + taosMemoryFree(result); } } @@ -6377,8 +6377,27 @@ static SSDataBlock* getAggregateResult(SOperatorInfo *pOperator, bool* newgroup) } static void aggEncodeResultRow(SOperatorInfo* pOperator, char **result, int32_t *length) { - SAggOperatorInfo *pAggInfo = pOperator->info; - SAggSupporter *pSup = &pAggInfo->aggSup; + SAggSupporter *pSup = NULL; + switch(pOperator->operatorType){ + case QUERY_NODE_PHYSICAL_PLAN_AGG:{ + SAggOperatorInfo *pAggInfo = pOperator->info; + pSup = &pAggInfo->aggSup; + break; + } + case QUERY_NODE_PHYSICAL_PLAN_GROUPBY:{ + SGroupbyOperatorInfo *pAggInfo = pOperator->info; + pSup = &pAggInfo->aggSup; + break; + } + case QUERY_NODE_PHYSICAL_PLAN_INTERVAL:{ + STableIntervalOperatorInfo *pAggInfo = pOperator->info; + pSup = &pAggInfo->aggSup; + break; + } + default:{ + qDebug("invalid operatorType: %d", pOperator->operatorType); + } + } int32_t size = taosHashGetSize(pSup->pResultRowHashTable); size_t keyLen = POINTER_BYTES; // estimate the key length @@ -6434,9 +6453,28 @@ static bool aggDecodeResultRow(SOperatorInfo* pOperator, char *result, int32_t l return false; } - SAggOperatorInfo *pAggInfo = pOperator->info; - SAggSupporter *pSup = &pAggInfo->aggSup; - SOptrBasicInfo *pInfo = &pAggInfo->binfo; + SAggSupporter *pSup = NULL; + switch(pOperator->operatorType){ + case QUERY_NODE_PHYSICAL_PLAN_AGG:{ + SAggOperatorInfo *pAggInfo = pOperator->info; + //SOptrBasicInfo *pInfo = &pAggInfo->binfo; + pSup = &pAggInfo->aggSup; + break; + } + case QUERY_NODE_PHYSICAL_PLAN_GROUPBY:{ + SGroupbyOperatorInfo *pAggInfo = pOperator->info; + pSup = &pAggInfo->aggSup; + break; + } + case QUERY_NODE_PHYSICAL_PLAN_INTERVAL:{ + STableIntervalOperatorInfo *pAggInfo = pOperator->info; + pSup = &pAggInfo->aggSup; + break; + } + default:{ + qDebug("invalid operatorType: %d", pOperator->operatorType); + } + } // int32_t size = taosHashGetSize(pSup->pResultRowHashTable); int32_t count = *(int32_t*)(result); @@ -6733,6 +6771,16 @@ static int32_t doOpenIntervalAgg(SOperatorInfo *pOperator) { // the pDataBlock are always the same one, no need to call this again setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, order); hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, 0); + + char *result = NULL; + int32_t length = 0; + pOperator->encodeResultRow(pOperator, &result, &length); + SAggSupporter *pSup = &pInfo->aggSup; + taosHashClear(pSup->pResultRowHashTable); + pOperator->decodeResultRow(pOperator, result, length); + if(result){ + taosMemoryFree(result); + } } closeAllResultRows(&pInfo->binfo.resultRowInfo); @@ -7149,6 +7197,16 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo *pOperator, bool* newgrou setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, order); // setTagValue(pOperator, pRuntimeEnv->current->pTable, pInfo->binfo.pCtx, pOperator->numOfOutput); doHashGroupbyAgg(pOperator, pBlock); + + char *result = NULL; + int32_t length = 0; + pOperator->encodeResultRow(pOperator, &result, &length); + SAggSupporter *pSup = &pInfo->aggSup; + taosHashClear(pSup->pResultRowHashTable); + pOperator->decodeResultRow(pOperator, result, length); + if(result){ + taosMemoryFree(result); + } } pOperator->status = OP_RES_TO_RETURN; @@ -7693,6 +7751,8 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pOperator->_openFn = doOpenIntervalAgg; pOperator->getNextFn = doBuildIntervalResult; pOperator->closeFn = destroyIntervalOperatorInfo; + pOperator->encodeResultRow = aggEncodeResultRow; + pOperator->decodeResultRow = aggDecodeResultRow; code = appendDownstream(pOperator, &downstream, 1); if (code != TSDB_CODE_SUCCESS) { @@ -7903,13 +7963,15 @@ SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pEx pOperator->name = "GroupbyAggOperator"; pOperator->blockingOptr = true; pOperator->status = OP_NOT_OPENED; -// pOperator->operatorType = OP_Groupby; + pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_GROUPBY; pOperator->pExpr = pExprInfo; pOperator->numOfOutput = numOfCols; pOperator->info = pInfo; pOperator->_openFn = operatorDummyOpenFn; pOperator->getNextFn = hashGroupbyAggregate; pOperator->closeFn = destroyGroupbyOperatorInfo; + pOperator->encodeResultRow = aggEncodeResultRow; + pOperator->decodeResultRow = aggDecodeResultRow; code = appendDownstream(pOperator, &downstream, 1); return pOperator; From bceb557809114fd62a896fa7f318a50739b03745 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Wed, 30 Mar 2022 10:05:20 +0800 Subject: [PATCH 3/8] rm useless code --- source/libs/executor/src/dataDispatcher.c | 1 - 1 file changed, 1 deletion(-) diff --git a/source/libs/executor/src/dataDispatcher.c b/source/libs/executor/src/dataDispatcher.c index 34894c235b..e527e3a315 100644 --- a/source/libs/executor/src/dataDispatcher.c +++ b/source/libs/executor/src/dataDispatcher.c @@ -125,7 +125,6 @@ static void toDataCacheEntry(const SDataDispatchHandle* pHandle, const SInputDat pBuf->useSize = sizeof(SRetrieveTableRsp); copyData(pInput, pHandle->pSchema, pEntry->data, pEntry->compressed, &pEntry->dataLen); - pEntry->dataLen = pEntry->dataLen; pBuf->useSize += pEntry->dataLen; } From 9a543fb331eb5babe931c458ca2ef45fe7f47c1b Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Mon, 11 Apr 2022 14:09:47 +0800 Subject: [PATCH 4/8] feat:fix error in result row --- source/libs/executor/src/executorimpl.c | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index c57079cf4b..947e705c39 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -302,10 +302,6 @@ SSDataBlock* createOutputBuf_rv1(SDataBlockDescNode* pNode) { } taosArrayPush(pBlock->pDataBlock, &idata); - - if (IS_VAR_DATA_TYPE(idata.info.type)) { - pBlock->info.hasVarCol = true; - } } return pBlock; @@ -365,7 +361,7 @@ static void prepareResultListBuffer(SResultRowInfo* pResultRowInfo, jmp_buf env) pResultRowInfo->pPosition = taosMemoryRealloc(pResultRowInfo->pPosition, newCapacity * sizeof(SResultRowPosition)); int32_t inc = (int32_t)newCapacity - pResultRowInfo->capacity; - memset(&pResultRowInfo->pPosition[pResultRowInfo->capacity], 0, sizeof(SResultRowPosition)); + memset(&pResultRowInfo->pPosition[pResultRowInfo->capacity], 0, sizeof(SResultRowPosition) * inc); pResultRowInfo->capacity = (int32_t)newCapacity; } @@ -500,7 +496,7 @@ SResultRow* getNewResultRow_rv(SDiskbasedBuf* pResultBuf, int64_t tableGroupId, pData = getBufPage(pResultBuf, getPageId(pi)); pageId = getPageId(pi); - if (pData->num + interBufSize + sizeof(SResultRow) > getBufPageSize(pResultBuf)) { + if (pData->num + interBufSize > getBufPageSize(pResultBuf)) { // release current page first, and prepare the next one releaseBufPageInfo(pResultBuf, pi); @@ -520,7 +516,7 @@ SResultRow* getNewResultRow_rv(SDiskbasedBuf* pResultBuf, int64_t tableGroupId, pResultRow->pageId = pageId; pResultRow->offset = (int32_t)pData->num; - pData->num += interBufSize + sizeof(SResultRow); + pData->num += interBufSize; return pResultRow; } @@ -588,7 +584,7 @@ static SResultRow* doSetResultOutBufByKey_rv(SDiskbasedBuf* pResultBuf, SResultR // add a new result set for a new group SResultRowPosition pos = {.pageId = pResult->pageId, .offset = pResult->offset}; - taosHashPut(pSup->pResultRowHashTable, pSup->keyBuf, GET_RES_WINDOW_KEY_LEN(bytes), &pos, POINTER_BYTES); + taosHashPut(pSup->pResultRowHashTable, pSup->keyBuf, GET_RES_WINDOW_KEY_LEN(bytes), &pos, sizeof(SResultRowPosition)); SResultRowCell cell = {.groupId = tableGroupId, .pos = pos}; taosArrayPush(pSup->pResultRowArrayList, &cell); } else { From 4961b6ab8ee7cb58d51e204f73422c1ccc630b39 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Mon, 11 Apr 2022 17:08:13 +0800 Subject: [PATCH 5/8] feat: optimize encode/decode resultRow --- source/libs/executor/inc/executorimpl.h | 9 +- source/libs/executor/src/executorimpl.c | 134 ++++++++++------------- source/libs/executor/src/groupoperator.c | 6 +- 3 files changed, 68 insertions(+), 81 deletions(-) diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index 9a75830ec4..526f43cb64 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -232,9 +232,11 @@ typedef struct STaskAttr { } STaskAttr; struct SOperatorInfo; +struct SAggSupporter; +struct SOptrBasicInfo; -typedef void (*__optr_encode_fn_t)(struct SOperatorInfo* pOperator, char **result, int32_t *length); -typedef bool (*__optr_decode_fn_t)(struct SOperatorInfo* pOperator, char *result, int32_t length); +typedef void (*__optr_encode_fn_t)(struct SOperatorInfo* pOperator, struct SAggSupporter *pSup, struct SOptrBasicInfo *pInfo, char **result, int32_t *length); +typedef bool (*__optr_decode_fn_t)(struct SOperatorInfo* pOperator, struct SAggSupporter *pSup, struct SOptrBasicInfo *pInfo, char *result, int32_t length); typedef int32_t (*__optr_open_fn_t)(struct SOperatorInfo* pOptr); typedef SSDataBlock* (*__optr_fn_t)(struct SOperatorInfo* pOptr, bool* newgroup); @@ -753,6 +755,9 @@ void setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status); int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHandle* pHandle, uint64_t taskId, EOPTR_EXEC_MODEL model); int32_t getOperatorExplainExecInfo(SOperatorInfo *operatorInfo, SExplainExecInfo **pRes, int32_t *capacity, int32_t *resNum); +bool aggDecodeResultRow(SOperatorInfo* pOperator, SAggSupporter *pSup, SOptrBasicInfo *pInfo, char* result, int32_t length); +void aggEncodeResultRow(SOperatorInfo* pOperator, SAggSupporter *pSup, SOptrBasicInfo *pInfo, char **result, int32_t *length); + #ifdef __cplusplus } #endif diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index d344b93485..8addd53479 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -4974,15 +4974,19 @@ static int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) { setInputDataBlock(pOperator, pInfo->pCtx, pBlock, order); doAggregateImpl(pOperator, 0, pInfo->pCtx); - char *result = NULL; - int32_t length = 0; - pOperator->encodeResultRow(pOperator, &result, &length); - SAggSupporter *pSup = &pAggInfo->aggSup; - taosHashClear(pSup->pResultRowHashTable); - pOperator->decodeResultRow(pOperator, result, length); - if(result){ - taosMemoryFree(result); + if(pOperator->encodeResultRow){ + char *result = NULL; + int32_t length = 0; + SAggSupporter *pSup = &pAggInfo->aggSup; + pOperator->encodeResultRow(pOperator, pSup, pInfo, &result, &length); + taosHashClear(pSup->pResultRowHashTable); + pInfo->resultRowInfo.size = 0; + pOperator->decodeResultRow(pOperator, pSup, pInfo, result, length); + if(result){ + taosMemoryFree(result); + } } + } finalizeQueryResult(pInfo->pCtx, pOperator->numOfOutput); @@ -5011,43 +5015,33 @@ static SSDataBlock* getAggregateResult(SOperatorInfo* pOperator, bool* newgroup) return (blockDataGetNumOfRows(pInfo->pRes) != 0) ? pInfo->pRes : NULL; } -static void aggEncodeResultRow(SOperatorInfo* pOperator, char **result, int32_t *length) { - SAggSupporter *pSup = NULL; - switch(pOperator->operatorType){ - case QUERY_NODE_PHYSICAL_PLAN_AGG:{ - SAggOperatorInfo *pAggInfo = pOperator->info; - pSup = &pAggInfo->aggSup; - break; - } - case QUERY_NODE_PHYSICAL_PLAN_GROUPBY:{ - SGroupbyOperatorInfo *pAggInfo = pOperator->info; - pSup = &pAggInfo->aggSup; - break; - } - case QUERY_NODE_PHYSICAL_PLAN_INTERVAL:{ - STableIntervalOperatorInfo *pAggInfo = pOperator->info; - pSup = &pAggInfo->aggSup; - break; - } - default:{ - qDebug("invalid operatorType: %d", pOperator->operatorType); - } - } - +void aggEncodeResultRow(SOperatorInfo* pOperator, SAggSupporter *pSup, SOptrBasicInfo *pInfo, char **result, int32_t *length) { int32_t size = taosHashGetSize(pSup->pResultRowHashTable); - size_t keyLen = POINTER_BYTES; // estimate the key length + size_t keyLen = sizeof(uint64_t) * 2; // estimate the key length int32_t totalSize = sizeof(int32_t) + size * (sizeof(int32_t) + keyLen + sizeof(int32_t) + pSup->resultRowSize); *result = taosMemoryCalloc(1, totalSize); if (*result == NULL) { - terrno = TSDB_CODE_OUT_OF_MEMORY; - return; + longjmp(pOperator->pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY); } *(int32_t*)(*result) = size; int32_t offset = sizeof(int32_t); + + // prepare memory + SResultRowPosition* pos = &pInfo->resultRowInfo.pPosition[pInfo->resultRowInfo.curPos]; + void* pPage = getBufPage(pSup->pResultBuf, pos->pageId); + SResultRow* pRow = (SResultRow*)((char*)pPage + pos->offset); + setBufPageDirty(pPage, true); + releaseBufPage(pSup->pResultBuf, pPage); + void* pIter = taosHashIterate(pSup->pResultRowHashTable, NULL); while (pIter) { void* key = taosHashGetKey(pIter, &keyLen); - SResultRow** p1 = (SResultRow**)pIter; + SResultRowPosition** p1 = (SResultRowPosition**)pIter; + + pPage = (SFilePage*) getBufPage(pSup->pResultBuf, (*p1)->pageId); + pRow = (SResultRow*)((char*)pPage + (*p1)->offset); + setBufPageDirty(pPage, true); + releaseBufPage(pSup->pResultBuf, pPage); // recalculate the result size int32_t realTotalSize = offset + sizeof(int32_t) + keyLen + sizeof(int32_t) + pSup->resultRowSize; @@ -5057,7 +5051,7 @@ static void aggEncodeResultRow(SOperatorInfo* pOperator, char **result, int32_t terrno = TSDB_CODE_OUT_OF_MEMORY; taosMemoryFree(*result); *result = NULL; - return; + longjmp(pOperator->pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY); } else { *result = tmp; } @@ -5071,7 +5065,7 @@ static void aggEncodeResultRow(SOperatorInfo* pOperator, char **result, int32_t // save value *(int32_t*)(*result + offset) = pSup->resultRowSize; offset += sizeof(int32_t); - memcpy(*result + offset, *p1, pSup->resultRowSize); + memcpy(*result + offset, pRow, pSup->resultRowSize); offset += pSup->resultRowSize; pIter = taosHashIterate(pSup->pResultRowHashTable, pIter); @@ -5083,34 +5077,11 @@ static void aggEncodeResultRow(SOperatorInfo* pOperator, char **result, int32_t return; } -static bool aggDecodeResultRow(SOperatorInfo* pOperator, char* result, int32_t length) { +bool aggDecodeResultRow(SOperatorInfo* pOperator, SAggSupporter *pSup, SOptrBasicInfo *pInfo, char* result, int32_t length) { if (!result || length <= 0) { return false; } - SAggSupporter *pSup = NULL; - switch(pOperator->operatorType){ - case QUERY_NODE_PHYSICAL_PLAN_AGG:{ - SAggOperatorInfo *pAggInfo = pOperator->info; - //SOptrBasicInfo *pInfo = &pAggInfo->binfo; - pSup = &pAggInfo->aggSup; - break; - } - case QUERY_NODE_PHYSICAL_PLAN_GROUPBY:{ - SGroupbyOperatorInfo *pAggInfo = pOperator->info; - pSup = &pAggInfo->aggSup; - break; - } - case QUERY_NODE_PHYSICAL_PLAN_INTERVAL:{ - STableIntervalOperatorInfo *pAggInfo = pOperator->info; - pSup = &pAggInfo->aggSup; - break; - } - default:{ - qDebug("invalid operatorType: %d", pOperator->operatorType); - } - } - // int32_t size = taosHashGetSize(pSup->pResultRowHashTable); int32_t count = *(int32_t*)(result); @@ -5122,17 +5093,16 @@ static bool aggDecodeResultRow(SOperatorInfo* pOperator, char* result, int32_t l uint64_t tableGroupId = *(uint64_t*)(result + offset); SResultRow* resultRow = getNewResultRow_rv(pSup->pResultBuf, tableGroupId, pSup->resultRowSize); if (!resultRow) { - terrno = TSDB_CODE_TSC_INVALID_INPUT; - return false; + longjmp(pOperator->pTaskInfo->env, TSDB_CODE_TSC_INVALID_INPUT); } // add a new result set for a new group - taosHashPut(pSup->pResultRowHashTable, result + offset, keyLen, &resultRow, POINTER_BYTES); + SResultRowPosition pos = {.pageId = resultRow->pageId, .offset = resultRow->offset}; + taosHashPut(pSup->pResultRowHashTable, result + offset, keyLen, &pos, sizeof(SResultRowPosition)); offset += keyLen; int32_t valueLen = *(int32_t*)(result + offset); if (valueLen != pSup->resultRowSize) { - terrno = TSDB_CODE_TSC_INVALID_INPUT; - return false; + longjmp(pOperator->pTaskInfo->env, TSDB_CODE_TSC_INVALID_INPUT); } offset += sizeof(int32_t); int32_t pageId = resultRow->pageId; @@ -5143,12 +5113,13 @@ static bool aggDecodeResultRow(SOperatorInfo* pOperator, char* result, int32_t l offset += valueLen; initResultRow(resultRow); - //pInfo->resultRowInfo.pPosition[pInfo->resultRowInfo.size++] = (SResultRowPosition) {.pageId = resultRow->pageId, .offset = resultRow->offset}; + prepareResultListBuffer(&pInfo->resultRowInfo, pOperator->pTaskInfo->env); + pInfo->resultRowInfo.curPos = pInfo->resultRowInfo.size; + pInfo->resultRowInfo.pPosition[pInfo->resultRowInfo.size++] = (SResultRowPosition) {.pageId = resultRow->pageId, .offset = resultRow->offset}; } if (offset != length) { - terrno = TSDB_CODE_TSC_INVALID_INPUT; - return false; + longjmp(pOperator->pTaskInfo->env, TSDB_CODE_TSC_INVALID_INPUT); } return true; } @@ -5375,16 +5346,21 @@ static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) { setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, order); hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, 0); - char *result = NULL; - int32_t length = 0; - pOperator->encodeResultRow(pOperator, &result, &length); - SAggSupporter *pSup = &pInfo->aggSup; - taosHashClear(pSup->pResultRowHashTable); - pOperator->decodeResultRow(pOperator, result, length); - if(result){ - taosMemoryFree(result); +#if 0 + if(pOperator->encodeResultRow){ + char *result = NULL; + int32_t length = 0; + SAggSupporter *pSup = &pInfo->aggSup; + pOperator->encodeResultRow(pOperator, pSup, &pInfo->binfo, &result, &length); + taosHashClear(pSup->pResultRowHashTable); + pInfo->binfo.resultRowInfo.size = 0; + pOperator->decodeResultRow(pOperator, pSup, &pInfo->binfo, result, length); + if(result){ + taosMemoryFree(result); + } } } +#endif closeAllResultRows(&pInfo->binfo.resultRowInfo); finalizeMultiTupleQueryResult(pInfo->binfo.pCtx, pOperator->numOfOutput, pInfo->aggSup.pResultBuf, @@ -6335,6 +6311,8 @@ SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SExprInf pOperator->info = pInfo; pOperator->getNextFn = doStateWindowAgg; pOperator->closeFn = destroyStateWindowOperatorInfo; + pOperator->encodeResultRow = aggEncodeResultRow; + pOperator->decodeResultRow = aggDecodeResultRow; int32_t code = appendDownstream(pOperator, &downstream, 1); return pOperator; @@ -6374,6 +6352,8 @@ SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo pOperator->info = pInfo; pOperator->getNextFn = doSessionWindowAgg; pOperator->closeFn = destroySWindowOperatorInfo; + pOperator->encodeResultRow = aggEncodeResultRow; + pOperator->decodeResultRow = aggDecodeResultRow; pOperator->pTaskInfo = pTaskInfo; code = appendDownstream(pOperator, &downstream, 1); diff --git a/source/libs/executor/src/groupoperator.c b/source/libs/executor/src/groupoperator.c index 3eb8ff1b72..1de33466e3 100644 --- a/source/libs/executor/src/groupoperator.c +++ b/source/libs/executor/src/groupoperator.c @@ -165,7 +165,7 @@ static int32_t buildGroupKeys(void* pKey, const SArray* pGroupColVals) { // assign the group keys or user input constant values if required static void doAssignGroupKeys(SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t totalRows, int32_t rowIndex) { for (int32_t i = 0; i < numOfOutput; ++i) { - if (pCtx[i].functionId == -1) { + if (pCtx[i].functionId == -1) { // select count(*),key from t group by key. SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(&pCtx[i]); SColumnInfoData* pColInfoData = pCtx[i].input.pData[0]; @@ -337,7 +337,7 @@ SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pEx pOperator->name = "GroupbyAggOperator"; pOperator->blockingOptr = true; pOperator->status = OP_NOT_OPENED; - // pOperator->operatorType = OP_Groupby; + pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_GROUPBY; pOperator->pExpr = pExprInfo; pOperator->numOfOutput = numOfCols; pOperator->info = pInfo; @@ -345,6 +345,8 @@ SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pEx pOperator->_openFn = operatorDummyOpenFn; pOperator->getNextFn = hashGroupbyAggregate; pOperator->closeFn = destroyGroupOperatorInfo; + pOperator->encodeResultRow = aggEncodeResultRow; + pOperator->decodeResultRow = aggDecodeResultRow; code = appendDownstream(pOperator, &downstream, 1); return pOperator; From 78fd2d3a3eabeb6d608c217bb96d04bb18776021 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Mon, 11 Apr 2022 17:30:27 +0800 Subject: [PATCH 6/8] feat: optimize encode/decode resultRow --- source/libs/executor/src/executorimpl.c | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 8addd53479..eff5555dd2 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -4974,6 +4974,7 @@ static int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) { setInputDataBlock(pOperator, pInfo->pCtx, pBlock, order); doAggregateImpl(pOperator, 0, pInfo->pCtx); +#if 0 // test for encode/decode result info if(pOperator->encodeResultRow){ char *result = NULL; int32_t length = 0; @@ -4986,7 +4987,7 @@ static int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) { taosMemoryFree(result); } } - +#endif } finalizeQueryResult(pInfo->pCtx, pOperator->numOfOutput); @@ -5036,10 +5037,10 @@ void aggEncodeResultRow(SOperatorInfo* pOperator, SAggSupporter *pSup, SOptrBasi void* pIter = taosHashIterate(pSup->pResultRowHashTable, NULL); while (pIter) { void* key = taosHashGetKey(pIter, &keyLen); - SResultRowPosition** p1 = (SResultRowPosition**)pIter; + SResultRowPosition* p1 = (SResultRowPosition*)pIter; - pPage = (SFilePage*) getBufPage(pSup->pResultBuf, (*p1)->pageId); - pRow = (SResultRow*)((char*)pPage + (*p1)->offset); + pPage = (SFilePage*) getBufPage(pSup->pResultBuf, p1->pageId); + pRow = (SResultRow*)((char*)pPage + p1->offset); setBufPageDirty(pPage, true); releaseBufPage(pSup->pResultBuf, pPage); @@ -5346,7 +5347,7 @@ static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) { setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, order); hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, 0); -#if 0 +#if 0 // test for encode/decode result info if(pOperator->encodeResultRow){ char *result = NULL; int32_t length = 0; From 45b766cd358f0eaeb7df335e6adbcebf507bbee8 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Mon, 11 Apr 2022 17:35:17 +0800 Subject: [PATCH 7/8] feat: optimize encode/decode resultRow --- include/libs/nodes/nodes.h | 3 +-- source/libs/executor/src/groupoperator.c | 2 +- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/include/libs/nodes/nodes.h b/include/libs/nodes/nodes.h index 68056eff47..c91779ba8b 100644 --- a/include/libs/nodes/nodes.h +++ b/include/libs/nodes/nodes.h @@ -184,8 +184,7 @@ typedef enum ENodeType { QUERY_NODE_PHYSICAL_PLAN_DISPATCH, QUERY_NODE_PHYSICAL_PLAN_INSERT, QUERY_NODE_PHYSICAL_SUBPLAN, - QUERY_NODE_PHYSICAL_PLAN, - QUERY_NODE_PHYSICAL_PLAN_GROUPBY + QUERY_NODE_PHYSICAL_PLAN } ENodeType; /** diff --git a/source/libs/executor/src/groupoperator.c b/source/libs/executor/src/groupoperator.c index 1de33466e3..8d780fc70c 100644 --- a/source/libs/executor/src/groupoperator.c +++ b/source/libs/executor/src/groupoperator.c @@ -337,7 +337,7 @@ SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pEx pOperator->name = "GroupbyAggOperator"; pOperator->blockingOptr = true; pOperator->status = OP_NOT_OPENED; - pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_GROUPBY; + // pOperator->operatorType = OP_Groupby; pOperator->pExpr = pExprInfo; pOperator->numOfOutput = numOfCols; pOperator->info = pInfo; From bc4e73a3ddc3b1f6eff98b3dbe225bdd8a063a0a Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Mon, 11 Apr 2022 17:55:53 +0800 Subject: [PATCH 8/8] feat: optimize encode/decode resultRow --- source/libs/executor/src/executorimpl.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index eff5555dd2..63cd6b77b4 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -5360,8 +5360,8 @@ static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) { taosMemoryFree(result); } } - } #endif + } closeAllResultRows(&pInfo->binfo.resultRowInfo); finalizeMultiTupleQueryResult(pInfo->binfo.pCtx, pOperator->numOfOutput, pInfo->aggSup.pResultBuf,