diff --git a/include/libs/sync/sync.h b/include/libs/sync/sync.h index c226d7c8ccbd2cb2e9c248ec895d1220ef40e104..06a58af0e3895b46ef680d85f556a172b68bdbfa 100644 --- a/include/libs/sync/sync.h +++ b/include/libs/sync/sync.h @@ -134,7 +134,7 @@ typedef struct SSyncFSM { int32_t (*FpSnapshotDoRead)(struct SSyncFSM* pFsm, void* pReader, void** ppBuf, int32_t* len); int32_t (*FpSnapshotStartWrite)(struct SSyncFSM* pFsm, void* pWriterParam, void** ppWriter); - int32_t (*FpSnapshotStopWrite)(struct SSyncFSM* pFsm, void* pWriter, bool isApply); + int32_t (*FpSnapshotStopWrite)(struct SSyncFSM* pFsm, void* pWriter, bool isApply, SSnapshot* pSnapshot); int32_t (*FpSnapshotDoWrite)(struct SSyncFSM* pFsm, void* pWriter, void* pBuf, int32_t len); } SSyncFSM; diff --git a/source/common/src/tdatablock.c b/source/common/src/tdatablock.c index 3b7663c4f4a9931cf52c6cf0814aab21a207129a..d9ac84ec06054922552ab17304ba8262874905c5 100644 --- a/source/common/src/tdatablock.c +++ b/source/common/src/tdatablock.c @@ -1479,10 +1479,12 @@ static int32_t colDataMoveVarData(SColumnInfoData* pColInfoData, size_t start, s } beigin++; } + if (dataOffset > 0) { memmove(pColInfoData->pData, pColInfoData->pData + dataOffset, dataLen); - memmove(pColInfoData->varmeta.offset, &pColInfoData->varmeta.offset[start], (end - start) * sizeof(int32_t)); } + + memmove(pColInfoData->varmeta.offset, &pColInfoData->varmeta.offset[start], (end - start) * sizeof(int32_t)); return dataLen; } diff --git a/source/dnode/mnode/impl/src/mndSync.c b/source/dnode/mnode/impl/src/mndSync.c index 2811aeb43aa77f94cffedfc13b342101bee9b2a9..36362bed87aa7f9fccc56ddeff19654f40544827 100644 --- a/source/dnode/mnode/impl/src/mndSync.c +++ b/source/dnode/mnode/impl/src/mndSync.c @@ -143,7 +143,7 @@ int32_t mndSnapshotStartWrite(struct SSyncFSM *pFsm, void *pParam, void **ppWrit return sdbStartWrite(pMnode->pSdb, (SSdbIter **)ppWriter); } -int32_t mndSnapshotStopWrite(struct SSyncFSM *pFsm, void *pWriter, bool isApply) { +int32_t mndSnapshotStopWrite(struct SSyncFSM *pFsm, void *pWriter, bool isApply, SSnapshot *pSnapshot) { mInfo("stop to apply snapshot to sdb, apply:%d", isApply); SMnode *pMnode = pFsm->data; return sdbStopWrite(pMnode->pSdb, pWriter, isApply); diff --git a/source/dnode/vnode/inc/vnode.h b/source/dnode/vnode/inc/vnode.h index 6b5f795eb033934adcd1c6595cbd359368a42127..57141bbbb52001a05a9af11307b883d761918e37 100644 --- a/source/dnode/vnode/inc/vnode.h +++ b/source/dnode/vnode/inc/vnode.h @@ -116,11 +116,11 @@ int32_t metaTbCursorNext(SMTbCursor *pTbCur); // typedef struct STsdb STsdb; typedef struct STsdbReader STsdbReader; -#define BLOCK_LOAD_OFFSET_ORDER 1 +#define BLOCK_LOAD_OFFSET_ORDER 1 #define BLOCK_LOAD_TABLESEQ_ORDER 2 -#define BLOCK_LOAD_EXTERN_ORDER 3 +#define BLOCK_LOAD_EXTERN_ORDER 3 -#define LASTROW_RETRIEVE_TYPE_ALL 0x1 +#define LASTROW_RETRIEVE_TYPE_ALL 0x1 #define LASTROW_RETRIEVE_TYPE_SINGLE 0x2 int32_t tsdbSetTableId(STsdbReader *pReader, int64_t uid); @@ -191,7 +191,7 @@ int32_t vnodeSnapReaderClose(SVSnapReader *pReader); int32_t vnodeSnapRead(SVSnapReader *pReader, uint8_t **ppData, uint32_t *nData); // SVSnapWriter int32_t vnodeSnapWriterOpen(SVnode *pVnode, int64_t sver, int64_t ever, SVSnapWriter **ppWriter); -int32_t vnodeSnapWriterClose(SVSnapWriter *pWriter, int8_t rollback); +int32_t vnodeSnapWriterClose(SVSnapWriter *pWriter, int8_t rollback, SSnapshot *pSnapshot); int32_t vnodeSnapWrite(SVSnapWriter *pWriter, uint8_t *pData, uint32_t nData); // structs @@ -237,8 +237,8 @@ typedef struct { uint64_t groupId; } STableKeyInfo; -#define TABLE_ROLLUP_ON ((int8_t)0x1) -#define TABLE_IS_ROLLUP(FLG) (((FLG) & (TABLE_ROLLUP_ON)) != 0) +#define TABLE_ROLLUP_ON ((int8_t)0x1) +#define TABLE_IS_ROLLUP(FLG) (((FLG) & (TABLE_ROLLUP_ON)) != 0) #define TABLE_SET_ROLLUP(FLG) ((FLG) |= TABLE_ROLLUP_ON) struct SMetaEntry { int64_t version; diff --git a/source/dnode/vnode/src/vnd/vnodeSnapshot.c b/source/dnode/vnode/src/vnd/vnodeSnapshot.c index 420cf3c473579866e6b8b7dd0233be61025480bf..c6b423f712952493abfc60c9e8fa7cc18b4cdf0d 100644 --- a/source/dnode/vnode/src/vnd/vnodeSnapshot.c +++ b/source/dnode/vnode/src/vnd/vnodeSnapshot.c @@ -173,7 +173,7 @@ _err: return code; } -int32_t vnodeSnapWriterClose(SVSnapWriter *pWriter, int8_t rollback) { +int32_t vnodeSnapWriterClose(SVSnapWriter *pWriter, int8_t rollback, SSnapshot *pSnapshot) { int32_t code = 0; SVnode *pVnode = pWriter->pVnode; diff --git a/source/dnode/vnode/src/vnd/vnodeSync.c b/source/dnode/vnode/src/vnd/vnodeSync.c index 81177fd5d85c040d0d1e39ac4d1aa79dd3f517df..712cee9fd05d82a5cf72f0842bf5dc8952590c49 100644 --- a/source/dnode/vnode/src/vnd/vnodeSync.c +++ b/source/dnode/vnode/src/vnd/vnodeSync.c @@ -515,10 +515,10 @@ static int32_t vnodeSnapshotStartWrite(struct SSyncFSM *pFsm, void *pParam, void #endif } -static int32_t vnodeSnapshotStopWrite(struct SSyncFSM *pFsm, void *pWriter, bool isApply) { +static int32_t vnodeSnapshotStopWrite(struct SSyncFSM *pFsm, void *pWriter, bool isApply, SSnapshot *pSnapshot) { #ifdef USE_TSDB_SNAPSHOT SVnode *pVnode = pFsm->data; - int32_t code = vnodeSnapWriterClose(pWriter, !isApply); + int32_t code = vnodeSnapWriterClose(pWriter, !isApply, pSnapshot); return code; #else taosMemoryFree(pWriter); diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index cd50a13eddafae18fd4a407e69ef4b668b39fc28..2ffb1be2603fcf12aca305a95f8661310e50b5ad 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -247,6 +247,16 @@ typedef struct SLoadRemoteDataInfo { uint64_t totalElapsed; // total elapsed time } SLoadRemoteDataInfo; +typedef struct SLimitInfo { + SLimit limit; + SLimit slimit; + uint64_t currentGroupId; + int64_t remainGroupOffset; + int64_t numOfOutputGroups; + int64_t remainOffset; + int64_t numOfOutputRows; +} SLimitInfo; + typedef struct SExchangeInfo { SArray* pSources; SArray* pSourceDataInfo; @@ -257,6 +267,7 @@ typedef struct SExchangeInfo { int32_t current; SLoadRemoteDataInfo loadInfo; uint64_t self; + SLimitInfo limitInfo; } SExchangeInfo; typedef struct SColMatchInfo { @@ -542,15 +553,7 @@ typedef struct SProjectOperatorInfo { SNode* pFilterNode; // filter info, which is push down by optimizer SSDataBlock* existDataBlock; SArray* pPseudoColInfo; - SLimit limit; - SLimit slimit; - - uint64_t groupId; - int64_t curSOffset; - int64_t curGroupOutput; - - int64_t curOffset; - int64_t curOutput; + SLimitInfo limitInfo; } SProjectOperatorInfo; typedef struct SIndefOperatorInfo { @@ -791,6 +794,9 @@ int32_t initAggInfo(SExprSupp *pSup, SAggSupporter* pAggSup, SExprInfo* pExprInf const char* pkey); void initResultSizeInfo(SOperatorInfo* pOperator, int32_t numOfRows); void doBuildResultDatablock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo, SDiskbasedBuf* pBuf); +int32_t handleLimitOffset(SOperatorInfo *pOperator, SLimitInfo* pLimitInfo, SSDataBlock* pBlock, bool holdDataInBuf); +bool hasLimitOffsetInfo(SLimitInfo* pLimitInfo); +void initLimitInfo(const SNode* pLimit, const SNode* pSLimit, SLimitInfo* pLimitInfo); void doApplyFunctions(SExecTaskInfo* taskInfo, SqlFunctionCtx* pCtx, STimeWindow* pWin, SColumnInfoData* pTimeWindowData, int32_t offset, int32_t forwardStep, TSKEY* tsCol, int32_t numOfTotal, int32_t numOfOutput, int32_t order); @@ -837,7 +843,7 @@ SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo* SOperatorInfo* createIndefinitOutputOperatorInfo(SOperatorInfo* downstream, SPhysiNode *pNode, SExecTaskInfo* pTaskInfo); SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhysiNode* pProjPhyNode, SExecTaskInfo* pTaskInfo); -SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* pSortPhyNode, SExecTaskInfo* pTaskInfo); +SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* pSortNode, SExecTaskInfo* pTaskInfo); SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** dowStreams, size_t numStreams, SMergePhysiNode* pMergePhysiNode, SExecTaskInfo* pTaskInfo); SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t numOfDownstream, SExprInfo* pExprInfo, int32_t num, SArray* pSortInfo, SArray* pGroupInfo, SExecTaskInfo* pTaskInfo); SOperatorInfo* createLastrowScanOperator(SLastRowScanPhysiNode* pTableScanNode, SReadHandle* readHandle, SExecTaskInfo* pTaskInfo); diff --git a/source/libs/executor/src/executil.c b/source/libs/executor/src/executil.c index 061ac239050a06cc7021475d53fb66692bd770b0..17484ec84b14b836ca1942afe8f9aca13872d23b 100644 --- a/source/libs/executor/src/executil.c +++ b/source/libs/executor/src/executil.c @@ -898,4 +898,23 @@ STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowI } return w; +} + +bool hasLimitOffsetInfo(SLimitInfo* pLimitInfo) { + return (pLimitInfo->limit.limit != -1 || pLimitInfo->limit.offset != -1 || pLimitInfo->slimit.limit != -1 || + pLimitInfo->slimit.offset != -1); +} + + +static int64_t getLimit(const SNode* pLimit) { return NULL == pLimit ? -1 : ((SLimitNode*)pLimit)->limit; } +static int64_t getOffset(const SNode* pLimit) { return NULL == pLimit ? -1 : ((SLimitNode*)pLimit)->offset; } + +void initLimitInfo(const SNode* pLimit, const SNode* pSLimit, SLimitInfo* pLimitInfo) { + SLimit limit = {.limit = getLimit(pLimit), .offset = getOffset(pLimit)}; + SLimit slimit = {.limit = getLimit(pSLimit), .offset = getOffset(pSLimit)}; + + pLimitInfo->limit = limit; + pLimitInfo->slimit= slimit; + pLimitInfo->remainOffset = limit.offset; + pLimitInfo->remainGroupOffset = slimit.offset; } \ No newline at end of file diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index c3d9f66479f3a277d8a855271b63303f7867d26b..fe90e58527429de01ef0e57e14fe86c38c12f206 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -43,6 +43,11 @@ #define GET_FORWARD_DIRECTION_FACTOR(ord) (((ord) == TSDB_ORDER_ASC) ? QUERY_ASC_FORWARD_STEP : QUERY_DESC_FORWARD_STEP) +enum { + PROJECT_RETRIEVE_CONTINUE = 0x1, + PROJECT_RETRIEVE_DONE = 0x2, +}; + #if 0 static UNUSED_FUNC void *u_malloc (size_t __size) { uint32_t v = taosRand(); @@ -2343,7 +2348,7 @@ static int32_t prepareLoadRemoteData(SOperatorInfo* pOperator) { return TSDB_CODE_SUCCESS; } -static SSDataBlock* doLoadRemoteData(SOperatorInfo* pOperator) { +static SSDataBlock* doLoadRemoteDataImpl(SOperatorInfo* pOperator) { SExchangeInfo* pExchangeInfo = pOperator->info; SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; @@ -2369,6 +2374,44 @@ static SSDataBlock* doLoadRemoteData(SOperatorInfo* pOperator) { } } +static SSDataBlock* doLoadRemoteData(SOperatorInfo* pOperator) { + SExchangeInfo* pExchangeInfo = pOperator->info; + SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; + + if (pOperator->status == OP_EXEC_DONE) { + return NULL; + } + + while(1) { + SSDataBlock* pBlock = doLoadRemoteDataImpl(pOperator); + if (pBlock == NULL) { + return NULL; + } + + ASSERT(pBlock == pExchangeInfo->pResult); + + SLimitInfo* pLimitInfo = &pExchangeInfo->limitInfo; + if (hasLimitOffsetInfo(pLimitInfo)) { + int32_t status = handleLimitOffset(pOperator, pLimitInfo, pExchangeInfo->pResult, false); + if (status == PROJECT_RETRIEVE_CONTINUE) { + continue; + } else if (status == PROJECT_RETRIEVE_DONE) { + size_t rows = pExchangeInfo->pResult->info.rows; + pExchangeInfo->limitInfo.numOfOutputRows += rows; + + if (rows == 0) { + doSetOperatorCompleted(pOperator); + return NULL; + } else { + return pExchangeInfo->pResult; + } + } + } else { + return pExchangeInfo->pResult; + } + } +} + static int32_t initDataSource(int32_t numOfSources, SExchangeInfo* pInfo, const char* id) { pInfo->pSourceDataInfo = taosArrayInit(numOfSources, sizeof(SSourceDataInfo)); if (pInfo->pSourceDataInfo == NULL) { @@ -2408,6 +2451,7 @@ static int32_t initExchangeOperator(SExchangePhysiNode* pExNode, SExchangeInfo* taosArrayPush(pInfo->pSources, pNode); } + initLimitInfo(pExNode->node.pLimit, pExNode->node.pSlimit, &pInfo->limitInfo); pInfo->self = taosAddRef(exchangeObjRefPool, pInfo); return initDataSource(numOfSources, pInfo, id); @@ -3151,68 +3195,60 @@ int32_t aggDecodeResultRow(SOperatorInfo* pOperator, char* result) { return TDB_CODE_SUCCESS; } -enum { - PROJECT_RETRIEVE_CONTINUE = 0x1, - PROJECT_RETRIEVE_DONE = 0x2, -}; - -static int32_t handleLimitOffset(SOperatorInfo* pOperator, SSDataBlock* pBlock) { - SProjectOperatorInfo* pProjectInfo = pOperator->info; - SOptrBasicInfo* pInfo = &pProjectInfo->binfo; - SSDataBlock* pRes = pInfo->pRes; - - if (pProjectInfo->curSOffset > 0) { - if (pProjectInfo->groupId == 0) { // it is the first group - pProjectInfo->groupId = pBlock->info.groupId; - blockDataCleanup(pInfo->pRes); +int32_t handleLimitOffset(SOperatorInfo* pOperator, SLimitInfo* pLimitInfo, SSDataBlock* pBlock, bool holdDataInBuf) { + if (pLimitInfo->remainGroupOffset > 0) { + if (pLimitInfo->currentGroupId == 0) { // it is the first group + pLimitInfo->currentGroupId = pBlock->info.groupId; + blockDataCleanup(pBlock); return PROJECT_RETRIEVE_CONTINUE; - } else if (pProjectInfo->groupId != pBlock->info.groupId) { - pProjectInfo->curSOffset -= 1; + } else if (pLimitInfo->currentGroupId != pBlock->info.groupId) { + // now it is the data from a new group + pLimitInfo->remainGroupOffset -= 1; // ignore data block in current group - if (pProjectInfo->curSOffset > 0) { - blockDataCleanup(pInfo->pRes); + if (pLimitInfo->remainGroupOffset > 0) { + blockDataCleanup(pBlock); return PROJECT_RETRIEVE_CONTINUE; } } // set current group id of the project operator - pProjectInfo->groupId = pBlock->info.groupId; + pLimitInfo->currentGroupId = pBlock->info.groupId; } - if (pProjectInfo->groupId != 0 && pProjectInfo->groupId != pBlock->info.groupId) { - pProjectInfo->curGroupOutput += 1; - if ((pProjectInfo->slimit.limit > 0) && (pProjectInfo->slimit.limit <= pProjectInfo->curGroupOutput)) { + if (pLimitInfo->currentGroupId != 0 && pLimitInfo->currentGroupId != pBlock->info.groupId) { + pLimitInfo->numOfOutputGroups += 1; + if ((pLimitInfo->slimit.limit > 0) && (pLimitInfo->slimit.limit <= pLimitInfo->numOfOutputGroups)) { pOperator->status = OP_EXEC_DONE; - blockDataCleanup(pRes); + blockDataCleanup(pBlock); return PROJECT_RETRIEVE_DONE; } // reset the value for a new group data - pProjectInfo->curOffset = 0; - pProjectInfo->curOutput = 0; + pLimitInfo->numOfOutputRows = 0; + pLimitInfo->remainOffset = pLimitInfo->limit.offset; } // here we reach the start position, according to the limit/offset requirements. // set current group id - pProjectInfo->groupId = pBlock->info.groupId; + pLimitInfo->currentGroupId = pBlock->info.groupId; - if (pProjectInfo->curOffset >= pRes->info.rows) { - pProjectInfo->curOffset -= pRes->info.rows; - blockDataCleanup(pRes); + if (pLimitInfo->remainOffset >= pBlock->info.rows) { + pLimitInfo->remainOffset -= pBlock->info.rows; + blockDataCleanup(pBlock); return PROJECT_RETRIEVE_CONTINUE; - } else if (pProjectInfo->curOffset < pRes->info.rows && pProjectInfo->curOffset > 0) { - blockDataTrimFirstNRows(pRes, pProjectInfo->curOffset); - pProjectInfo->curOffset = 0; + } else if (pLimitInfo->remainOffset < pBlock->info.rows && pLimitInfo->remainOffset > 0) { + blockDataTrimFirstNRows(pBlock, pLimitInfo->remainOffset); + pLimitInfo->remainOffset = 0; } // check for the limitation in each group - if (pProjectInfo->limit.limit >= 0 && pProjectInfo->curOutput + pRes->info.rows >= pProjectInfo->limit.limit) { - int32_t keepRows = (int32_t)(pProjectInfo->limit.limit - pProjectInfo->curOutput); - blockDataKeepFirstNRows(pRes, keepRows); - if (pProjectInfo->slimit.limit > 0 && pProjectInfo->slimit.limit <= pProjectInfo->curGroupOutput) { + if (pLimitInfo->limit.limit >= 0 && pLimitInfo->numOfOutputRows + pBlock->info.rows >= pLimitInfo->limit.limit) { + int32_t keepRows = (int32_t)(pLimitInfo->limit.limit - pLimitInfo->numOfOutputRows); + blockDataKeepFirstNRows(pBlock, keepRows); + if (pLimitInfo->slimit.limit > 0 && pLimitInfo->slimit.limit <= pLimitInfo->numOfOutputGroups) { pOperator->status = OP_EXEC_DONE; } @@ -3222,8 +3258,8 @@ static int32_t handleLimitOffset(SOperatorInfo* pOperator, SSDataBlock* pBlock) // todo optimize performance // If there are slimit/soffset value exists, multi-round result can not be packed into one group, since the // they may not belong to the same group the limit/offset value is not valid in this case. - if (pRes->info.rows >= pOperator->resultInfo.threshold || pProjectInfo->slimit.offset != -1 || - pProjectInfo->slimit.limit != -1) { + if ((!holdDataInBuf) || (pBlock->info.rows >= pOperator->resultInfo.threshold) || pLimitInfo->slimit.offset != -1 || + pLimitInfo->slimit.limit != -1) { return PROJECT_RETRIEVE_DONE; } else { // not full enough, continue to accumulate the output data in the buffer. return PROJECT_RETRIEVE_CONTINUE; @@ -3309,7 +3345,7 @@ static SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) { longjmp(pTaskInfo->env, code); } - int32_t status = handleLimitOffset(pOperator, pBlock); + int32_t status = handleLimitOffset(pOperator, &pProjectInfo->limitInfo, pInfo->pRes, true); // filter shall be applied after apply functions and limit/offset on the result doFilter(pProjectInfo->pFilterNode, pInfo->pRes); @@ -3321,9 +3357,9 @@ static SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) { } } - pProjectInfo->curOutput += pInfo->pRes->info.rows; - size_t rows = pInfo->pRes->info.rows; + pProjectInfo->limitInfo.numOfOutputRows += rows; + pOperator->resultInfo.totalRows += rows; if (pOperator->cost.openCost == 0) { @@ -3767,10 +3803,6 @@ static SArray* setRowTsColumnOutputInfo(SqlFunctionCtx* pCtx, int32_t numOfCols) return pList; } -static int64_t getLimit(SNode* pLimit) { return NULL == pLimit ? -1 : ((SLimitNode*)pLimit)->limit; } - -static int64_t getOffset(SNode* pLimit) { return NULL == pLimit ? -1 : ((SLimitNode*)pLimit)->offset; } - SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhysiNode* pProjPhyNode, SExecTaskInfo* pTaskInfo) { SProjectOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SProjectOperatorInfo)); @@ -3783,13 +3815,8 @@ SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhys SExprInfo* pExprInfo = createExprInfo(pProjPhyNode->pProjections, NULL, &numOfCols); SSDataBlock* pResBlock = createResDataBlock(pProjPhyNode->node.pOutputDataBlockDesc); - SLimit limit = {.limit = getLimit(pProjPhyNode->node.pLimit), .offset = getOffset(pProjPhyNode->node.pLimit)}; - SLimit slimit = {.limit = getLimit(pProjPhyNode->node.pSlimit), .offset = getOffset(pProjPhyNode->node.pSlimit)}; + initLimitInfo(pProjPhyNode->node.pLimit, pProjPhyNode->node.pSlimit, &pInfo->limitInfo); - pInfo->limit = limit; - pInfo->slimit = slimit; - pInfo->curOffset = limit.offset; - pInfo->curSOffset = slimit.offset; pInfo->binfo.pRes = pResBlock; pInfo->pFilterNode = pProjPhyNode->node.pConditions; diff --git a/source/libs/executor/src/groupoperator.c b/source/libs/executor/src/groupoperator.c index c266e8be23df970a0c087852873a8fe0ccfacbc9..2964948e70af588ec6284e31c37278ec802f3981 100644 --- a/source/libs/executor/src/groupoperator.c +++ b/source/libs/executor/src/groupoperator.c @@ -440,19 +440,19 @@ static void doHashPartition(SOperatorInfo* pOperator, SSDataBlock* pBlock) { recordNewGroupKeys(pInfo->pGroupCols, pInfo->pGroupColVals, pBlock, j); int32_t len = buildGroupKeys(pInfo->keyBuf, pInfo->pGroupColVals); - SDataGroupInfo* pGInfo = NULL; - void *pPage = getCurrentDataGroupInfo(pInfo, &pGInfo, len); + SDataGroupInfo* pGroupInfo = NULL; + void *pPage = getCurrentDataGroupInfo(pInfo, &pGroupInfo, len); - pGInfo->numOfRows += 1; - if (pGInfo->groupId == 0) { - pGInfo->groupId = calcGroupId(pInfo->keyBuf, len); + pGroupInfo->numOfRows += 1; + + // group id + if (pGroupInfo->groupId == 0) { + pGroupInfo->groupId = calcGroupId(pInfo->keyBuf, len); } // number of rows int32_t* rows = (int32_t*) pPage; - // group id - size_t numOfCols = pOperator->exprSupp.numOfExprs; for(int32_t i = 0; i < numOfCols; ++i) { SExprInfo* pExpr = &pOperator->exprSupp.pExprInfo[i]; @@ -603,7 +603,13 @@ static void clearPartitionOperator(SPartitionOperatorInfo* pInfo) { static int compareDataGroupInfo(const void* group1, const void* group2) { const SDataGroupInfo* pGroupInfo1 = group1; const SDataGroupInfo* pGroupInfo2 = group2; - return pGroupInfo1->groupId - pGroupInfo2->groupId; + + if (pGroupInfo1->groupId == pGroupInfo2->groupId) { + ASSERT(0); + return 0; + } + + return (pGroupInfo1->groupId < pGroupInfo2->groupId)? -1:1; } static SSDataBlock* buildPartitionResult(SOperatorInfo* pOperator) { diff --git a/source/libs/executor/src/sortoperator.c b/source/libs/executor/src/sortoperator.c index 229d7dd67096b16b50d869d4a304d31f2a2aea68..a3b79d959790c68ce8865496d5998d1e518d0b53 100644 --- a/source/libs/executor/src/sortoperator.c +++ b/source/libs/executor/src/sortoperator.c @@ -22,31 +22,31 @@ static int32_t getExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain static void destroyOrderOperatorInfo(void* param, int32_t numOfOutput); -SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* pSortPhyNode, - SExecTaskInfo* pTaskInfo) { +// todo add limit/offset impl +SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* pSortNode, SExecTaskInfo* pTaskInfo) { SSortOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SSortOperatorInfo)); SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); if (pInfo == NULL || pOperator == NULL /* || rowSize > 100 * 1024 * 1024*/) { goto _error; } - SDataBlockDescNode* pDescNode = pSortPhyNode->node.pOutputDataBlockDesc; + SDataBlockDescNode* pDescNode = pSortNode->node.pOutputDataBlockDesc; int32_t numOfCols = 0; SSDataBlock* pResBlock = createResDataBlock(pDescNode); - SExprInfo* pExprInfo = createExprInfo(pSortPhyNode->pExprs, NULL, &numOfCols); + SExprInfo* pExprInfo = createExprInfo(pSortNode->pExprs, NULL, &numOfCols); int32_t numOfOutputCols = 0; SArray* pColMatchColInfo = - extractColMatchInfo(pSortPhyNode->pTargets, pDescNode, &numOfOutputCols, COL_MATCH_FROM_SLOT_ID); + extractColMatchInfo(pSortNode->pTargets, pDescNode, &numOfOutputCols, COL_MATCH_FROM_SLOT_ID); pOperator->exprSupp.pCtx = createSqlFunctionCtx(pExprInfo, numOfCols, &pOperator->exprSupp.rowEntryInfoOffset); pInfo->binfo.pRes = pResBlock; initResultSizeInfo(pOperator, 1024); - pInfo->pSortInfo = createSortInfo(pSortPhyNode->pSortKeys); - pInfo->pCondition = pSortPhyNode->node.pConditions; + pInfo->pSortInfo = createSortInfo(pSortNode->pSortKeys); + pInfo->pCondition = pSortNode->node.pConditions; pInfo->pColMatchInfo = pColMatchColInfo; pOperator->name = "SortOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_SORT; diff --git a/source/libs/executor/src/tsort.c b/source/libs/executor/src/tsort.c index cd8470883b185cde1393ece684f5ad2a33ba462c..4f525441b96b6169219dc66a29ff58fc377787f2 100644 --- a/source/libs/executor/src/tsort.c +++ b/source/libs/executor/src/tsort.c @@ -91,7 +91,7 @@ SSortHandle* tsortCreateSortHandle(SArray* pSortInfo, int32_t type, int32_t page tsortSetComparFp(pSortHandle, msortComparFn); if (idstr != NULL) { - pSortHandle->idStr = strdup(idstr); + pSortHandle->idStr = strdup(idstr); } return pSortHandle; diff --git a/source/libs/sync/src/syncSnapshot.c b/source/libs/sync/src/syncSnapshot.c index 87cc5685f36b0e39b4ff28df3daa17843205cfdd..924a4df90d5b46544539da7dd7999227e76d8b4a 100644 --- a/source/libs/sync/src/syncSnapshot.c +++ b/source/libs/sync/src/syncSnapshot.c @@ -374,14 +374,14 @@ cJSON *snapshotSender2Json(SSyncSnapshotSender *pSender) { char *snapshotSender2Str(SSyncSnapshotSender *pSender) { cJSON *pJson = snapshotSender2Json(pSender); - char * serialized = cJSON_Print(pJson); + char *serialized = cJSON_Print(pJson); cJSON_Delete(pJson); return serialized; } char *snapshotSender2SimpleStr(SSyncSnapshotSender *pSender, char *event) { int32_t len = 256; - char * s = taosMemoryMalloc(len); + char *s = taosMemoryMalloc(len); SRaftId destId = pSender->pSyncNode->replicasId[pSender->replicaIndex]; char host[64]; @@ -434,8 +434,8 @@ void snapshotReceiverDestroy(SSyncSnapshotReceiver *pReceiver) { if (pReceiver != NULL) { // close writer if (pReceiver->pWriter != NULL) { - int32_t ret = - pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, false); + int32_t ret = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, + false, &(pReceiver->snapshot)); ASSERT(ret == 0); pReceiver->pWriter = NULL; } @@ -483,8 +483,8 @@ static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver, SyncSnapsh static void snapshotReceiverForceStop(SSyncSnapshotReceiver *pReceiver) { // force close, abandon incomplete data if (pReceiver->pWriter != NULL) { - int32_t ret = - pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, false); + int32_t ret = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, false, + &(pReceiver->snapshot)); ASSERT(ret == 0); pReceiver->pWriter = NULL; } @@ -524,8 +524,8 @@ int32_t snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend // FpSnapshotStopWrite should not be called, assert writer == NULL int32_t snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver) { if (pReceiver->pWriter != NULL) { - int32_t ret = - pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, false); + int32_t ret = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, false, + &(pReceiver->snapshot)); ASSERT(ret == 0); pReceiver->pWriter = NULL; } @@ -574,7 +574,8 @@ static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnap } // stop writer, apply data - code = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, true); + code = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, true, + &(pReceiver->snapshot)); if (code != 0) { syncNodeErrorLog(pReceiver->pSyncNode, "snapshot stop writer true error"); ASSERT(0); @@ -646,7 +647,7 @@ cJSON *snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver) { cJSON_AddStringToObject(pFromId, "addr", u64buf); { uint64_t u64 = pReceiver->fromId.addr; - cJSON * pTmp = pFromId; + cJSON *pTmp = pFromId; char host[128] = {0}; uint16_t port; syncUtilU642Addr(u64, host, sizeof(host), &port); @@ -679,14 +680,14 @@ cJSON *snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver) { char *snapshotReceiver2Str(SSyncSnapshotReceiver *pReceiver) { cJSON *pJson = snapshotReceiver2Json(pReceiver); - char * serialized = cJSON_Print(pJson); + char *serialized = cJSON_Print(pJson); cJSON_Delete(pJson); return serialized; } char *snapshotReceiver2SimpleStr(SSyncSnapshotReceiver *pReceiver, char *event) { int32_t len = 256; - char * s = taosMemoryMalloc(len); + char *s = taosMemoryMalloc(len); SRaftId fromId = pReceiver->fromId; char host[128]; diff --git a/source/libs/sync/test/syncConfigChangeSnapshotTest.cpp b/source/libs/sync/test/syncConfigChangeSnapshotTest.cpp index 339ebe90e7e8e84e97ff3be1e64cb62dab1e21d9..8c6c68bbf820bd9bf7cead1cf1e1fe63ed030b40 100644 --- a/source/libs/sync/test/syncConfigChangeSnapshotTest.cpp +++ b/source/libs/sync/test/syncConfigChangeSnapshotTest.cpp @@ -125,7 +125,7 @@ int32_t SnapshotStartWrite(struct SSyncFSM* pFsm, void* pParam, void** ppWriter) return 0; } -int32_t SnapshotStopWrite(struct SSyncFSM* pFsm, void* pWriter, bool isApply) { +int32_t SnapshotStopWrite(struct SSyncFSM* pFsm, void* pWriter, bool isApply, SSnapshot *pSnapshot) { char logBuf[256] = {0}; snprintf(logBuf, sizeof(logBuf), "==callback== ==SnapshotStopWrite== pFsm:%p, pWriter:%p, isApply:%d", pFsm, pWriter, isApply); diff --git a/source/libs/sync/test/syncSnapshotReceiverTest.cpp b/source/libs/sync/test/syncSnapshotReceiverTest.cpp index e5d93ddff421941e71b53caac9319ec61cf23bbf..b744843b1e859bdc34a0bb9706782d9c243c5314 100644 --- a/source/libs/sync/test/syncSnapshotReceiverTest.cpp +++ b/source/libs/sync/test/syncSnapshotReceiverTest.cpp @@ -30,7 +30,7 @@ int32_t SnapshotStopRead(struct SSyncFSM* pFsm, void* pReader) { return 0; } int32_t SnapshotDoRead(struct SSyncFSM* pFsm, void* pReader, void** ppBuf, int32_t* len) { return 0; } int32_t SnapshotStartWrite(struct SSyncFSM* pFsm, void* pParam, void** ppWriter) { return 0; } -int32_t SnapshotStopWrite(struct SSyncFSM* pFsm, void* pWriter, bool isApply) { return 0; } +int32_t SnapshotStopWrite(struct SSyncFSM* pFsm, void* pWriter, bool isApply, SSnapshot *pSnapshot) { return 0; } int32_t SnapshotDoWrite(struct SSyncFSM* pFsm, void* pWriter, void* pBuf, int32_t len) { return 0; } SSyncSnapshotReceiver* createReceiver() { diff --git a/source/libs/sync/test/syncTestTool.cpp b/source/libs/sync/test/syncTestTool.cpp index f35c6f8a2f8ff0514e4e92e3ae9c48155c3fcfdc..b0a561cb89060ccf0bae93b9803bfd64ba827cf8 100644 --- a/source/libs/sync/test/syncTestTool.cpp +++ b/source/libs/sync/test/syncTestTool.cpp @@ -126,7 +126,7 @@ int32_t SnapshotStartWrite(struct SSyncFSM* pFsm, void* pParam, void** ppWriter) return 0; } -int32_t SnapshotStopWrite(struct SSyncFSM* pFsm, void* pWriter, bool isApply) { +int32_t SnapshotStopWrite(struct SSyncFSM* pFsm, void* pWriter, bool isApply, SSnapshot *pSnapshot) { if (isApply) { gSnapshotLastApplyIndex = gFinishLastApplyIndex; gSnapshotLastApplyTerm = gFinishLastApplyTerm;