From 1c52b59344dfeb8bcaee8d287dc6527e7470aaea Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Fri, 17 Jun 2022 15:48:32 +0800 Subject: [PATCH 1/8] refactor: do some internal refactor. --- source/libs/executor/inc/executorimpl.h | 5 +- source/libs/executor/inc/tsort.h | 2 +- source/libs/executor/src/executorimpl.c | 62 ++++++++++++++----------- source/libs/executor/src/scanoperator.c | 2 +- source/libs/executor/src/sortoperator.c | 4 +- source/libs/executor/src/tsort.c | 2 +- 6 files changed, 41 insertions(+), 36 deletions(-) diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index 034e2893df..6452f7cf7f 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -831,10 +831,9 @@ SOperatorInfo* createDataBlockInfoScanOperator(void* dataReader, SExecTaskInfo* SOperatorInfo* createStreamScanOperatorInfo(void* pDataReader, SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode, SExecTaskInfo* pTaskInfo, STimeWindowAggSupp* pTwSup); +SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode* pPhyFillNode, bool multigroupResult, + SExecTaskInfo* pTaskInfo); -SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols, - SInterval* pInterval, STimeWindow* pWindow, SSDataBlock* pResBlock, int32_t fillType, SNodeListNode* fillVal, - bool multigroupResult, SExecTaskInfo* pTaskInfo); SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols, SSDataBlock* pResBlock, STimeWindowAggSupp *pTwAggSupp, int32_t tsSlotId, SColumn* pStateKeyCol, SExecTaskInfo* pTaskInfo); diff --git a/source/libs/executor/inc/tsort.h b/source/libs/executor/inc/tsort.h index 86ee841cc2..363f379ee4 100644 --- a/source/libs/executor/inc/tsort.h +++ b/source/libs/executor/inc/tsort.h @@ -63,7 +63,7 @@ typedef int32_t (*_sort_merge_compar_fn_t)(const void* p1, const void* p2, void* * @param type * @return */ -SSortHandle* tsortCreateSortHandle(SArray* pOrderInfo, SArray* pIndexMap, int32_t type, int32_t pageSize, int32_t numOfPages, SSDataBlock* pBlock, const char* idstr); +SSortHandle* tsortCreateSortHandle(SArray* pOrderInfo, int32_t type, int32_t pageSize, int32_t numOfPages, SSDataBlock* pBlock, const char* idstr); /** * diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 40b019eb5d..6847605979 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -3073,7 +3073,7 @@ static SSDataBlock* doSortedMerge(SOperatorInfo* pOperator) { } int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize; - pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, NULL, SORT_MULTISOURCE_MERGE, pInfo->bufPageSize, + pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, SORT_MULTISOURCE_MERGE, pInfo->bufPageSize, numOfBufPage, pInfo->binfo.pRes, "GET_TASKID(pTaskInfo)"); tsortSetFetchRawDataFp(pInfo->pSortHandle, loadNextDataBlock, NULL, NULL); @@ -4162,18 +4162,9 @@ static int32_t initFillInfo(SFillOperatorInfo* pInfo, SExprInfo* pExpr, int32_t } } -SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols, - SInterval* pInterval, STimeWindow* pWindow, SSDataBlock* pResBlock, - int32_t fillType, SNodeListNode* pValueNode, bool multigroupResult, - SExecTaskInfo* pTaskInfo) { - SFillOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SFillOperatorInfo)); - SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); - - pInfo->pRes = pResBlock; - pInfo->multigroupResult = multigroupResult; - +static int32_t convertFillType(int32_t mode) { int32_t type = TSDB_FILL_NONE; - switch (fillType) { + switch (mode) { case FILL_MODE_PREV: type = TSDB_FILL_PREV; break; @@ -4196,26 +4187,46 @@ SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExp type = TSDB_FILL_NONE; } + return type; +} + +SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode* pPhyFillNode, bool multigroupResult, + SExecTaskInfo* pTaskInfo) { + SFillOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SFillOperatorInfo)); + SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); + if (pInfo == NULL || pOperator == NULL) { + goto _error; + } + + int32_t num = 0; + SSDataBlock* pResBlock = createResDataBlock(pPhyFillNode->node.pOutputDataBlockDesc); + SExprInfo* pExprInfo = createExprInfo(pPhyFillNode->pTargets, NULL, &num); + SInterval* pInterval = &((SIntervalAggOperatorInfo*)downstream->info)->interval; + int32_t type = convertFillType(pPhyFillNode->mode); + SResultInfo* pResultInfo = &pOperator->resultInfo; initResultSizeInfo(pOperator, 4096); - int32_t code = initFillInfo(pInfo, pExpr, numOfCols, pValueNode, *pWindow, pResultInfo->capacity, pTaskInfo->id.str, - pInterval, type); + int32_t code = initFillInfo(pInfo, pExprInfo, num, (SNodeListNode*)pPhyFillNode->pValues, pPhyFillNode->timeRange, + pResultInfo->capacity, pTaskInfo->id.str, pInterval, type); if (code != TSDB_CODE_SUCCESS) { goto _error; } - pOperator->name = "FillOperator"; - pOperator->blocking = false; - pOperator->status = OP_NOT_OPENED; + pInfo->pRes = pResBlock; + pInfo->multigroupResult = multigroupResult; + pOperator->name = "FillOperator"; + pOperator->blocking = false; + pOperator->status = OP_NOT_OPENED; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_FILL; - pOperator->pExpr = pExpr; - pOperator->numOfExprs = numOfCols; - pOperator->info = pInfo; + pOperator->pExpr = pExprInfo; + pOperator->numOfExprs = num; + pOperator->info = pInfo; + pOperator->pTaskInfo = pTaskInfo; pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doFill, NULL, NULL, destroySFillOperatorInfo, NULL, NULL, NULL); - pOperator->pTaskInfo = pTaskInfo; + code = appendDownstream(pOperator, &downstream, 1); return pOperator; @@ -4590,6 +4601,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo int32_t code = getTableList(pHandle->meta, pScanPhyNode->tableType, pScanPhyNode->uid, pTableListInfo, pScanPhyNode->node.pConditions); if (code != TSDB_CODE_SUCCESS) { + pTaskInfo->code = terrno; return NULL; } @@ -4763,13 +4775,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo SExprInfo* pExprInfo = createExprInfo(pJoinNode->pTargets, NULL, &num); pOptr = createMergeJoinOperatorInfo(ops, size, pExprInfo, num, pResBlock, pJoinNode->pOnConditions, pTaskInfo); } else if (QUERY_NODE_PHYSICAL_PLAN_FILL == type) { - SFillPhysiNode* pFillNode = (SFillPhysiNode*)pPhyNode; - SSDataBlock* pResBlock = createResDataBlock(pPhyNode->pOutputDataBlockDesc); - SExprInfo* pExprInfo = createExprInfo(pFillNode->pTargets, NULL, &num); - - SInterval* pInterval = &((SIntervalAggOperatorInfo*)ops[0]->info)->interval; - pOptr = createFillOperatorInfo(ops[0], pExprInfo, num, pInterval, &pFillNode->timeRange, pResBlock, pFillNode->mode, - (SNodeListNode*)pFillNode->pValues, false, pTaskInfo); + pOptr = createFillOperatorInfo(ops[0], (SFillPhysiNode*)pPhyNode, false, pTaskInfo); } else if (QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC == type) { pOptr = createIndefinitOutputOperatorInfo(ops[0], pPhyNode, pTaskInfo); } else { diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 80276007de..b0325ef8d1 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -2133,7 +2133,7 @@ int32_t doOpenTableMergeScanOperator(SOperatorInfo* pOperator) { int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize; pInfo->pSortHandle = - tsortCreateSortHandle(pInfo->pSortInfo, pInfo->pColMatchInfo, SORT_MULTISOURCE_MERGE, pInfo->bufPageSize, + tsortCreateSortHandle(pInfo->pSortInfo, SORT_MULTISOURCE_MERGE, pInfo->bufPageSize, numOfBufPage, pInfo->pSortInputBlock, pTaskInfo->id.str); tsortSetFetchRawDataFp(pInfo->pSortHandle, getTableDataBlock, NULL, NULL); diff --git a/source/libs/executor/src/sortoperator.c b/source/libs/executor/src/sortoperator.c index 77eeb24210..35e153f8c5 100644 --- a/source/libs/executor/src/sortoperator.c +++ b/source/libs/executor/src/sortoperator.c @@ -154,7 +154,7 @@ int32_t doOpenSortOperator(SOperatorInfo* pOperator) { pInfo->startTs = taosGetTimestampUs(); // pInfo->binfo.pRes is not equalled to the input datablock. - pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, pInfo->pColMatchInfo, SORT_SINGLESOURCE_SORT, -1, -1, + pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, SORT_SINGLESOURCE_SORT, -1, -1, NULL, pTaskInfo->id.str); tsortSetFetchRawDataFp(pInfo->pSortHandle, loadNextDataBlock, applyScalarFunction, pOperator); @@ -248,7 +248,7 @@ int32_t doOpenMultiwaySortMergeOperator(SOperatorInfo* pOperator) { int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize; - pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, pInfo->pColMatchInfo, SORT_MULTISOURCE_MERGE, + pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, SORT_MULTISOURCE_MERGE, pInfo->bufPageSize, numOfBufPage, pInfo->pInputBlock, pTaskInfo->id.str); tsortSetFetchRawDataFp(pInfo->pSortHandle, loadNextDataBlock, NULL, NULL); diff --git a/source/libs/executor/src/tsort.c b/source/libs/executor/src/tsort.c index 00e07c7199..f21cad2dd6 100644 --- a/source/libs/executor/src/tsort.c +++ b/source/libs/executor/src/tsort.c @@ -71,7 +71,7 @@ SSDataBlock* tsortGetSortedDataBlock(const SSortHandle* pSortHandle) { * @param type * @return */ -SSortHandle* tsortCreateSortHandle(SArray* pSortInfo, SArray* pIndexMap, int32_t type, int32_t pageSize, int32_t numOfPages, SSDataBlock* pBlock, const char* idstr) { +SSortHandle* tsortCreateSortHandle(SArray* pSortInfo, int32_t type, int32_t pageSize, int32_t numOfPages, SSDataBlock* pBlock, const char* idstr) { SSortHandle* pSortHandle = taosMemoryCalloc(1, sizeof(SSortHandle)); pSortHandle->type = type; From 78d1f75c9b73d78e2c7ef2e15749088d0ca3ad20 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Fri, 17 Jun 2022 19:01:45 +0800 Subject: [PATCH 2/8] refactor(query): do some internal refactor. --- source/libs/executor/inc/executil.h | 30 +- source/libs/executor/inc/executorimpl.h | 37 +- source/libs/executor/src/executil.c | 602 +++++++++++++-- source/libs/executor/src/executorimpl.c | 728 ++---------------- source/libs/executor/src/groupoperator.c | 24 +- source/libs/executor/src/joinoperator.c | 27 +- source/libs/executor/src/scanoperator.c | 63 +- source/libs/executor/src/sortoperator.c | 35 +- source/libs/executor/src/timewindowoperator.c | 42 +- source/libs/executor/test/sortTests.cpp | 6 +- 10 files changed, 727 insertions(+), 867 deletions(-) diff --git a/source/libs/executor/inc/executil.h b/source/libs/executor/inc/executil.h index a70cff5552..8fd5e7f41e 100644 --- a/source/libs/executor/inc/executil.h +++ b/source/libs/executor/inc/executil.h @@ -15,7 +15,9 @@ #ifndef TDENGINE_QUERYUTIL_H #define TDENGINE_QUERYUTIL_H -#include +#include "function.h" +#include "nodes.h" +#include "plannodes.h" #include "tbuffer.h" #include "tcommon.h" #include "tpagedbuf.h" @@ -77,7 +79,7 @@ typedef struct SResultRowInfo { struct SqlFunctionCtx; size_t getResultRowSize(struct SqlFunctionCtx* pCtx, int32_t numOfOutput); -int32_t initResultRowInfo(SResultRowInfo* pResultRowInfo, int32_t size); +void initResultRowInfo(SResultRowInfo* pResultRowInfo); void cleanupResultRowInfo(SResultRowInfo* pResultRowInfo); void closeAllResultRows(SResultRowInfo* pResultRowInfo); @@ -86,7 +88,7 @@ void initResultRow(SResultRow *pResultRow); void closeResultRow(SResultRow* pResultRow); bool isResultRowClosed(SResultRow* pResultRow); -struct SResultRowEntryInfo* getResultCell(const SResultRow* pRow, int32_t index, const int32_t* offset); +struct SResultRowEntryInfo* getResultEntryInfo(const SResultRow* pRow, int32_t index, const int32_t* offset); static FORCE_INLINE SResultRow *getResultRowByPos(SDiskbasedBuf* pBuf, SResultRowPosition* pos) { SFilePage* bufPage = (SFilePage*) getBufPage(pBuf, pos->pageId); @@ -98,9 +100,27 @@ void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, void initMultiResInfoFromArrayList(SGroupResInfo* pGroupResInfo, SArray* pArrayList); void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo); -bool hashRemainDataInGroupInfo(SGroupResInfo* pGroupResInfo); +bool hasDataInGroupInfo(SGroupResInfo* pGroupResInfo); -bool incNextGroup(SGroupResInfo* pGroupResInfo); int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo); +SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode); + +int32_t getTableList(void* metaHandle, SScanPhysiNode* pScanNode, STableListInfo* pListInfo, SNode* pTagCond); +SArray* createSortInfo(SNodeList* pNodeList); +SArray* extractPartitionColInfo(SNodeList* pNodeList); +SArray* extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNodeList, int32_t* numOfOutputCols, int32_t type); + +SExprInfo* createExprInfo(SNodeList* pNodeList, SNodeList* pGroupKeys, int32_t* numOfExprs); + +SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput, int32_t** rowCellInfoOffset); +void relocateColumnData(SSDataBlock* pBlock, const SArray* pColMatchInfo, SArray* pCols); +void initExecTimeWindowInfo(SColumnInfoData* pColData, STimeWindow* pQueryWindow); + +SInterval extractIntervalInfo(const STableScanPhysiNode* pTableScanNode); +SColumn extractColumnFromColumnNode(SColumnNode* pColNode); + +int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode); +void cleanupQueryTableDataCond(SQueryTableDataCond* pCond); + #endif // TDENGINE_QUERYUTIL_H diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index 6452f7cf7f..f031540996 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -747,43 +747,27 @@ void doBuildResultDatablock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, 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); -int32_t setGroupResultOutputBuf(SOptrBasicInfo* binfo, int32_t numOfCols, char* pData, int16_t type, int16_t bytes, - int32_t groupId, SDiskbasedBuf* pBuf, SExecTaskInfo* pTaskInfo, SAggSupporter* pAggSup); + void doDestroyBasicInfo(SOptrBasicInfo* pInfo, int32_t numOfOutput); -int32_t setDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadInfo, int32_t numOfRows, char* pData, +int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadInfo, int32_t numOfRows, char* pData, int32_t compLen, int32_t numOfOutput, int64_t startTs, uint64_t* total, SArray* pColList); void getAlignQueryTimeWindow(SInterval* pInterval, int32_t precision, int64_t key, STimeWindow* win); int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t *order, int32_t* scanFlag); int32_t getBufferPgSize(int32_t rowSize, uint32_t* defaultPgsz, uint32_t* defaultBufsz); -SArray* extractPartitionColInfo(SNodeList* pNodeList); - void doSetOperatorCompleted(SOperatorInfo* pOperator); void doFilter(const SNode* pFilterNode, SSDataBlock* pBlock); -SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput, int32_t** rowCellInfoOffset); -void relocateColumnData(SSDataBlock* pBlock, const SArray* pColMatchInfo, SArray* pCols); -void initExecTimeWindowInfo(SColumnInfoData* pColData, STimeWindow* pQueryWindow); + void cleanupAggSup(SAggSupporter* pAggSup); void destroyBasicOperatorInfo(void* param, int32_t numOfOutput); void appendOneRowToDataBlock(SSDataBlock* pBlock, STupleHandle* pTupleHandle); void setTbNameColData(void* pMeta, const SSDataBlock* pBlock, SColumnInfoData* pColInfoData, int32_t functionId); -SInterval extractIntervalInfo(const STableScanPhysiNode* pTableScanNode); -SColumn extractColumnFromColumnNode(SColumnNode* pColNode); -SSDataBlock* getSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlock, int32_t capacity, SArray* pColMatchInfo, SSortOperatorInfo* pInfo); SSDataBlock* loadNextDataBlock(void* param); void setResultRowInitCtx(SResultRow* pResult, SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowCellInfoOffset); -SArray* extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNodeList, int32_t* numOfOutputCols, - SExecTaskInfo* pTaskInfo, int32_t type); - -SExprInfo* createExprInfo(SNodeList* pNodeList, SNodeList* pGroupKeys, int32_t* numOfExprs); -SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode); -int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode); -void clearupQueryTableDataCond(SQueryTableDataCond* pCond); - SResultRow* doSetResultOutBufByKey(SDiskbasedBuf* pResultBuf, SResultRowInfo* pResultRowInfo, char* pData, int16_t bytes, bool masterscan, uint64_t groupId, SExecTaskInfo* pTaskInfo, bool isIntervalQuery, SAggSupporter* pSup); @@ -799,9 +783,9 @@ SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo* int32_t numOfScalarExpr, SExecTaskInfo* pTaskInfo); SOperatorInfo* createIndefinitOutputOperatorInfo(SOperatorInfo* downstream, SPhysiNode *pNode, SExecTaskInfo* pTaskInfo); -SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t num, SSDataBlock* pResBlock, SLimit* pLimit, SLimit* pSlimit, SNode* pCondition, SExecTaskInfo* pTaskInfo); -SOperatorInfo *createSortOperatorInfo(SOperatorInfo* downstream, SSDataBlock* pResBlock, SArray* pSortInfo, SExprInfo* pExprInfo, int32_t numOfCols, - SArray* pIndexMap, SExecTaskInfo* pTaskInfo); +SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhysiNode* pProjPhyNode, SExecTaskInfo* pTaskInfo); +SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* pSortPhyNode, SExecTaskInfo* pTaskInfo); + SOperatorInfo* createMultiwaySortMergeOperatorInfo(SOperatorInfo** downStreams, int32_t numStreams, SSDataBlock* pInputBlock, SSDataBlock* pResBlock, SArray* pSortInfo, SArray* pColMatchColInfo, SExecTaskInfo* pTaskInfo); @@ -842,7 +826,8 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock, const SNodeListNode* pValNode, SExecTaskInfo* pTaskInfo); -SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOfDownstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, SNode* pOnCondition, SExecTaskInfo* pTaskInfo); +SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOfDownstream, SJoinPhysiNode* pJoinNode, + SExecTaskInfo* pTaskInfo); SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, int64_t gap, int32_t tsSlotId, STimeWindowAggSupp* pTwAggSupp, SExecTaskInfo* pTaskInfo); @@ -861,8 +846,8 @@ void setInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlo bool isTaskKilled(SExecTaskInfo* pTaskInfo); int32_t checkForQueryBuf(size_t numOfTables); -void setTaskKilled(SExecTaskInfo* pTaskInfo); -void queryCostStatis(SExecTaskInfo* pTaskInfo); +void setTaskKilled(SExecTaskInfo* pTaskInfo); +void queryCostStatis(SExecTaskInfo* pTaskInfo); void doDestroyTask(SExecTaskInfo* pTaskInfo); int32_t getMaximumIdleDurationSec(); @@ -911,8 +896,6 @@ int32_t finalizeResultRowIntoResultDataBlock(SDiskbasedBuf* pBuf, SResultRowPosi SqlFunctionCtx* pCtx, SExprInfo* pExprInfo, int32_t numOfExprs, const int32_t* rowCellOffset, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo); -int32_t getTableList(void* metaHandle, int32_t tableType, uint64_t tableUid, STableListInfo* pListInfo, - SNode* pTagCond); int32_t createMultipleDataReaders(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle, STableListInfo* pTableListInfo, SArray* arrayReader, uint64_t queryId, uint64_t taskId, SNode* pTagCond); diff --git a/source/libs/executor/src/executil.c b/source/libs/executor/src/executil.c index 01ed30c189..c8d9252013 100644 --- a/source/libs/executor/src/executil.c +++ b/source/libs/executor/src/executil.c @@ -14,24 +14,20 @@ */ #include "os.h" -#include "tmsg.h" +#include "index.h" +#include "function.h" +#include "functionMgt.h" +#include "tdatablock.h" #include "thash.h" +#include "tmsg.h" #include "executil.h" #include "executorimpl.h" #include "tcompression.h" -#include "tlosertree.h" -typedef struct SCompSupporter { - STableQueryInfo **pTableQueryInfo; - int32_t *rowIndex; - int32_t order; -} SCompSupporter; - -int32_t initResultRowInfo(SResultRowInfo *pResultRowInfo, int32_t size) { +void initResultRowInfo(SResultRowInfo *pResultRowInfo) { pResultRowInfo->size = 0; pResultRowInfo->cur.pageId = -1; - return TSDB_CODE_SUCCESS; } void cleanupResultRowInfo(SResultRowInfo *pResultRowInfo) { @@ -74,7 +70,7 @@ void closeResultRow(SResultRow* pResultRow) { } // TODO refactor: use macro -SResultRowEntryInfo* getResultCell(const SResultRow* pRow, int32_t index, const int32_t* offset) { +SResultRowEntryInfo* getResultEntryInfo(const SResultRow* pRow, int32_t index, const int32_t* offset) { assert(index >= 0 && offset != NULL); return (SResultRowEntryInfo*)((char*) pRow->pEntryInfo + offset[index]); } @@ -160,7 +156,7 @@ void initMultiResInfoFromArrayList(SGroupResInfo* pGroupResInfo, SArray* pArrayL ASSERT(pGroupResInfo->index <= getNumOfTotalRes(pGroupResInfo)); } -bool hashRemainDataInGroupInfo(SGroupResInfo* pGroupResInfo) { +bool hasDataInGroupInfo(SGroupResInfo* pGroupResInfo) { if (pGroupResInfo->pRows == NULL) { return false; } @@ -177,86 +173,532 @@ int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo) { return (int32_t) taosArrayGetSize(pGroupResInfo->pRows); } -static int32_t tableResultComparFn(const void *pLeft, const void *pRight, void *param) { - int32_t left = *(int32_t *)pLeft; - int32_t right = *(int32_t *)pRight; - - SCompSupporter * supporter = (SCompSupporter *)param; - - int32_t leftPos = supporter->rowIndex[left]; - int32_t rightPos = supporter->rowIndex[right]; - - /* left source is exhausted */ - if (leftPos == -1) { - return 1; +SArray* createSortInfo(SNodeList* pNodeList) { + size_t numOfCols = LIST_LENGTH(pNodeList); + SArray* pList = taosArrayInit(numOfCols, sizeof(SBlockOrderInfo)); + if (pList == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return pList; } - /* right source is exhausted*/ - if (rightPos == -1) { - return -1; + for (int32_t i = 0; i < numOfCols; ++i) { + SOrderByExprNode* pSortKey = (SOrderByExprNode*)nodesListGetNode(pNodeList, i); + SBlockOrderInfo bi = {0}; + bi.order = (pSortKey->order == ORDER_ASC) ? TSDB_ORDER_ASC : TSDB_ORDER_DESC; + bi.nullFirst = (pSortKey->nullOrder == NULL_ORDER_FIRST); + + SColumnNode* pColNode = (SColumnNode*)pSortKey->pExpr; + bi.slotId = pColNode->slotId; + taosArrayPush(pList, &bi); } - ASSERT(0); - STableQueryInfo** pList = supporter->pTableQueryInfo; -// SResultRow* pWindowRes1 = pList[left]->resInfo.pResult[leftPos]; -// SResultRow * pWindowRes1 = getResultRow(&(pList[left]->resInfo), leftPos); -// TSKEY leftTimestamp = pWindowRes1->win.skey; - -// SResultRowInfo *pWindowResInfo2 = &(pList[right]->resInfo); -// SResultRow * pWindowRes2 = getResultRow(pWindowResInfo2, rightPos); -// SResultRow* pWindowRes2 = pList[right]->resInfo.pResult[rightPos]; -// TSKEY rightTimestamp = pWindowRes2->win.skey; - -// if (leftTimestamp == rightTimestamp) { - return 0; -// } - -// if (supporter->order == TSDB_ORDER_ASC) { -// return (leftTimestamp > rightTimestamp)? 1:-1; -// } else { -// return (leftTimestamp < rightTimestamp)? 1:-1; -// } + return pList; } -int32_t tsAscOrder(const void* p1, const void* p2) { - SResultRowCell* pc1 = (SResultRowCell*) p1; - SResultRowCell* pc2 = (SResultRowCell*) p2; +SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode) { + int32_t numOfCols = LIST_LENGTH(pNode->pSlots); - if (pc1->groupId == pc2->groupId) { - ASSERT(0); -// if (pc1->pRow->win.skey == pc2->pRow->win.skey) { -// return 0; -// } else { -// return (pc1->pRow->win.skey < pc2->pRow->win.skey)? -1:1; -// } + SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock)); + pBlock->pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData)); + + pBlock->info.blockId = pNode->dataBlockId; + pBlock->info.rowSize = pNode->totalRowSize; // todo ?? + pBlock->info.type = STREAM_INVALID; + + for (int32_t i = 0; i < numOfCols; ++i) { + SColumnInfoData idata = {{0}}; + SSlotDescNode* pDescNode = (SSlotDescNode*)nodesListGetNode(pNode->pSlots, i); + // if (!pDescNode->output) { // todo disable it temporarily + // continue; + // } + + idata.info.type = pDescNode->dataType.type; + idata.info.bytes = pDescNode->dataType.bytes; + idata.info.scale = pDescNode->dataType.scale; + idata.info.slotId = pDescNode->slotId; + idata.info.precision = pDescNode->dataType.precision; + + if (IS_VAR_DATA_TYPE(idata.info.type)) { + pBlock->info.hasVarCol = true; + } + + taosArrayPush(pBlock->pDataBlock, &idata); + } + + pBlock->info.numOfCols = taosArrayGetSize(pBlock->pDataBlock); + return pBlock; +} + +int32_t getTableList(void* metaHandle, SScanPhysiNode* pScanNode, STableListInfo* pListInfo, SNode* pTagCond) { + int32_t code = TSDB_CODE_SUCCESS; + pListInfo->pTableList = taosArrayInit(8, sizeof(STableKeyInfo)); + + uint64_t tableUid = pScanNode->uid; + + if (pScanNode->tableType == TSDB_SUPER_TABLE) { + if (pTagCond) { + SIndexMetaArg metaArg = { + .metaEx = metaHandle, .idx = tsdbGetIdx(metaHandle), .ivtIdx = tsdbGetIvtIdx(metaHandle), .suid = tableUid}; + + SArray* res = taosArrayInit(8, sizeof(uint64_t)); + code = doFilterTag(pTagCond, &metaArg, res); + if (code == TSDB_CODE_INDEX_REBUILDING) { // todo + // doFilter(); + } else if (code != TSDB_CODE_SUCCESS) { + qError("failed to get tableIds, reason: %s, suid: %" PRIu64 "", tstrerror(code), tableUid); + taosArrayDestroy(res); + terrno = code; + return code; + } else { + qDebug("sucess to get tableIds, size: %d, suid: %" PRIu64 "", (int)taosArrayGetSize(res), tableUid); + } + + for (int i = 0; i < taosArrayGetSize(res); i++) { + STableKeyInfo info = {.lastKey = TSKEY_INITIAL_VAL, .uid = *(uint64_t*)taosArrayGet(res, i)}; + taosArrayPush(pListInfo->pTableList, &info); + } + taosArrayDestroy(res); + } else { + code = tsdbGetAllTableList(metaHandle, tableUid, pListInfo->pTableList); + } + } else { // Create one table group. + STableKeyInfo info = {.lastKey = 0, .uid = tableUid}; + taosArrayPush(pListInfo->pTableList, &info); + } + + return code; +} + +SArray* extractPartitionColInfo(SNodeList* pNodeList) { + if(!pNodeList) { + return NULL; + } + + size_t numOfCols = LIST_LENGTH(pNodeList); + SArray* pList = taosArrayInit(numOfCols, sizeof(SColumn)); + if (pList == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return NULL; + } + + for (int32_t i = 0; i < numOfCols; ++i) { + SColumnNode* pColNode = (SColumnNode*)nodesListGetNode(pNodeList, i); + + // todo extract method + SColumn c = {0}; + c.slotId = pColNode->slotId; + c.colId = pColNode->colId; + c.type = pColNode->node.resType.type; + c.bytes = pColNode->node.resType.bytes; + c.precision = pColNode->node.resType.precision; + c.scale = pColNode->node.resType.scale; + + taosArrayPush(pList, &c); + } + + return pList; +} + + +SArray* extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNodeList, int32_t* numOfOutputCols, + int32_t type) { + size_t numOfCols = LIST_LENGTH(pNodeList); + SArray* pList = taosArrayInit(numOfCols, sizeof(SColMatchInfo)); + if (pList == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return NULL; + } + + for (int32_t i = 0; i < numOfCols; ++i) { + STargetNode* pNode = (STargetNode*)nodesListGetNode(pNodeList, i); + SColumnNode* pColNode = (SColumnNode*)pNode->pExpr; + + SColMatchInfo c = {0}; + c.output = true; + c.colId = pColNode->colId; + c.srcSlotId = pColNode->slotId; + c.matchType = type; + c.targetSlotId = pNode->slotId; + taosArrayPush(pList, &c); + } + + *numOfOutputCols = 0; + int32_t num = LIST_LENGTH(pOutputNodeList->pSlots); + for (int32_t i = 0; i < num; ++i) { + SSlotDescNode* pNode = (SSlotDescNode*)nodesListGetNode(pOutputNodeList->pSlots, i); + + // todo: add reserve flag check + // it is a column reserved for the arithmetic expression calculation + if (pNode->slotId >= numOfCols) { + (*numOfOutputCols) += 1; + continue; + } + + SColMatchInfo* info = taosArrayGet(pList, pNode->slotId); + if (pNode->output) { + (*numOfOutputCols) += 1; + } else { + info->output = false; + } + } + + return pList; +} + +static SResSchema createResSchema(int32_t type, int32_t bytes, int32_t slotId, int32_t scale, int32_t precision, + const char* name) { + SResSchema s = {0}; + s.scale = scale; + s.type = type; + s.bytes = bytes; + s.slotId = slotId; + s.precision = precision; + strncpy(s.name, name, tListLen(s.name)); + + return s; +} + +static SColumn* createColumn(int32_t blockId, int32_t slotId, int32_t colId, SDataType* pType) { + SColumn* pCol = taosMemoryCalloc(1, sizeof(SColumn)); + if (pCol == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return NULL; + } + + pCol->slotId = slotId; + pCol->colId = colId; + pCol->bytes = pType->bytes; + pCol->type = pType->type; + pCol->scale = pType->scale; + pCol->precision = pType->precision; + pCol->dataBlockId = blockId; + + return pCol; +} + +SExprInfo* createExprInfo(SNodeList* pNodeList, SNodeList* pGroupKeys, int32_t* numOfExprs) { + int32_t numOfFuncs = LIST_LENGTH(pNodeList); + int32_t numOfGroupKeys = 0; + if (pGroupKeys != NULL) { + numOfGroupKeys = LIST_LENGTH(pGroupKeys); + } + + *numOfExprs = numOfFuncs + numOfGroupKeys; + SExprInfo* pExprs = taosMemoryCalloc(*numOfExprs, sizeof(SExprInfo)); + + for (int32_t i = 0; i < (*numOfExprs); ++i) { + STargetNode* pTargetNode = NULL; + if (i < numOfFuncs) { + pTargetNode = (STargetNode*)nodesListGetNode(pNodeList, i); + } else { + pTargetNode = (STargetNode*)nodesListGetNode(pGroupKeys, i - numOfFuncs); + } + + SExprInfo* pExp = &pExprs[i]; + + pExp->pExpr = taosMemoryCalloc(1, sizeof(tExprNode)); + pExp->pExpr->_function.num = 1; + pExp->pExpr->_function.functionId = -1; + + int32_t type = nodeType(pTargetNode->pExpr); + // it is a project query, or group by column + if (type == QUERY_NODE_COLUMN) { + pExp->pExpr->nodeType = QUERY_NODE_COLUMN; + SColumnNode* pColNode = (SColumnNode*)pTargetNode->pExpr; + + pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam)); + pExp->base.numOfParams = 1; + + SDataType* pType = &pColNode->node.resType; + pExp->base.resSchema = createResSchema(pType->type, pType->bytes, pTargetNode->slotId, pType->scale, + pType->precision, pColNode->colName); + pExp->base.pParam[0].pCol = createColumn(pColNode->dataBlockId, pColNode->slotId, pColNode->colId, pType); + pExp->base.pParam[0].type = FUNC_PARAM_TYPE_COLUMN; + } else if (type == QUERY_NODE_VALUE) { + pExp->pExpr->nodeType = QUERY_NODE_VALUE; + SValueNode* pValNode = (SValueNode*)pTargetNode->pExpr; + + pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam)); + pExp->base.numOfParams = 1; + + SDataType* pType = &pValNode->node.resType; + pExp->base.resSchema = createResSchema(pType->type, pType->bytes, pTargetNode->slotId, pType->scale, + pType->precision, pValNode->node.aliasName); + pExp->base.pParam[0].type = FUNC_PARAM_TYPE_VALUE; + nodesValueNodeToVariant(pValNode, &pExp->base.pParam[0].param); + } else if (type == QUERY_NODE_FUNCTION) { + pExp->pExpr->nodeType = QUERY_NODE_FUNCTION; + SFunctionNode* pFuncNode = (SFunctionNode*)pTargetNode->pExpr; + + SDataType* pType = &pFuncNode->node.resType; + pExp->base.resSchema = createResSchema(pType->type, pType->bytes, pTargetNode->slotId, pType->scale, + pType->precision, pFuncNode->node.aliasName); + + pExp->pExpr->_function.functionId = pFuncNode->funcId; + pExp->pExpr->_function.pFunctNode = pFuncNode; + + strncpy(pExp->pExpr->_function.functionName, pFuncNode->functionName, + tListLen(pExp->pExpr->_function.functionName)); +#if 1 + // todo refactor: add the parameter for tbname function + if (strcmp(pExp->pExpr->_function.functionName, "tbname") == 0) { + pFuncNode->pParameterList = nodesMakeList(); + ASSERT(LIST_LENGTH(pFuncNode->pParameterList) == 0); + SValueNode* res = (SValueNode*)nodesMakeNode(QUERY_NODE_VALUE); + if (NULL == res) { // todo handle error + } else { + res->node.resType = (SDataType){.bytes = sizeof(int64_t), .type = TSDB_DATA_TYPE_BIGINT}; + nodesListAppend(pFuncNode->pParameterList, (SNode*)res); + } + } +#endif + + int32_t numOfParam = LIST_LENGTH(pFuncNode->pParameterList); + + pExp->base.pParam = taosMemoryCalloc(numOfParam, sizeof(SFunctParam)); + pExp->base.numOfParams = numOfParam; + + for (int32_t j = 0; j < numOfParam; ++j) { + SNode* p1 = nodesListGetNode(pFuncNode->pParameterList, j); + if (p1->type == QUERY_NODE_COLUMN) { + SColumnNode* pcn = (SColumnNode*)p1; + + pExp->base.pParam[j].type = FUNC_PARAM_TYPE_COLUMN; + pExp->base.pParam[j].pCol = createColumn(pcn->dataBlockId, pcn->slotId, pcn->colId, &pcn->node.resType); + } else if (p1->type == QUERY_NODE_VALUE) { + SValueNode* pvn = (SValueNode*)p1; + pExp->base.pParam[j].type = FUNC_PARAM_TYPE_VALUE; + nodesValueNodeToVariant(pvn, &pExp->base.pParam[j].param); + } + } + } else if (type == QUERY_NODE_OPERATOR) { + pExp->pExpr->nodeType = QUERY_NODE_OPERATOR; + SOperatorNode* pNode = (SOperatorNode*)pTargetNode->pExpr; + + pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam)); + pExp->base.numOfParams = 1; + + SDataType* pType = &pNode->node.resType; + pExp->base.resSchema = createResSchema(pType->type, pType->bytes, pTargetNode->slotId, pType->scale, + pType->precision, pNode->node.aliasName); + pExp->pExpr->_optrRoot.pRootNode = pTargetNode->pExpr; + } else { + ASSERT(0); + } + } + + return pExprs; +} + +// set the output buffer for the selectivity + tag query +static int32_t setSelectValueColumnInfo(SqlFunctionCtx* pCtx, int32_t numOfOutput) { + int32_t num = 0; + + SqlFunctionCtx* p = NULL; + SqlFunctionCtx** pValCtx = taosMemoryCalloc(numOfOutput, POINTER_BYTES); + if (pValCtx == NULL) { + return TSDB_CODE_QRY_OUT_OF_MEMORY; + } + + for (int32_t i = 0; i < numOfOutput; ++i) { + if (strcmp(pCtx[i].pExpr->pExpr->_function.functionName, "_select_value") == 0) { + pValCtx[num++] = &pCtx[i]; + } else if (fmIsSelectFunc(pCtx[i].functionId)) { + p = &pCtx[i]; + } + } + + if (p != NULL) { + p->subsidiaries.pCtx = pValCtx; + p->subsidiaries.num = num; } else { - return (pc1->groupId < pc2->groupId)? -1:1; + taosMemoryFreeClear(pValCtx); + } + + return TSDB_CODE_SUCCESS; +} + +SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput, int32_t** rowCellInfoOffset) { + SqlFunctionCtx* pFuncCtx = (SqlFunctionCtx*)taosMemoryCalloc(numOfOutput, sizeof(SqlFunctionCtx)); + if (pFuncCtx == NULL) { + return NULL; + } + + *rowCellInfoOffset = taosMemoryCalloc(numOfOutput, sizeof(int32_t)); + if (*rowCellInfoOffset == 0) { + taosMemoryFreeClear(pFuncCtx); + return NULL; + } + + for (int32_t i = 0; i < numOfOutput; ++i) { + SExprInfo* pExpr = &pExprInfo[i]; + + SExprBasicInfo* pFunct = &pExpr->base; + SqlFunctionCtx* pCtx = &pFuncCtx[i]; + + pCtx->functionId = -1; + pCtx->curBufPage = -1; + pCtx->pExpr = pExpr; + + if (pExpr->pExpr->nodeType == QUERY_NODE_FUNCTION) { + SFuncExecEnv env = {0}; + pCtx->functionId = pExpr->pExpr->_function.pFunctNode->funcId; + + if (fmIsAggFunc(pCtx->functionId) || fmIsIndefiniteRowsFunc(pCtx->functionId)) { + bool isUdaf = fmIsUserDefinedFunc(pCtx->functionId); + if (!isUdaf) { + fmGetFuncExecFuncs(pCtx->functionId, &pCtx->fpSet); + } else { + char* udfName = pExpr->pExpr->_function.pFunctNode->functionName; + strncpy(pCtx->udfName, udfName, strlen(udfName)); + fmGetUdafExecFuncs(pCtx->functionId, &pCtx->fpSet); + } + pCtx->fpSet.getEnv(pExpr->pExpr->_function.pFunctNode, &env); + } else { + fmGetScalarFuncExecFuncs(pCtx->functionId, &pCtx->sfp); + if (pCtx->sfp.getEnv != NULL) { + pCtx->sfp.getEnv(pExpr->pExpr->_function.pFunctNode, &env); + } + } + pCtx->resDataInfo.interBufSize = env.calcMemSize; + } else if (pExpr->pExpr->nodeType == QUERY_NODE_COLUMN || pExpr->pExpr->nodeType == QUERY_NODE_OPERATOR || + pExpr->pExpr->nodeType == QUERY_NODE_VALUE) { + // for simple column, the result buffer needs to hold at least one element. + pCtx->resDataInfo.interBufSize = pFunct->resSchema.bytes; + } + + pCtx->input.numOfInputCols = pFunct->numOfParams; + pCtx->input.pData = taosMemoryCalloc(pFunct->numOfParams, POINTER_BYTES); + pCtx->input.pColumnDataAgg = taosMemoryCalloc(pFunct->numOfParams, POINTER_BYTES); + + pCtx->pTsOutput = NULL; + pCtx->resDataInfo.bytes = pFunct->resSchema.bytes; + pCtx->resDataInfo.type = pFunct->resSchema.type; + pCtx->order = TSDB_ORDER_ASC; + pCtx->start.key = INT64_MIN; + pCtx->end.key = INT64_MIN; + pCtx->numOfParams = pExpr->base.numOfParams; + pCtx->increase = false; + + pCtx->param = pFunct->pParam; + } + + for (int32_t i = 1; i < numOfOutput; ++i) { + (*rowCellInfoOffset)[i] = + (int32_t)((*rowCellInfoOffset)[i - 1] + sizeof(SResultRowEntryInfo) + pFuncCtx[i - 1].resDataInfo.interBufSize); + } + + setSelectValueColumnInfo(pFuncCtx, numOfOutput); + return pFuncCtx; +} + +// NOTE: sources columns are more than the destination SSDatablock columns. +void relocateColumnData(SSDataBlock* pBlock, const SArray* pColMatchInfo, SArray* pCols) { + size_t numOfSrcCols = taosArrayGetSize(pCols); + + int32_t i = 0, j = 0; + while (i < numOfSrcCols && j < taosArrayGetSize(pColMatchInfo)) { + SColumnInfoData* p = taosArrayGet(pCols, i); + SColMatchInfo* pmInfo = taosArrayGet(pColMatchInfo, j); + if (!pmInfo->output) { + j++; + continue; + } + + if (p->info.colId == pmInfo->colId) { + SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, pmInfo->targetSlotId); + colDataAssign(pDst, p, pBlock->info.rows); + i++; + j++; + } else if (p->info.colId < pmInfo->colId) { + i++; + } else { + ASSERT(0); + } } } -int32_t tsDescOrder(const void* p1, const void* p2) { - SResultRowCell* pc1 = (SResultRowCell*) p1; - SResultRowCell* pc2 = (SResultRowCell*) p2; +SInterval extractIntervalInfo(const STableScanPhysiNode* pTableScanNode) { + SInterval interval = { + .interval = pTableScanNode->interval, + .sliding = pTableScanNode->sliding, + .intervalUnit = pTableScanNode->intervalUnit, + .slidingUnit = pTableScanNode->slidingUnit, + .offset = pTableScanNode->offset, + }; - if (pc1->groupId == pc2->groupId) { - ASSERT(0); -// if (pc1->pRow->win.skey == pc2->pRow->win.skey) { -// return 0; -// } else { -// return (pc1->pRow->win.skey < pc2->pRow->win.skey)? 1:-1; -// } - } else { - return (pc1->groupId < pc2->groupId)? -1:1; + return interval; +} + +SColumn extractColumnFromColumnNode(SColumnNode* pColNode) { + SColumn c = {0}; + c.slotId = pColNode->slotId; + c.colId = pColNode->colId; + c.type = pColNode->node.resType.type; + c.bytes = pColNode->node.resType.bytes; + c.scale = pColNode->node.resType.scale; + c.precision = pColNode->node.resType.precision; + return c; +} + +int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode) { + pCond->loadExternalRows = false; + + pCond->order = pTableScanNode->scanSeq[0] > 0 ? TSDB_ORDER_ASC : TSDB_ORDER_DESC; + pCond->numOfCols = LIST_LENGTH(pTableScanNode->scan.pScanCols); + pCond->colList = taosMemoryCalloc(pCond->numOfCols, sizeof(SColumnInfo)); + if (pCond->colList == NULL) { + terrno = TSDB_CODE_QRY_OUT_OF_MEMORY; + return terrno; } + + // pCond->twindow = pTableScanNode->scanRange; + // TODO: get it from stable scan node + pCond->numOfTWindows = 1; + pCond->twindows = taosMemoryCalloc(pCond->numOfTWindows, sizeof(STimeWindow)); + pCond->twindows[0] = pTableScanNode->scanRange; + pCond->suid = pTableScanNode->scan.suid; + +#if 1 + // todo work around a problem, remove it later + for (int32_t i = 0; i < pCond->numOfTWindows; ++i) { + if ((pCond->order == TSDB_ORDER_ASC && pCond->twindows[i].skey > pCond->twindows[i].ekey) || + (pCond->order == TSDB_ORDER_DESC && pCond->twindows[i].skey < pCond->twindows[i].ekey)) { + TSWAP(pCond->twindows[i].skey, pCond->twindows[i].ekey); + } + } +#endif + + for (int32_t i = 0; i < pCond->numOfTWindows; ++i) { + if ((pCond->order == TSDB_ORDER_ASC && pCond->twindows[i].skey > pCond->twindows[i].ekey) || + (pCond->order == TSDB_ORDER_DESC && pCond->twindows[i].skey < pCond->twindows[i].ekey)) { + TSWAP(pCond->twindows[i].skey, pCond->twindows[i].ekey); + } + } + taosqsort(pCond->twindows, pCond->numOfTWindows, sizeof(STimeWindow), pCond, compareTimeWindow); + + pCond->type = BLOCK_LOAD_OFFSET_SEQ_ORDER; + // pCond->type = pTableScanNode->scanFlag; + + int32_t j = 0; + for (int32_t i = 0; i < pCond->numOfCols; ++i) { + STargetNode* pNode = (STargetNode*)nodesListGetNode(pTableScanNode->scan.pScanCols, i); + SColumnNode* pColNode = (SColumnNode*)pNode->pExpr; + if (pColNode->colType == COLUMN_TYPE_TAG) { + continue; + } + + pCond->colList[j].type = pColNode->node.resType.type; + pCond->colList[j].bytes = pColNode->node.resType.bytes; + pCond->colList[j].colId = pColNode->colId; + j += 1; + } + + pCond->numOfCols = j; + return TSDB_CODE_SUCCESS; } -void orderTheResultRows(STaskRuntimeEnv* pRuntimeEnv) { - __compar_fn_t fn = NULL; -// if (pRuntimeEnv->pQueryAttr->order.order == TSDB_ORDER_ASC) { -// fn = tsAscOrder; -// } else { -// fn = tsDescOrder; -// } - - taosArraySort(pRuntimeEnv->pResultRowArrayList, fn); -} +void cleanupQueryTableDataCond(SQueryTableDataCond* pCond) { + taosMemoryFree(pCond->twindows); + taosMemoryFree(pCond->colList); +} \ No newline at end of file diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 6847605979..e41a6a2fdb 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -174,40 +174,6 @@ static int compareRowData(const void* a, const void* b, const void* userData) { } // setup the output buffer for each operator -SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode) { - int32_t numOfCols = LIST_LENGTH(pNode->pSlots); - - SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock)); - pBlock->pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData)); - - pBlock->info.blockId = pNode->dataBlockId; - pBlock->info.rowSize = pNode->totalRowSize; // todo ?? - pBlock->info.type = STREAM_INVALID; - - for (int32_t i = 0; i < numOfCols; ++i) { - SColumnInfoData idata = {{0}}; - SSlotDescNode* pDescNode = (SSlotDescNode*)nodesListGetNode(pNode->pSlots, i); - // if (!pDescNode->output) { // todo disable it temporarily - // continue; - // } - - idata.info.type = pDescNode->dataType.type; - idata.info.bytes = pDescNode->dataType.bytes; - idata.info.scale = pDescNode->dataType.scale; - idata.info.slotId = pDescNode->slotId; - idata.info.precision = pDescNode->dataType.precision; - - if (IS_VAR_DATA_TYPE(idata.info.type)) { - pBlock->info.hasVarCol = true; - } - - taosArrayPush(pBlock->pDataBlock, &idata); - } - - pBlock->info.numOfCols = taosArrayGetSize(pBlock->pDataBlock); - return pBlock; -} - static bool hasNull(SColumn* pColumn, SColumnDataAgg* pStatis) { if (TSDB_COL_IS_TAG(pColumn->flag) || TSDB_COL_IS_UD_COL(pColumn->flag) || pColumn->colId == PRIMARYKEY_TIMESTAMP_COL_ID) { @@ -802,20 +768,6 @@ static void setResultRowKey(SResultRow* pResultRow, char* pData, int16_t type) { } } -int32_t setGroupResultOutputBuf(SOptrBasicInfo* binfo, int32_t numOfCols, char* pData, int16_t type, int16_t bytes, - int32_t groupId, SDiskbasedBuf* pBuf, SExecTaskInfo* pTaskInfo, - SAggSupporter* pAggSup) { - SResultRowInfo* pResultRowInfo = &binfo->resultRowInfo; - SqlFunctionCtx* pCtx = binfo->pCtx; - - SResultRow* pResultRow = - doSetResultOutBufByKey(pBuf, pResultRowInfo, (char*)pData, bytes, true, groupId, pTaskInfo, false, pAggSup); - assert(pResultRow != NULL); - - setResultRowInitCtx(pResultRow, pCtx, numOfCols, binfo->rowCellInfoOffset); - return TSDB_CODE_SUCCESS; -} - bool functionNeedToExecute(SqlFunctionCtx* pCtx) { struct SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx); @@ -927,137 +879,6 @@ void setBlockStatisInfo(SqlFunctionCtx* pCtx, SExprInfo* pExprInfo, SSDataBlock* // } } -// set the output buffer for the selectivity + tag query -static int32_t setSelectValueColumnInfo(SqlFunctionCtx* pCtx, int32_t numOfOutput) { - int32_t num = 0; - - SqlFunctionCtx* p = NULL; - SqlFunctionCtx** pValCtx = taosMemoryCalloc(numOfOutput, POINTER_BYTES); - if (pValCtx == NULL) { - return TSDB_CODE_QRY_OUT_OF_MEMORY; - } - - for (int32_t i = 0; i < numOfOutput; ++i) { - if (strcmp(pCtx[i].pExpr->pExpr->_function.functionName, "_select_value") == 0) { - pValCtx[num++] = &pCtx[i]; - } else if (fmIsSelectFunc(pCtx[i].functionId)) { - p = &pCtx[i]; - } - // if (functionId == FUNCTION_TAG_DUMMY || functionId == FUNCTION_TS_DUMMY) { - // tagLen += pCtx[i].resDataInfo.bytes; - // pTagCtx[num++] = &pCtx[i]; - // } else if (functionId == FUNCTION_TS || functionId == FUNCTION_TAG) { - // // tag function may be the group by tag column - // // ts may be the required primary timestamp column - // continue; - // } else { - // // the column may be the normal column, group by normal_column, the functionId is FUNCTION_PRJ - // } - } - - if (p != NULL) { - p->subsidiaries.pCtx = pValCtx; - p->subsidiaries.num = num; - } else { - taosMemoryFreeClear(pValCtx); - } - - return TSDB_CODE_SUCCESS; -} - -SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput, int32_t** rowCellInfoOffset) { - SqlFunctionCtx* pFuncCtx = (SqlFunctionCtx*)taosMemoryCalloc(numOfOutput, sizeof(SqlFunctionCtx)); - if (pFuncCtx == NULL) { - return NULL; - } - - *rowCellInfoOffset = taosMemoryCalloc(numOfOutput, sizeof(int32_t)); - if (*rowCellInfoOffset == 0) { - taosMemoryFreeClear(pFuncCtx); - return NULL; - } - - for (int32_t i = 0; i < numOfOutput; ++i) { - SExprInfo* pExpr = &pExprInfo[i]; - - SExprBasicInfo* pFunct = &pExpr->base; - SqlFunctionCtx* pCtx = &pFuncCtx[i]; - - pCtx->functionId = -1; - pCtx->curBufPage = -1; - pCtx->pExpr = pExpr; - - if (pExpr->pExpr->nodeType == QUERY_NODE_FUNCTION) { - SFuncExecEnv env = {0}; - pCtx->functionId = pExpr->pExpr->_function.pFunctNode->funcId; - - if (fmIsAggFunc(pCtx->functionId) || fmIsIndefiniteRowsFunc(pCtx->functionId)) { - bool isUdaf = fmIsUserDefinedFunc(pCtx->functionId); - if (!isUdaf) { - fmGetFuncExecFuncs(pCtx->functionId, &pCtx->fpSet); - } else { - char* udfName = pExpr->pExpr->_function.pFunctNode->functionName; - strncpy(pCtx->udfName, udfName, strlen(udfName)); - fmGetUdafExecFuncs(pCtx->functionId, &pCtx->fpSet); - } - pCtx->fpSet.getEnv(pExpr->pExpr->_function.pFunctNode, &env); - } else { - fmGetScalarFuncExecFuncs(pCtx->functionId, &pCtx->sfp); - if (pCtx->sfp.getEnv != NULL) { - pCtx->sfp.getEnv(pExpr->pExpr->_function.pFunctNode, &env); - } - } - pCtx->resDataInfo.interBufSize = env.calcMemSize; - } else if (pExpr->pExpr->nodeType == QUERY_NODE_COLUMN || pExpr->pExpr->nodeType == QUERY_NODE_OPERATOR || - pExpr->pExpr->nodeType == QUERY_NODE_VALUE) { - // for simple column, the result buffer needs to hold at least one element. - pCtx->resDataInfo.interBufSize = pFunct->resSchema.bytes; - } - - pCtx->input.numOfInputCols = pFunct->numOfParams; - pCtx->input.pData = taosMemoryCalloc(pFunct->numOfParams, POINTER_BYTES); - pCtx->input.pColumnDataAgg = taosMemoryCalloc(pFunct->numOfParams, POINTER_BYTES); - - pCtx->pTsOutput = NULL; - pCtx->resDataInfo.bytes = pFunct->resSchema.bytes; - pCtx->resDataInfo.type = pFunct->resSchema.type; - pCtx->order = TSDB_ORDER_ASC; - pCtx->start.key = INT64_MIN; - pCtx->end.key = INT64_MIN; - pCtx->numOfParams = pExpr->base.numOfParams; - pCtx->increase = false; - - pCtx->param = pFunct->pParam; - } - - for (int32_t i = 1; i < numOfOutput; ++i) { - (*rowCellInfoOffset)[i] = - (int32_t)((*rowCellInfoOffset)[i - 1] + sizeof(SResultRowEntryInfo) + pFuncCtx[i - 1].resDataInfo.interBufSize); - } - - setSelectValueColumnInfo(pFuncCtx, numOfOutput); - return pFuncCtx; -} - -static void* destroySqlFunctionCtx(SqlFunctionCtx* pCtx, int32_t numOfOutput) { - if (pCtx == NULL) { - return NULL; - } - - for (int32_t i = 0; i < numOfOutput; ++i) { - for (int32_t j = 0; j < pCtx[i].numOfParams; ++j) { - taosVariantDestroy(&pCtx[i].param[j].param); - } - - taosMemoryFreeClear(pCtx[i].subsidiaries.pCtx); - taosMemoryFree(pCtx[i].input.pData); - taosMemoryFree(pCtx[i].input.pColumnDataAgg); - } - - taosMemoryFreeClear(pCtx); - return NULL; -} - bool isTaskKilled(SExecTaskInfo* pTaskInfo) { // query has been executed more than tsShellActivityTimer, and the retrieve has not arrived // abort current query execution. @@ -1568,11 +1389,10 @@ void initResultRow(SResultRow* pResultRow) { void setFunctionResultOutput(SOptrBasicInfo* pInfo, SAggSupporter* pSup, int32_t stage, int32_t numOfExprs, SExecTaskInfo* pTaskInfo) { SqlFunctionCtx* pCtx = pInfo->pCtx; - SSDataBlock* pDataBlock = pInfo->pRes; int32_t* rowCellInfoOffset = pInfo->rowCellInfoOffset; SResultRowInfo* pResultRowInfo = &pInfo->resultRowInfo; - initResultRowInfo(pResultRowInfo, 16); + initResultRowInfo(pResultRowInfo); int64_t tid = 0; int64_t groupId = 0; @@ -1580,7 +1400,7 @@ void setFunctionResultOutput(SOptrBasicInfo* pInfo, SAggSupporter* pSup, int32_t pTaskInfo, false, pSup); for (int32_t i = 0; i < numOfExprs; ++i) { - struct SResultRowEntryInfo* pEntry = getResultCell(pRow, i, rowCellInfoOffset); + struct SResultRowEntryInfo* pEntry = getResultEntryInfo(pRow, i, rowCellInfoOffset); cleanupResultRowEntry(pEntry); pCtx[i].resultInfo = pEntry; @@ -1590,42 +1410,6 @@ void setFunctionResultOutput(SOptrBasicInfo* pInfo, SAggSupporter* pSup, int32_t initCtxOutputBuffer(pCtx, numOfExprs); } -void updateOutputBuf(SOptrBasicInfo* pBInfo, int32_t* bufCapacity, int32_t numOfInputRows) { - SSDataBlock* pDataBlock = pBInfo->pRes; - - int32_t newSize = pDataBlock->info.rows + numOfInputRows + 5; // extra output buffer - if ((*bufCapacity) < newSize) { - for (int32_t i = 0; i < pDataBlock->info.numOfCols; ++i) { - SColumnInfoData* pColInfo = taosArrayGet(pDataBlock->pDataBlock, i); - - char* p = taosMemoryRealloc(pColInfo->pData, newSize * pColInfo->info.bytes); - if (p != NULL) { - pColInfo->pData = p; - - // it starts from the tail of the previously generated results. - pBInfo->pCtx[i].pOutput = pColInfo->pData; - (*bufCapacity) = newSize; - } else { - // longjmp - } - } - } - - for (int32_t i = 0; i < pDataBlock->info.numOfCols; ++i) { - SColumnInfoData* pColInfo = taosArrayGet(pDataBlock->pDataBlock, i); - pBInfo->pCtx[i].pOutput = pColInfo->pData + pColInfo->info.bytes * pDataBlock->info.rows; - - // set the correct pointer after the memory buffer reallocated. - int32_t functionId = pBInfo->pCtx[i].functionId; -#if 0 - if (functionId == FUNCTION_TOP || functionId == FUNCTION_BOTTOM || functionId == FUNCTION_DIFF || - functionId == FUNCTION_DERIVATIVE) { - // if (i > 0) pBInfo->pCtx[i].pTsOutput = pBInfo->pCtx[i - 1].pOutput; - } -#endif - } -} - void initCtxOutputBuffer(SqlFunctionCtx* pCtx, int32_t size) { for (int32_t j = 0; j < size; ++j) { struct SResultRowEntryInfo* pResInfo = GET_RES_INFO(&pCtx[j]); @@ -1659,7 +1443,7 @@ void destroyTableQueryInfoImpl(STableQueryInfo* pTableQueryInfo) { void setResultRowInitCtx(SResultRow* pResult, SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowCellInfoOffset) { for (int32_t i = 0; i < numOfOutput; ++i) { - pCtx[i].resultInfo = getResultCell(pResult, i, rowCellInfoOffset); + pCtx[i].resultInfo = getResultEntryInfo(pResult, i, rowCellInfoOffset); struct SResultRowEntryInfo* pResInfo = pCtx[i].resultInfo; if (isRowEntryCompleted(pResInfo) && isRowEntryInitialized(pResInfo)) { @@ -1793,7 +1577,7 @@ void setExecutionContext(int32_t numOfOutput, uint64_t groupId, SExecTaskInfo* p static void doUpdateNumOfRows(SResultRow* pRow, int32_t numOfExprs, const int32_t* rowCellOffset) { for (int32_t j = 0; j < numOfExprs; ++j) { - struct SResultRowEntryInfo* pResInfo = getResultCell(pRow, j, rowCellOffset); + struct SResultRowEntryInfo* pResInfo = getResultEntryInfo(pRow, j, rowCellOffset); if (!isRowEntryInitialized(pResInfo)) { continue; } @@ -1829,7 +1613,7 @@ int32_t finalizeResultRowIntoResultDataBlock(SDiskbasedBuf* pBuf, SResultRowPosi for (int32_t j = 0; j < numOfExprs; ++j) { int32_t slotId = pExprInfo[j].base.resSchema.slotId; - pCtx[j].resultInfo = getResultCell(pRow, j, rowCellOffset); + pCtx[j].resultInfo = getResultEntryInfo(pRow, j, rowCellOffset); if (pCtx[j].fpSet.finalize) { int32_t code = pCtx[j].fpSet.finalize(&pCtx[j], pBlock); if (TAOS_FAILED(code)) { @@ -1894,7 +1678,7 @@ int32_t doCopyToSDataBlock(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock, SExprI for (int32_t j = 0; j < numOfExprs; ++j) { int32_t slotId = pExprInfo[j].base.resSchema.slotId; - pCtx[j].resultInfo = getResultCell(pRow, j, rowCellOffset); + pCtx[j].resultInfo = getResultEntryInfo(pRow, j, rowCellOffset); if (pCtx[j].fpSet.finalize) { int32_t code = pCtx[j].fpSet.finalize(&pCtx[j], pBlock); if (TAOS_FAILED(code)) { @@ -1946,7 +1730,7 @@ void doBuildResultDatablock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SG SqlFunctionCtx* pCtx = pbInfo->pCtx; blockDataCleanup(pBlock); - if (!hashRemainDataInGroupInfo(pGroupResInfo)) { + if (!hasDataInGroupInfo(pGroupResInfo)) { return; } @@ -1971,7 +1755,7 @@ static void updateNumOfRowsInResultRows(SqlFunctionCtx* pCtx, int32_t numOfOutpu continue; } - SResultRowEntryInfo* pCell = getResultCell(pResult, j, rowCellInfoOffset); + SResultRowEntryInfo* pCell = getResultEntryInfo(pResult, j, rowCellInfoOffset); pResult->numOfRows = (uint16_t)(TMAX(pResult->numOfRows, pCell->numOfRes)); } } @@ -2413,33 +2197,7 @@ static int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInf return TSDB_CODE_SUCCESS; } -// NOTE: sources columns are more than the destination SSDatablock columns. -void relocateColumnData(SSDataBlock* pBlock, const SArray* pColMatchInfo, SArray* pCols) { - size_t numOfSrcCols = taosArrayGetSize(pCols); - - int32_t i = 0, j = 0; - while (i < numOfSrcCols && j < taosArrayGetSize(pColMatchInfo)) { - SColumnInfoData* p = taosArrayGet(pCols, i); - SColMatchInfo* pmInfo = taosArrayGet(pColMatchInfo, j); - if (!pmInfo->output) { - j++; - continue; - } - - if (p->info.colId == pmInfo->colId) { - SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, pmInfo->targetSlotId); - colDataAssign(pDst, p, pBlock->info.rows); - i++; - j++; - } else if (p->info.colId < pmInfo->colId) { - i++; - } else { - ASSERT(0); - } - } -} - -int32_t setDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadInfo, int32_t numOfRows, char* pData, +int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadInfo, int32_t numOfRows, char* pData, int32_t compLen, int32_t numOfOutput, int64_t startTs, uint64_t* total, SArray* pColList) { if (pColList == NULL) { // data from other sources @@ -2565,7 +2323,7 @@ static SSDataBlock* concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SEx } SRetrieveTableRsp* pTableRsp = pDataInfo->pRsp; - code = setDataBlockFromFetchRsp(pExchangeInfo->pResult, pLoadInfo, pTableRsp->numOfRows, pTableRsp->data, + code = extractDataBlockFromFetchRsp(pExchangeInfo->pResult, pLoadInfo, pTableRsp->numOfRows, pTableRsp->data, pTableRsp->compLen, pTableRsp->numOfCols, startTs, &pDataInfo->totalRows, NULL); if (code != 0) { taosMemoryFreeClear(pDataInfo->pRsp); @@ -2680,7 +2438,7 @@ static SSDataBlock* seqLoadRemoteData(SOperatorInfo* pOperator) { SSDataBlock* pRes = pExchangeInfo->pResult; SRetrieveTableRsp* pTableRsp = pDataInfo->pRsp; int32_t code = - setDataBlockFromFetchRsp(pExchangeInfo->pResult, pLoadInfo, pTableRsp->numOfRows, pTableRsp->data, + extractDataBlockFromFetchRsp(pExchangeInfo->pResult, pLoadInfo, pTableRsp->numOfRows, pTableRsp->data, pTableRsp->compLen, pTableRsp->numOfCols, startTs, &pDataInfo->totalRows, NULL); if (pRsp->completed == 1) { @@ -3152,7 +2910,7 @@ SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t } pInfo->binfo.pCtx = createSqlFunctionCtx(pExprInfo, num, &pInfo->binfo.rowCellInfoOffset); - initResultRowInfo(&pInfo->binfo.resultRowInfo, (int32_t)1); + initResultRowInfo(&pInfo->binfo.resultRowInfo); if (pInfo->binfo.pCtx == NULL || pInfo->binfo.pRes == NULL) { goto _error; @@ -3316,7 +3074,7 @@ static SSDataBlock* getAggregateResult(SOperatorInfo* pOperator) { blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity); doBuildResultDatablock(pOperator, pInfo, &pAggInfo->groupResInfo, pAggInfo->aggSup.pResultBuf); - if (pInfo->pRes->info.rows == 0 || !hashRemainDataInGroupInfo(&pAggInfo->groupResInfo)) { + if (pInfo->pRes->info.rows == 0 || !hasDataInGroupInfo(&pAggInfo->groupResInfo)) { doSetOperatorCompleted(pOperator); } @@ -3843,9 +3601,8 @@ SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo* goto _error; } - int32_t numOfGroup = 10; // todo replaced with true value pInfo->groupId = INT32_MIN; - initResultRowInfo(&pInfo->binfo.resultRowInfo, numOfGroup); + initResultRowInfo(&pInfo->binfo.resultRowInfo); pInfo->pScalarExprInfo = pScalarExprInfo; pInfo->numOfScalarExpr = numOfScalarExpr; @@ -3879,6 +3636,25 @@ _error: return NULL; } +static void* destroySqlFunctionCtx(SqlFunctionCtx* pCtx, int32_t numOfOutput) { + if (pCtx == NULL) { + return NULL; + } + + for (int32_t i = 0; i < numOfOutput; ++i) { + for (int32_t j = 0; j < pCtx[i].numOfParams; ++j) { + taosVariantDestroy(&pCtx[i].param[j].param); + } + + taosMemoryFreeClear(pCtx[i].subsidiaries.pCtx); + taosMemoryFree(pCtx[i].input.pData); + taosMemoryFree(pCtx[i].input.pColumnDataAgg); + } + + taosMemoryFreeClear(pCtx); + return NULL; +} + void doDestroyBasicInfo(SOptrBasicInfo* pInfo, int32_t numOfOutput) { assert(pInfo != NULL); @@ -3957,23 +3733,27 @@ static SArray* setRowTsColumnOutputInfo(SqlFunctionCtx* pCtx, int32_t numOfCols) return pList; } -SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t num, - SSDataBlock* pResBlock, SLimit* pLimit, SLimit* pSlimit, SNode* pCondition, - SExecTaskInfo* pTaskInfo) { +SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhysiNode* pProjPhyNode, SExecTaskInfo* pTaskInfo) { SProjectOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SProjectOperatorInfo)); SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); if (pInfo == NULL || pOperator == NULL) { goto _error; } - pInfo->limit = *pLimit; - pInfo->slimit = *pSlimit; - pInfo->curOffset = pLimit->offset; - pInfo->curSOffset = pSlimit->offset; - pInfo->binfo.pRes = pResBlock; - pInfo->pFilterNode = pCondition; + int32_t numOfCols = 0; + SExprInfo* pExprInfo = createExprInfo(pProjPhyNode->pProjections, NULL, &numOfCols); + + SSDataBlock* pResBlock = createResDataBlock(pProjPhyNode->node.pOutputDataBlockDesc); + SLimit limit = {.limit = pProjPhyNode->limit, .offset = pProjPhyNode->offset}; + SLimit slimit = {.limit = pProjPhyNode->slimit, .offset = pProjPhyNode->soffset}; + + pInfo->limit = limit; + pInfo->slimit = slimit; + pInfo->curOffset = limit.offset; + pInfo->curSOffset = slimit.offset; + pInfo->binfo.pRes = pResBlock; + pInfo->pFilterNode = pProjPhyNode->node.pConditions; - int32_t numOfCols = num; int32_t numOfRows = 4096; size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES; @@ -3988,14 +3768,14 @@ SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SExprInfo* p setFunctionResultOutput(&pInfo->binfo, &pInfo->aggSup, MAIN_SCAN, numOfCols, pTaskInfo); pInfo->pPseudoColInfo = setRowTsColumnOutputInfo(pInfo->binfo.pCtx, numOfCols); - pOperator->name = "ProjectOperator"; + pOperator->name = "ProjectOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_PROJECT; - pOperator->blocking = false; - pOperator->status = OP_NOT_OPENED; - pOperator->info = pInfo; - pOperator->pExpr = pExprInfo; - pOperator->numOfExprs = num; - pOperator->pTaskInfo = pTaskInfo; + pOperator->blocking = false; + pOperator->status = OP_NOT_OPENED; + pOperator->info = pInfo; + pOperator->pExpr = pExprInfo; + pOperator->numOfExprs = numOfCols; + pOperator->pTaskInfo = pTaskInfo; pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doProjectOperation, NULL, NULL, destroyProjectOperatorInfo, NULL, NULL, NULL); @@ -4236,151 +4016,6 @@ _error: return NULL; } -static SResSchema createResSchema(int32_t type, int32_t bytes, int32_t slotId, int32_t scale, int32_t precision, - const char* name) { - SResSchema s = {0}; - s.scale = scale; - s.type = type; - s.bytes = bytes; - s.slotId = slotId; - s.precision = precision; - strncpy(s.name, name, tListLen(s.name)); - - return s; -} - -static SColumn* createColumn(int32_t blockId, int32_t slotId, int32_t colId, SDataType* pType) { - SColumn* pCol = taosMemoryCalloc(1, sizeof(SColumn)); - if (pCol == NULL) { - terrno = TSDB_CODE_OUT_OF_MEMORY; - return NULL; - } - - pCol->slotId = slotId; - pCol->colId = colId; - pCol->bytes = pType->bytes; - pCol->type = pType->type; - pCol->scale = pType->scale; - pCol->precision = pType->precision; - pCol->dataBlockId = blockId; - - return pCol; -} - -SExprInfo* createExprInfo(SNodeList* pNodeList, SNodeList* pGroupKeys, int32_t* numOfExprs) { - int32_t numOfFuncs = LIST_LENGTH(pNodeList); - int32_t numOfGroupKeys = 0; - if (pGroupKeys != NULL) { - numOfGroupKeys = LIST_LENGTH(pGroupKeys); - } - - *numOfExprs = numOfFuncs + numOfGroupKeys; - SExprInfo* pExprs = taosMemoryCalloc(*numOfExprs, sizeof(SExprInfo)); - - for (int32_t i = 0; i < (*numOfExprs); ++i) { - STargetNode* pTargetNode = NULL; - if (i < numOfFuncs) { - pTargetNode = (STargetNode*)nodesListGetNode(pNodeList, i); - } else { - pTargetNode = (STargetNode*)nodesListGetNode(pGroupKeys, i - numOfFuncs); - } - - SExprInfo* pExp = &pExprs[i]; - - pExp->pExpr = taosMemoryCalloc(1, sizeof(tExprNode)); - pExp->pExpr->_function.num = 1; - pExp->pExpr->_function.functionId = -1; - - int32_t type = nodeType(pTargetNode->pExpr); - // it is a project query, or group by column - if (type == QUERY_NODE_COLUMN) { - pExp->pExpr->nodeType = QUERY_NODE_COLUMN; - SColumnNode* pColNode = (SColumnNode*)pTargetNode->pExpr; - - pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam)); - pExp->base.numOfParams = 1; - - SDataType* pType = &pColNode->node.resType; - pExp->base.resSchema = createResSchema(pType->type, pType->bytes, pTargetNode->slotId, pType->scale, - pType->precision, pColNode->colName); - pExp->base.pParam[0].pCol = createColumn(pColNode->dataBlockId, pColNode->slotId, pColNode->colId, pType); - pExp->base.pParam[0].type = FUNC_PARAM_TYPE_COLUMN; - } else if (type == QUERY_NODE_VALUE) { - pExp->pExpr->nodeType = QUERY_NODE_VALUE; - SValueNode* pValNode = (SValueNode*)pTargetNode->pExpr; - - pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam)); - pExp->base.numOfParams = 1; - - SDataType* pType = &pValNode->node.resType; - pExp->base.resSchema = createResSchema(pType->type, pType->bytes, pTargetNode->slotId, pType->scale, - pType->precision, pValNode->node.aliasName); - pExp->base.pParam[0].type = FUNC_PARAM_TYPE_VALUE; - nodesValueNodeToVariant(pValNode, &pExp->base.pParam[0].param); - } else if (type == QUERY_NODE_FUNCTION) { - pExp->pExpr->nodeType = QUERY_NODE_FUNCTION; - SFunctionNode* pFuncNode = (SFunctionNode*)pTargetNode->pExpr; - - SDataType* pType = &pFuncNode->node.resType; - pExp->base.resSchema = createResSchema(pType->type, pType->bytes, pTargetNode->slotId, pType->scale, - pType->precision, pFuncNode->node.aliasName); - - pExp->pExpr->_function.functionId = pFuncNode->funcId; - pExp->pExpr->_function.pFunctNode = pFuncNode; - - strncpy(pExp->pExpr->_function.functionName, pFuncNode->functionName, - tListLen(pExp->pExpr->_function.functionName)); -#if 1 - // todo refactor: add the parameter for tbname function - if (strcmp(pExp->pExpr->_function.functionName, "tbname") == 0) { - pFuncNode->pParameterList = nodesMakeList(); - ASSERT(LIST_LENGTH(pFuncNode->pParameterList) == 0); - SValueNode* res = (SValueNode*)nodesMakeNode(QUERY_NODE_VALUE); - if (NULL == res) { // todo handle error - } else { - res->node.resType = (SDataType){.bytes = sizeof(int64_t), .type = TSDB_DATA_TYPE_BIGINT}; - nodesListAppend(pFuncNode->pParameterList, (SNode*)res); - } - } -#endif - - int32_t numOfParam = LIST_LENGTH(pFuncNode->pParameterList); - - pExp->base.pParam = taosMemoryCalloc(numOfParam, sizeof(SFunctParam)); - pExp->base.numOfParams = numOfParam; - - for (int32_t j = 0; j < numOfParam; ++j) { - SNode* p1 = nodesListGetNode(pFuncNode->pParameterList, j); - if (p1->type == QUERY_NODE_COLUMN) { - SColumnNode* pcn = (SColumnNode*)p1; - - pExp->base.pParam[j].type = FUNC_PARAM_TYPE_COLUMN; - pExp->base.pParam[j].pCol = createColumn(pcn->dataBlockId, pcn->slotId, pcn->colId, &pcn->node.resType); - } else if (p1->type == QUERY_NODE_VALUE) { - SValueNode* pvn = (SValueNode*)p1; - pExp->base.pParam[j].type = FUNC_PARAM_TYPE_VALUE; - nodesValueNodeToVariant(pvn, &pExp->base.pParam[j].param); - } - } - } else if (type == QUERY_NODE_OPERATOR) { - pExp->pExpr->nodeType = QUERY_NODE_OPERATOR; - SOperatorNode* pNode = (SOperatorNode*)pTargetNode->pExpr; - - pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam)); - pExp->base.numOfParams = 1; - - SDataType* pType = &pNode->node.resType; - pExp->base.resSchema = createResSchema(pType->type, pType->bytes, pTargetNode->slotId, pType->scale, - pType->precision, pNode->node.aliasName); - pExp->pExpr->_optrRoot.pRootNode = pTargetNode->pExpr; - } else { - ASSERT(0); - } - } - - return pExprs; -} - static SExecTaskInfo* createExecTaskInfo(uint64_t queryId, uint64_t taskId, EOPTR_EXEC_MODEL model, char* dbFName) { SExecTaskInfo* pTaskInfo = taosMemoryCalloc(1, sizeof(SExecTaskInfo)); setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED); @@ -4403,8 +4038,6 @@ static tsdbReaderT doCreateDataReader(STableScanPhysiNode* pTableScanNode, SRead static SArray* extractColumnInfo(SNodeList* pNodeList); -static SArray* createSortInfo(SNodeList* pNodeList); - int32_t extractTableSchemaVersion(SReadHandle* pHandle, uint64_t uid, SExecTaskInfo* pTaskInfo) { SMetaReader mr = {0}; metaReaderInit(&mr, pHandle->meta, 0); @@ -4571,7 +4204,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo if (pHandle->vnode) { pDataReader = doCreateDataReader(pTableScanNode, pHandle, pTableListInfo, (uint64_t)queryId, taskId, pTagCond); } else { - getTableList(pHandle->meta, pScanPhyNode->tableType, pScanPhyNode->uid, pTableListInfo, pTagCond); + getTableList(pHandle->meta, pScanPhyNode, pTableListInfo, pTagCond); } if (pDataReader == NULL && terrno != 0) { @@ -4598,8 +4231,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo } else if (QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN == type) { STagScanPhysiNode* pScanPhyNode = (STagScanPhysiNode*)pPhyNode; - int32_t code = getTableList(pHandle->meta, pScanPhyNode->tableType, pScanPhyNode->uid, pTableListInfo, - pScanPhyNode->node.pConditions); + int32_t code = getTableList(pHandle->meta, pScanPhyNode, pTableListInfo, pScanPhyNode->node.pConditions); if (code != TSDB_CODE_SUCCESS) { pTaskInfo->code = terrno; return NULL; @@ -4625,14 +4257,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo SOperatorInfo* pOptr = NULL; if (QUERY_NODE_PHYSICAL_PLAN_PROJECT == type) { - SProjectPhysiNode* pProjPhyNode = (SProjectPhysiNode*)pPhyNode; - SExprInfo* pExprInfo = createExprInfo(pProjPhyNode->pProjections, NULL, &num); - - SSDataBlock* pResBlock = createResDataBlock(pPhyNode->pOutputDataBlockDesc); - SLimit limit = {.limit = pProjPhyNode->limit, .offset = pProjPhyNode->offset}; - SLimit slimit = {.limit = pProjPhyNode->slimit, .offset = pProjPhyNode->soffset}; - pOptr = createProjectOperatorInfo(ops[0], pExprInfo, num, pResBlock, &limit, &slimit, - pProjPhyNode->node.pConditions, pTaskInfo); + pOptr = createProjectOperatorInfo(ops[0], (SProjectPhysiNode*)pPhyNode, pTaskInfo); } else if (QUERY_NODE_PHYSICAL_PLAN_HASH_AGG == type) { SAggPhysiNode* pAggNode = (SAggPhysiNode*)pPhyNode; SExprInfo* pExprInfo = createExprInfo(pAggNode->pAggFuncs, pAggNode->pGroupKeys, &num); @@ -4699,21 +4324,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo int32_t children = 1; pOptr = createStreamFinalIntervalOperatorInfo(ops[0], pPhyNode, pTaskInfo, children); } else if (QUERY_NODE_PHYSICAL_PLAN_SORT == type) { - SSortPhysiNode* pSortPhyNode = (SSortPhysiNode*)pPhyNode; - - SDataBlockDescNode* pDescNode = pPhyNode->pOutputDataBlockDesc; - - SSDataBlock* pResBlock = createResDataBlock(pDescNode); - SArray* info = createSortInfo(pSortPhyNode->pSortKeys); - - int32_t numOfCols = 0; - SExprInfo* pExprInfo = createExprInfo(pSortPhyNode->pExprs, NULL, &numOfCols); - - int32_t numOfOutputCols = 0; - SArray* pColList = - extractColMatchInfo(pSortPhyNode->pTargets, pDescNode, &numOfOutputCols, pTaskInfo, COL_MATCH_FROM_SLOT_ID); - - pOptr = createSortOperatorInfo(ops[0], pResBlock, info, pExprInfo, numOfCols, pColList, pTaskInfo); + pOptr = createSortOperatorInfo(ops[0], (SSortPhysiNode*)pPhyNode, pTaskInfo); } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE == type) { SMergePhysiNode* pMergePhyNode = (SMergePhysiNode*)pPhyNode; @@ -4723,7 +4334,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo SArray* sortInfo = createSortInfo(pMergePhyNode->pMergeKeys); int32_t numOfOutputCols = 0; SArray* pColList = - extractColMatchInfo(pMergePhyNode->pTargets, pDescNode, &numOfOutputCols, pTaskInfo, COL_MATCH_FROM_SLOT_ID); + extractColMatchInfo(pMergePhyNode->pTargets, pDescNode, &numOfOutputCols, COL_MATCH_FROM_SLOT_ID); SPhysiNode* pChildNode = (SPhysiNode*)nodesListGetNode(pPhyNode->pChildren, 0); SSDataBlock* pInputDataBlock = createResDataBlock(pChildNode->pOutputDataBlockDesc); pOptr = createMultiwaySortMergeOperatorInfo(ops, size, pInputDataBlock, pResBlock, sortInfo, pColList, pTaskInfo); @@ -4769,11 +4380,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE == type) { pOptr = createStreamStateAggOperatorInfo(ops[0], pPhyNode, pTaskInfo); } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN == type) { - SJoinPhysiNode* pJoinNode = (SJoinPhysiNode*)pPhyNode; - SSDataBlock* pResBlock = createResDataBlock(pPhyNode->pOutputDataBlockDesc); - - SExprInfo* pExprInfo = createExprInfo(pJoinNode->pTargets, NULL, &num); - pOptr = createMergeJoinOperatorInfo(ops, size, pExprInfo, num, pResBlock, pJoinNode->pOnConditions, pTaskInfo); + pOptr = createMergeJoinOperatorInfo(ops, size, (SJoinPhysiNode*)pPhyNode, pTaskInfo); } else if (QUERY_NODE_PHYSICAL_PLAN_FILL == type) { pOptr = createFillOperatorInfo(ops[0], (SFillPhysiNode*)pPhyNode, false, pTaskInfo); } else if (QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC == type) { @@ -4798,79 +4405,6 @@ int32_t compareTimeWindow(const void* p1, const void* p2, const void* param) { return 0; } -int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode) { - pCond->loadExternalRows = false; - - pCond->order = pTableScanNode->scanSeq[0] > 0 ? TSDB_ORDER_ASC : TSDB_ORDER_DESC; - pCond->numOfCols = LIST_LENGTH(pTableScanNode->scan.pScanCols); - pCond->colList = taosMemoryCalloc(pCond->numOfCols, sizeof(SColumnInfo)); - if (pCond->colList == NULL) { - terrno = TSDB_CODE_QRY_OUT_OF_MEMORY; - return terrno; - } - - // pCond->twindow = pTableScanNode->scanRange; - // TODO: get it from stable scan node - pCond->numOfTWindows = 1; - pCond->twindows = taosMemoryCalloc(pCond->numOfTWindows, sizeof(STimeWindow)); - pCond->twindows[0] = pTableScanNode->scanRange; - pCond->suid = pTableScanNode->scan.suid; - -#if 1 - // todo work around a problem, remove it later - for (int32_t i = 0; i < pCond->numOfTWindows; ++i) { - if ((pCond->order == TSDB_ORDER_ASC && pCond->twindows[i].skey > pCond->twindows[i].ekey) || - (pCond->order == TSDB_ORDER_DESC && pCond->twindows[i].skey < pCond->twindows[i].ekey)) { - TSWAP(pCond->twindows[i].skey, pCond->twindows[i].ekey); - } - } -#endif - - for (int32_t i = 0; i < pCond->numOfTWindows; ++i) { - if ((pCond->order == TSDB_ORDER_ASC && pCond->twindows[i].skey > pCond->twindows[i].ekey) || - (pCond->order == TSDB_ORDER_DESC && pCond->twindows[i].skey < pCond->twindows[i].ekey)) { - TSWAP(pCond->twindows[i].skey, pCond->twindows[i].ekey); - } - } - taosqsort(pCond->twindows, pCond->numOfTWindows, sizeof(STimeWindow), pCond, compareTimeWindow); - - pCond->type = BLOCK_LOAD_OFFSET_SEQ_ORDER; - // pCond->type = pTableScanNode->scanFlag; - - int32_t j = 0; - for (int32_t i = 0; i < pCond->numOfCols; ++i) { - STargetNode* pNode = (STargetNode*)nodesListGetNode(pTableScanNode->scan.pScanCols, i); - SColumnNode* pColNode = (SColumnNode*)pNode->pExpr; - if (pColNode->colType == COLUMN_TYPE_TAG) { - continue; - } - - pCond->colList[j].type = pColNode->node.resType.type; - pCond->colList[j].bytes = pColNode->node.resType.bytes; - pCond->colList[j].colId = pColNode->colId; - j += 1; - } - - pCond->numOfCols = j; - return TSDB_CODE_SUCCESS; -} - -void clearupQueryTableDataCond(SQueryTableDataCond* pCond) { - taosMemoryFree(pCond->twindows); - taosMemoryFree(pCond->colList); -} - -SColumn extractColumnFromColumnNode(SColumnNode* pColNode) { - SColumn c = {0}; - c.slotId = pColNode->slotId; - c.colId = pColNode->colId; - c.type = pColNode->node.resType.type; - c.bytes = pColNode->node.resType.bytes; - c.scale = pColNode->node.resType.scale; - c.precision = pColNode->node.resType.precision; - return c; -} - SArray* extractColumnInfo(SNodeList* pNodeList) { size_t numOfCols = LIST_LENGTH(pNodeList); SArray* pList = taosArrayInit(numOfCols, sizeof(SColumn)); @@ -4892,7 +4426,7 @@ SArray* extractColumnInfo(SNodeList* pNodeList) { SColumn c = {0}; c.slotId = pNode->slotId; c.colId = pNode->slotId; - c.type = pValNode->node.type; + c.type = pValNode->node.type; c.bytes = pValNode->node.resType.bytes; c.scale = pValNode->node.resType.scale; c.precision = pValNode->node.resType.precision; @@ -4904,146 +4438,10 @@ SArray* extractColumnInfo(SNodeList* pNodeList) { return pList; } -SArray* extractPartitionColInfo(SNodeList* pNodeList) { - if(!pNodeList) { - return NULL; - } - - size_t numOfCols = LIST_LENGTH(pNodeList); - SArray* pList = taosArrayInit(numOfCols, sizeof(SColumn)); - if (pList == NULL) { - terrno = TSDB_CODE_OUT_OF_MEMORY; - return NULL; - } - - for (int32_t i = 0; i < numOfCols; ++i) { - SColumnNode* pColNode = (SColumnNode*)nodesListGetNode(pNodeList, i); - - // todo extract method - SColumn c = {0}; - c.slotId = pColNode->slotId; - c.colId = pColNode->colId; - c.type = pColNode->node.resType.type; - c.bytes = pColNode->node.resType.bytes; - c.precision = pColNode->node.resType.precision; - c.scale = pColNode->node.resType.scale; - - taosArrayPush(pList, &c); - } - - return pList; -} - -SArray* createSortInfo(SNodeList* pNodeList) { - size_t numOfCols = LIST_LENGTH(pNodeList); - SArray* pList = taosArrayInit(numOfCols, sizeof(SBlockOrderInfo)); - if (pList == NULL) { - terrno = TSDB_CODE_OUT_OF_MEMORY; - return pList; - } - - for (int32_t i = 0; i < numOfCols; ++i) { - SOrderByExprNode* pSortKey = (SOrderByExprNode*)nodesListGetNode(pNodeList, i); - SBlockOrderInfo bi = {0}; - bi.order = (pSortKey->order == ORDER_ASC) ? TSDB_ORDER_ASC : TSDB_ORDER_DESC; - bi.nullFirst = (pSortKey->nullOrder == NULL_ORDER_FIRST); - - SColumnNode* pColNode = (SColumnNode*)pSortKey->pExpr; - bi.slotId = pColNode->slotId; - taosArrayPush(pList, &bi); - } - - return pList; -} - -SArray* extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNodeList, int32_t* numOfOutputCols, - SExecTaskInfo* pTaskInfo, int32_t type) { - size_t numOfCols = LIST_LENGTH(pNodeList); - SArray* pList = taosArrayInit(numOfCols, sizeof(SColMatchInfo)); - if (pList == NULL) { - terrno = TSDB_CODE_OUT_OF_MEMORY; - return NULL; - } - - for (int32_t i = 0; i < numOfCols; ++i) { - STargetNode* pNode = (STargetNode*)nodesListGetNode(pNodeList, i); - SColumnNode* pColNode = (SColumnNode*)pNode->pExpr; - - SColMatchInfo c = {0}; - c.output = true; - c.colId = pColNode->colId; - c.srcSlotId = pColNode->slotId; - c.matchType = type; - c.targetSlotId = pNode->slotId; - taosArrayPush(pList, &c); - } - - *numOfOutputCols = 0; - int32_t num = LIST_LENGTH(pOutputNodeList->pSlots); - for (int32_t i = 0; i < num; ++i) { - SSlotDescNode* pNode = (SSlotDescNode*)nodesListGetNode(pOutputNodeList->pSlots, i); - - // todo: add reserve flag check - // it is a column reserved for the arithmetic expression calculation - if (pNode->slotId >= numOfCols) { - (*numOfOutputCols) += 1; - continue; - } - - SColMatchInfo* info = taosArrayGet(pList, pNode->slotId); - if (pNode->output) { - (*numOfOutputCols) += 1; - } else { - info->output = false; - } - } - - return pList; -} - -int32_t getTableList(void* metaHandle, int32_t tableType, uint64_t tableUid, STableListInfo* pListInfo, - SNode* pTagCond) { - int32_t code = TSDB_CODE_SUCCESS; - pListInfo->pTableList = taosArrayInit(8, sizeof(STableKeyInfo)); - - if (tableType == TSDB_SUPER_TABLE) { - if (pTagCond) { - SIndexMetaArg metaArg = { - .metaEx = metaHandle, .idx = tsdbGetIdx(metaHandle), .ivtIdx = tsdbGetIvtIdx(metaHandle), .suid = tableUid}; - - SArray* res = taosArrayInit(8, sizeof(uint64_t)); - code = doFilterTag(pTagCond, &metaArg, res); - if (code == TSDB_CODE_INDEX_REBUILDING) { // todo - // doFilter(); - } else if (code != TSDB_CODE_SUCCESS) { - qError("failed to get tableIds, reason: %s, suid: %" PRIu64 "", tstrerror(code), tableUid); - taosArrayDestroy(res); - terrno = code; - return code; - } else { - qDebug("sucess to get tableIds, size: %d, suid: %" PRIu64 "", (int)taosArrayGetSize(res), tableUid); - } - - for (int i = 0; i < taosArrayGetSize(res); i++) { - STableKeyInfo info = {.lastKey = TSKEY_INITIAL_VAL, .uid = *(uint64_t*)taosArrayGet(res, i)}; - taosArrayPush(pListInfo->pTableList, &info); - } - taosArrayDestroy(res); - } else { - code = tsdbGetAllTableList(metaHandle, tableUid, pListInfo->pTableList); - } - } else { // Create one table group. - STableKeyInfo info = {.lastKey = 0, .uid = tableUid}; - taosArrayPush(pListInfo->pTableList, &info); - } - - return code; -} - tsdbReaderT doCreateDataReader(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle, STableListInfo* pTableListInfo, uint64_t queryId, uint64_t taskId, SNode* pTagCond) { int32_t code = - getTableList(pHandle->meta, pTableScanNode->scan.tableType, pTableScanNode->scan.uid, pTableListInfo, pTagCond); + getTableList(pHandle->meta, &pTableScanNode->scan, pTableListInfo, pTagCond); if (code != TSDB_CODE_SUCCESS) { goto _error; } @@ -5061,7 +4459,7 @@ tsdbReaderT doCreateDataReader(STableScanPhysiNode* pTableScanNode, SReadHandle* } tsdbReaderT* pReader = tsdbReaderOpen(pHandle->vnode, &cond, pTableListInfo, queryId, taskId); - clearupQueryTableDataCond(&cond); + cleanupQueryTableDataCond(&cond); return pReader; diff --git a/source/libs/executor/src/groupoperator.c b/source/libs/executor/src/groupoperator.c index 4fc25688c4..75ba1d5d7b 100644 --- a/source/libs/executor/src/groupoperator.c +++ b/source/libs/executor/src/groupoperator.c @@ -26,8 +26,10 @@ #include "ttypes.h" #include "executorInt.h" +static void* getCurrentDataGroupInfo(const SPartitionOperatorInfo* pInfo, SDataGroupInfo** pGroupInfo, int32_t len); static int32_t* setupColumnOffset(const SSDataBlock* pBlock, int32_t rowCapacity); -static void* getCurrentDataGroupInfo(const SPartitionOperatorInfo* pInfo, SDataGroupInfo** pGroupInfo, int32_t len); +static int32_t setGroupResultOutputBuf(SOptrBasicInfo* binfo, int32_t numOfCols, char* pData, int16_t type, int16_t bytes, + int32_t groupId, SDiskbasedBuf* pBuf, SExecTaskInfo* pTaskInfo, SAggSupporter* pAggSup); static void destroyGroupOperatorInfo(void* param, int32_t numOfOutput) { SGroupbyOperatorInfo* pInfo = (SGroupbyOperatorInfo*)param; @@ -291,7 +293,7 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator) { doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf); size_t rows = pRes->info.rows; - if (rows == 0 || !hashRemainDataInGroupInfo(&pInfo->groupResInfo)) { + if (rows == 0 || !hasDataInGroupInfo(&pInfo->groupResInfo)) { doSetOperatorCompleted(pOperator); } @@ -355,7 +357,7 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator) { doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf); doFilter(pInfo->pCondition, pRes); - bool hasRemain = hashRemainDataInGroupInfo(&pInfo->groupResInfo); + bool hasRemain = hasDataInGroupInfo(&pInfo->groupResInfo); if (!hasRemain) { doSetOperatorCompleted(pOperator); break; @@ -395,7 +397,7 @@ SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pEx initResultSizeInfo(pOperator, 4096); initAggInfo(&pInfo->binfo, &pInfo->aggSup, pExprInfo, numOfCols, pResultBlock, pInfo->groupKeyLen, pTaskInfo->id.str); - initResultRowInfo(&pInfo->binfo.resultRowInfo, 8); + initResultRowInfo(&pInfo->binfo.resultRowInfo); pOperator->name = "GroupbyAggOperator"; pOperator->blocking = true; @@ -738,4 +740,18 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition taosMemoryFreeClear(pInfo); taosMemoryFreeClear(pOperator); return NULL; +} + +int32_t setGroupResultOutputBuf(SOptrBasicInfo* binfo, int32_t numOfCols, char* pData, int16_t type, int16_t bytes, + int32_t groupId, SDiskbasedBuf* pBuf, SExecTaskInfo* pTaskInfo, + SAggSupporter* pAggSup) { + SResultRowInfo* pResultRowInfo = &binfo->resultRowInfo; + SqlFunctionCtx* pCtx = binfo->pCtx; + + SResultRow* pResultRow = + doSetResultOutBufByKey(pBuf, pResultRowInfo, (char*)pData, bytes, true, groupId, pTaskInfo, false, pAggSup); + assert(pResultRow != NULL); + + setResultRowInitCtx(pResultRow, pCtx, numOfCols, binfo->rowCellInfoOffset); + return TSDB_CODE_SUCCESS; } \ No newline at end of file diff --git a/source/libs/executor/src/joinoperator.c b/source/libs/executor/src/joinoperator.c index 7c8ab244a1..6ac3f1a16c 100644 --- a/source/libs/executor/src/joinoperator.c +++ b/source/libs/executor/src/joinoperator.c @@ -28,27 +28,32 @@ static SSDataBlock* doMergeJoin(struct SOperatorInfo* pOperator); static void destroyMergeJoinOperator(void* param, int32_t numOfOutput); static void extractTimeCondition(SJoinOperatorInfo* Info, SLogicConditionNode* pLogicConditionNode); -SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOfDownstream, SExprInfo* pExprInfo, - int32_t numOfCols, SSDataBlock* pResBlock, SNode* pOnCondition, - SExecTaskInfo* pTaskInfo) { +SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOfDownstream, SJoinPhysiNode* pJoinNode, + SExecTaskInfo* pTaskInfo) { SJoinOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SJoinOperatorInfo)); SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); if (pOperator == NULL || pInfo == NULL) { goto _error; } + SSDataBlock* pResBlock = createResDataBlock(pJoinNode->node.pOutputDataBlockDesc); + + int32_t numOfCols = 0; + SExprInfo* pExprInfo = createExprInfo(pJoinNode->pTargets, NULL, &numOfCols); + initResultSizeInfo(pOperator, 4096); - pInfo->pRes = pResBlock; - pOperator->name = "MergeJoinOperator"; + pInfo->pRes = pResBlock; + pOperator->name = "MergeJoinOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN; - pOperator->blocking = false; - pOperator->status = OP_NOT_OPENED; - pOperator->pExpr = pExprInfo; - pOperator->numOfExprs = numOfCols; - pOperator->info = pInfo; - pOperator->pTaskInfo = pTaskInfo; + pOperator->blocking = false; + pOperator->status = OP_NOT_OPENED; + pOperator->pExpr = pExprInfo; + pOperator->numOfExprs = numOfCols; + pOperator->info = pInfo; + pOperator->pTaskInfo = pTaskInfo; + SNode* pOnCondition = pJoinNode->pOnConditions; if (nodeType(pOnCondition) == QUERY_NODE_OPERATOR) { SOperatorNode* pNode = (SOperatorNode*)pOnCondition; setJoinColumnInfo(&pInfo->leftCol, (SColumnNode*)pNode->pLeft); diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index b0325ef8d1..8a9972dda1 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -496,18 +496,6 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) { return NULL; } -SInterval extractIntervalInfo(const STableScanPhysiNode* pTableScanNode) { - SInterval interval = { - .interval = pTableScanNode->interval, - .sliding = pTableScanNode->sliding, - .intervalUnit = pTableScanNode->intervalUnit, - .slidingUnit = pTableScanNode->slidingUnit, - .offset = pTableScanNode->offset, - }; - - return interval; -} - static int32_t getTableScannerExecInfo(struct SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) { SFileBlockLoadRecorder* pRecorder = taosMemoryCalloc(1, sizeof(SFileBlockLoadRecorder)); STableScanInfo* pTableScanInfo = pOptr->info; @@ -520,7 +508,7 @@ static int32_t getTableScannerExecInfo(struct SOperatorInfo* pOptr, void** pOptr static void destroyTableScanOperatorInfo(void* param, int32_t numOfOutput) { STableScanInfo* pTableScanInfo = (STableScanInfo*)param; blockDataDestroy(pTableScanInfo->pResBlock); - clearupQueryTableDataCond(&pTableScanInfo->cond); + cleanupQueryTableDataCond(&pTableScanInfo->cond); tsdbCleanupReadHandle(pTableScanInfo->dataReader); @@ -540,8 +528,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc; int32_t numOfCols = 0; - SArray* pColList = - extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, pTaskInfo, COL_MATCH_FROM_COL_ID); + SArray* pColList = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID); int32_t code = initQueryTableDataCond(&pInfo->cond, pTableScanNode); if (code != TSDB_CODE_SUCCESS) { @@ -1064,8 +1051,7 @@ SOperatorInfo* createStreamScanOperatorInfo(void* pDataReader, SReadHandle* pHan STableScanInfo* pSTInfo = (STableScanInfo*)pTableScanDummy->info; int32_t numOfCols = 0; - pInfo->pColMatchInfo = - extractColMatchInfo(pScanPhyNode->pScanCols, pDescNode, &numOfCols, pTaskInfo, COL_MATCH_FROM_COL_ID); + pInfo->pColMatchInfo = extractColMatchInfo(pScanPhyNode->pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID); int32_t numOfOutput = taosArrayGetSize(pInfo->pColMatchInfo); SArray* pColIds = taosArrayInit(numOfOutput, sizeof(int16_t)); @@ -1523,7 +1509,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) { } } - setDataBlockFromFetchRsp(pInfo->pRes, &pInfo->loadInfo, pRsp->numOfRows, pRsp->data, pRsp->compLen, + extractDataBlockFromFetchRsp(pInfo->pRes, &pInfo->loadInfo, pRsp->numOfRows, pRsp->data, pRsp->compLen, pOperator->numOfExprs, startTs, NULL, pInfo->scanCols); // todo log the filter info @@ -1612,7 +1598,7 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScan SSDataBlock* pResBlock = createResDataBlock(pDescNode); int32_t num = 0; - SArray* colList = extractColMatchInfo(pScanNode->pScanCols, pDescNode, &num, pTaskInfo, COL_MATCH_FROM_COL_ID); + SArray* colList = extractColMatchInfo(pScanNode->pScanCols, pDescNode, &num, COL_MATCH_FROM_COL_ID); pInfo->accountId = pScanPhyNode->accountId; pInfo->showRewrite = pScanPhyNode->showRewrite; @@ -1818,27 +1804,26 @@ SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysi SDataBlockDescNode* pDescNode = pPhyNode->node.pOutputDataBlockDesc; + int32_t num = 0; int32_t numOfExprs = 0; SExprInfo* pExprInfo = createExprInfo(pPhyNode->pScanPseudoCols, NULL, &numOfExprs); + SArray* colList = extractColMatchInfo(pPhyNode->pScanPseudoCols, pDescNode, &num, COL_MATCH_FROM_COL_ID); - int32_t num = 0; - SArray* colList = extractColMatchInfo(pPhyNode->pScanPseudoCols, pDescNode, &num, pTaskInfo, COL_MATCH_FROM_COL_ID); + pInfo->pTableList = pTableListInfo; + pInfo->pColMatchInfo = colList; + pInfo->pRes = createResDataBlock(pDescNode); + pInfo->readHandle = *pReadHandle; + pInfo->curPos = 0; + pInfo->pFilterNode = pPhyNode->node.pConditions; - pInfo->pTableList = pTableListInfo; - pInfo->pColMatchInfo = colList; - pInfo->pRes = createResDataBlock(pDescNode); - ; - pInfo->readHandle = *pReadHandle; - pInfo->curPos = 0; - pInfo->pFilterNode = pPhyNode->node.pConditions; - pOperator->name = "TagScanOperator"; + pOperator->name = "TagScanOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN; - pOperator->blocking = false; - pOperator->status = OP_NOT_OPENED; - pOperator->info = pInfo; - pOperator->pExpr = pExprInfo; - pOperator->numOfExprs = numOfExprs; - pOperator->pTaskInfo = pTaskInfo; + pOperator->blocking = false; + pOperator->status = OP_NOT_OPENED; + pOperator->info = pInfo; + pOperator->pExpr = pExprInfo; + pOperator->numOfExprs = numOfExprs; + pOperator->pTaskInfo = pTaskInfo; initResultSizeInfo(pOperator, 4096); blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity); @@ -1908,7 +1893,7 @@ int32_t createMultipleDataReaders(STableScanPhysiNode* pTableScanNode, SReadHand STableListInfo* pTableListInfo, SArray* arrayReader, uint64_t queryId, uint64_t taskId, SNode* pTagCond) { int32_t code = - getTableList(pHandle->meta, pTableScanNode->scan.tableType, pTableScanNode->scan.uid, pTableListInfo, pTagCond); + getTableList(pHandle->meta, &pTableScanNode->scan, pTableListInfo, pTagCond); if (code != TSDB_CODE_SUCCESS) { goto _error; } @@ -1935,7 +1920,7 @@ int32_t createMultipleDataReaders(STableScanPhysiNode* pTableScanNode, SReadHand taosArrayDestroy(subListInfo->pTableList); taosMemoryFree(subListInfo); } - clearupQueryTableDataCond(&cond); + cleanupQueryTableDataCond(&cond); return 0; @@ -2211,7 +2196,7 @@ SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) { void destroyTableMergeScanOperatorInfo(void* param, int32_t numOfOutput) { STableMergeScanInfo* pTableScanInfo = (STableMergeScanInfo*)param; - clearupQueryTableDataCond(&pTableScanInfo->cond); + cleanupQueryTableDataCond(&pTableScanInfo->cond); for (int32_t i = 0; i < taosArrayGetSize(pTableScanInfo->dataReaders); ++i) { tsdbReaderT* reader = taosArrayGetP(pTableScanInfo->dataReaders, i); @@ -2261,7 +2246,7 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN int32_t numOfCols = 0; SArray* pColList = - extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, pTaskInfo, COL_MATCH_FROM_COL_ID); + extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID); int32_t code = initQueryTableDataCond(&pInfo->cond, pTableScanNode); if (code != TSDB_CODE_SUCCESS) { diff --git a/source/libs/executor/src/sortoperator.c b/source/libs/executor/src/sortoperator.c index 35e153f8c5..9821e87249 100644 --- a/source/libs/executor/src/sortoperator.c +++ b/source/libs/executor/src/sortoperator.c @@ -22,41 +22,52 @@ static int32_t getExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain static void destroyOrderOperatorInfo(void* param, int32_t numOfOutput); -SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSDataBlock* pResBlock, SArray* pSortInfo, - SExprInfo* pExprInfo, int32_t numOfCols, SArray* pColMatchColInfo, - SExecTaskInfo* pTaskInfo) { +SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* pSortPhyNode, SExecTaskInfo* pTaskInfo) { SSortOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SSortOperatorInfo)); SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); - int32_t rowSize = pResBlock->info.rowSize; - - if (pInfo == NULL || pOperator == NULL || rowSize > 100 * 1024 * 1024) { + if (pInfo == NULL || pOperator == NULL/* || rowSize > 100 * 1024 * 1024*/) { goto _error; } - pOperator->pExpr = pExprInfo; - pOperator->numOfExprs = numOfCols; + SDataBlockDescNode* pDescNode = pSortPhyNode->node.pOutputDataBlockDesc; + + int32_t numOfCols = 0; + SSDataBlock* pResBlock = createResDataBlock(pDescNode); + SExprInfo* pExprInfo = createExprInfo(pSortPhyNode->pExprs, NULL, &numOfCols); + + int32_t numOfOutputCols = 0; + SArray* pColMatchColInfo = + extractColMatchInfo(pSortPhyNode->pTargets, pDescNode, &numOfOutputCols, COL_MATCH_FROM_SLOT_ID); + pInfo->binfo.pCtx = createSqlFunctionCtx(pExprInfo, numOfCols, &pInfo->binfo.rowCellInfoOffset); pInfo->binfo.pRes = pResBlock; initResultSizeInfo(pOperator, 1024); - pInfo->pSortInfo = pSortInfo; + pInfo->pSortInfo = createSortInfo(pSortPhyNode->pSortKeys);; pInfo->pColMatchInfo = pColMatchColInfo; pOperator->name = "SortOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_SORT; pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; pOperator->info = pInfo; + pOperator->pExpr = pExprInfo; + pOperator->numOfExprs = numOfCols; + pOperator->pTaskInfo = pTaskInfo; // lazy evaluation for the following parameter since the input datablock is not known till now. - // pInfo->bufPageSize = rowSize < 1024 ? 1024 * 2 : rowSize * 2; // there are headers, so pageSize = rowSize + - // header pInfo->sortBufSize = pInfo->bufPageSize * 16; // TODO dynamic set the available sort buffer + // pInfo->bufPageSize = rowSize < 1024 ? 1024 * 2 : rowSize * 2; + // there are headers, so pageSize = rowSize + header pInfo->sortBufSize = pInfo->bufPageSize * 16; + // TODO dynamic set the available sort buffer - pOperator->pTaskInfo = pTaskInfo; pOperator->fpSet = createOperatorFpSet(doOpenSortOperator, doSort, NULL, NULL, destroyOrderOperatorInfo, NULL, NULL, getExplainExecInfo); int32_t code = appendDownstream(pOperator, &downstream, 1); + if (code != TSDB_CODE_SUCCESS) { + goto _error; + } + return pOperator; _error: diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c index d7ae823522..118970ee80 100644 --- a/source/libs/executor/src/timewindowoperator.c +++ b/source/libs/executor/src/timewindowoperator.c @@ -1090,7 +1090,7 @@ static SSDataBlock* doStateWindowAgg(SOperatorInfo* pOperator) { if (pOperator->status == OP_RES_TO_RETURN) { doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf); - if (pBInfo->pRes->info.rows == 0 || !hashRemainDataInGroupInfo(&pInfo->groupResInfo)) { + if (pBInfo->pRes->info.rows == 0 || !hasDataInGroupInfo(&pInfo->groupResInfo)) { doSetOperatorCompleted(pOperator); return NULL; } @@ -1122,7 +1122,7 @@ static SSDataBlock* doStateWindowAgg(SOperatorInfo* pOperator) { initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, TSDB_ORDER_ASC); blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity); doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf); - if (pBInfo->pRes->info.rows == 0 || !hashRemainDataInGroupInfo(&pInfo->groupResInfo)) { + if (pBInfo->pRes->info.rows == 0 || !hasDataInGroupInfo(&pInfo->groupResInfo)) { doSetOperatorCompleted(pOperator); } @@ -1153,7 +1153,7 @@ static SSDataBlock* doBuildIntervalResult(SOperatorInfo* pOperator) { blockDataEnsureCapacity(pBlock, pOperator->resultInfo.capacity); doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf); - if (pBlock->info.rows == 0 || !hashRemainDataInGroupInfo(&pInfo->groupResInfo)) { + if (pBlock->info.rows == 0 || !hasDataInGroupInfo(&pInfo->groupResInfo)) { doSetOperatorCompleted(pOperator); } @@ -1176,7 +1176,7 @@ static void finalizeUpdatedResult(int32_t numOfOutput, SDiskbasedBuf* pBuf, SArr SResultRow* pRow = (SResultRow*)((char*)bufPage + pPos->pos.offset); for (int32_t j = 0; j < numOfOutput; ++j) { - SResultRowEntryInfo* pEntry = getResultCell(pRow, j, rowCellInfoOffset); + SResultRowEntryInfo* pEntry = getResultEntryInfo(pRow, j, rowCellInfoOffset); if (pRow->numOfRows < pEntry->numOfRes) { pRow->numOfRows = pEntry->numOfRes; } @@ -1199,7 +1199,7 @@ void doClearWindowImpl(SResultRowPosition* p1, SDiskbasedBuf* pResultBuf, SOptrB SResultRow* pResult = getResultRowByPos(pResultBuf, p1); SqlFunctionCtx* pCtx = pBinfo->pCtx; for (int32_t i = 0; i < numOfOutput; ++i) { - pCtx[i].resultInfo = getResultCell(pResult, i, pBinfo->rowCellInfoOffset); + pCtx[i].resultInfo = getResultEntryInfo(pResult, i, pBinfo->rowCellInfoOffset); struct SResultRowEntryInfo* pResInfo = pCtx[i].resultInfo; if (fmIsWindowPseudoColumnFunc(pCtx[i].functionId)) { continue; @@ -1301,7 +1301,7 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) { if (pOperator->status == OP_RES_TO_RETURN) { doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf); - if (pInfo->binfo.pRes->info.rows == 0 || !hashRemainDataInGroupInfo(&pInfo->groupResInfo)) { + if (pInfo->binfo.pRes->info.rows == 0 || !hasDataInGroupInfo(&pInfo->groupResInfo)) { pOperator->status = OP_EXEC_DONE; } return pInfo->binfo.pRes->info.rows == 0 ? NULL : pInfo->binfo.pRes; @@ -1476,7 +1476,7 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* } } - initResultRowInfo(&pInfo->binfo.resultRowInfo, (int32_t)1); + initResultRowInfo(&pInfo->binfo.resultRowInfo); pOperator->name = "TimeIntervalAggOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL; @@ -1533,7 +1533,7 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SExpr goto _error; } - initResultRowInfo(&pInfo->binfo.resultRowInfo, (int32_t)1); + initResultRowInfo(&pInfo->binfo.resultRowInfo); pOperator->name = "StreamTimeIntervalAggOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL; @@ -1643,7 +1643,7 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) { if (pOperator->status == OP_RES_TO_RETURN) { doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf); - if (pBInfo->pRes->info.rows == 0 || !hashRemainDataInGroupInfo(&pInfo->groupResInfo)) { + if (pBInfo->pRes->info.rows == 0 || !hasDataInGroupInfo(&pInfo->groupResInfo)) { doSetOperatorCompleted(pOperator); return NULL; } @@ -1678,7 +1678,7 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) { initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, TSDB_ORDER_ASC); blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity); doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf); - if (pBInfo->pRes->info.rows == 0 || !hashRemainDataInGroupInfo(&pInfo->groupResInfo)) { + if (pBInfo->pRes->info.rows == 0 || !hasDataInGroupInfo(&pInfo->groupResInfo)) { doSetOperatorCompleted(pOperator); } @@ -1714,7 +1714,7 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) { // if (pOperator->status == OP_RES_TO_RETURN) { // // doBuildResultDatablock(&pRuntimeEnv->groupResInfo, pRuntimeEnv, pIntervalInfo->pRes); -// if (pResBlock->info.rows == 0 || !hashRemainDataInGroupInfo(&pSliceInfo->groupResInfo)) { +// if (pResBlock->info.rows == 0 || !hasDataInGroupInfo(&pSliceInfo->groupResInfo)) { // doSetOperatorCompleted(pOperator); // } // @@ -1908,7 +1908,7 @@ SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SExprInfo* goto _error; } - initResultRowInfo(&pInfo->binfo.resultRowInfo, 8); + initResultRowInfo(&pInfo->binfo.resultRowInfo); pInfo->pFillColInfo = createFillColInfo(pExprInfo, numOfCols, pValNode); pInfo->binfo.pRes = pResultBlock; @@ -1956,7 +1956,7 @@ SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SExprInf initResultSizeInfo(pOperator, 4096); initAggInfo(&pInfo->binfo, &pInfo->aggSup, pExpr, numOfCols, pResBlock, keyBufSize, pTaskInfo->id.str); - initResultRowInfo(&pInfo->binfo.resultRowInfo, 8); + initResultRowInfo(&pInfo->binfo.resultRowInfo); pInfo->twAggSup = *pTwAggSup; initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window); @@ -2006,7 +2006,7 @@ SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo } pInfo->twAggSup = *pTwAggSupp; - initResultRowInfo(&pInfo->binfo.resultRowInfo, 8); + initResultRowInfo(&pInfo->binfo.resultRowInfo); initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window); pInfo->tsSlotId = tsSlotId; @@ -2153,7 +2153,7 @@ static void clearStreamIntervalOperator(SStreamFinalIntervalOperatorInfo* pInfo) taosHashClear(pInfo->aggSup.pResultRowHashTable); clearDiskbasedBuf(pInfo->aggSup.pResultBuf); cleanupResultRowInfo(&pInfo->binfo.resultRowInfo); - initResultRowInfo(&pInfo->binfo.resultRowInfo, 1); + initResultRowInfo(&pInfo->binfo.resultRowInfo); } static void clearUpdateDataBlock(SSDataBlock* pBlock) { @@ -2319,7 +2319,7 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, if (code != TSDB_CODE_SUCCESS) { goto _error; } - initResultRowInfo(&pInfo->binfo.resultRowInfo, (int32_t)1); + initResultRowInfo(&pInfo->binfo.resultRowInfo); pInfo->pChildren = NULL; if (numOfChild > 0) { pInfo->pChildren = taosArrayInit(numOfChild, sizeof(SOperatorInfo)); @@ -2454,7 +2454,7 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SEx initDummyFunction(pInfo->pDummyCtx, pInfo->binfo.pCtx, numOfCols); pInfo->twAggSup = *pTwAggSupp; - initResultRowInfo(&pInfo->binfo.resultRowInfo, 8); + initResultRowInfo(&pInfo->binfo.resultRowInfo); initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window); pInfo->primaryTsIndex = tsSlotId; @@ -2896,7 +2896,7 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) { return pInfo->pDelRes; } doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf); - if (pBInfo->pRes->info.rows == 0 || !hashRemainDataInGroupInfo(&pInfo->groupResInfo)) { + if (pBInfo->pRes->info.rows == 0 || !hasDataInGroupInfo(&pInfo->groupResInfo)) { doSetOperatorCompleted(pOperator); } return pBInfo->pRes->info.rows == 0 ? NULL : pBInfo->pRes; @@ -3269,7 +3269,7 @@ static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) { return pInfo->pDelRes; } doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf); - if (pBInfo->pRes->info.rows == 0 || !hashRemainDataInGroupInfo(&pInfo->groupResInfo)) { + if (pBInfo->pRes->info.rows == 0 || !hasDataInGroupInfo(&pInfo->groupResInfo)) { doSetOperatorCompleted(pOperator); } return pBInfo->pRes->info.rows == 0 ? NULL : pBInfo->pRes; @@ -3342,7 +3342,7 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys pInfo->stateCol = extractColumnFromColumnNode(pColNode); initResultSizeInfo(pOperator, 4096); - initResultRowInfo(&pInfo->binfo.resultRowInfo, 8); + initResultRowInfo(&pInfo->binfo.resultRowInfo); pInfo->twAggSup = (STimeWindowAggSupp){ .waterMark = pStateNode->window.watermark, .calTrigger = pStateNode->window.triggerType, @@ -3590,7 +3590,7 @@ SOperatorInfo* createMergeIntervalOperatorInfo(SOperatorInfo* downstream, SExprI goto _error; } - initResultRowInfo(&iaInfo->binfo.resultRowInfo, (int32_t)1); + initResultRowInfo(&iaInfo->binfo.resultRowInfo); pOperator->name = "TimeMergeIntervalAggOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_MERGE_INTERVAL; diff --git a/source/libs/executor/test/sortTests.cpp b/source/libs/executor/test/sortTests.cpp index c037fae75f..66ed078bbe 100644 --- a/source/libs/executor/test/sortTests.cpp +++ b/source/libs/executor/test/sortTests.cpp @@ -209,7 +209,7 @@ TEST(testCase, inMem_sort_Test) { SArray* orderInfo = taosArrayInit(1, sizeof(SBlockOrderInfo)); taosArrayPush(orderInfo, &oi); - SSortHandle* phandle = tsortCreateSortHandle(orderInfo, NULL, SORT_SINGLESOURCE_SORT, 1024, 5, NULL, "test_abc"); + SSortHandle* phandle = tsortCreateSortHandle(orderInfo, SORT_SINGLESOURCE_SORT, 1024, 5, NULL, "test_abc"); tsortSetFetchRawDataFp(phandle, getSingleColDummyBlock, NULL, NULL); _info* pInfo = (_info*) taosMemoryCalloc(1, sizeof(_info)); @@ -298,7 +298,7 @@ TEST(testCase, external_mem_sort_Test) { SArray* orderInfo = taosArrayInit(1, sizeof(SBlockOrderInfo)); taosArrayPush(orderInfo, &oi); - SSortHandle* phandle = tsortCreateSortHandle(orderInfo, NULL, SORT_SINGLESOURCE_SORT, 128, 3, NULL, "test_abc"); + SSortHandle* phandle = tsortCreateSortHandle(orderInfo, SORT_SINGLESOURCE_SORT, 128, 3, NULL, "test_abc"); tsortSetFetchRawDataFp(phandle, getSingleColDummyBlock, NULL, NULL); SSortSource* ps = static_cast(taosMemoryCalloc(1, sizeof(SSortSource))); @@ -365,7 +365,7 @@ TEST(testCase, ordered_merge_sort_Test) { taosArrayPush(pBlock->pDataBlock, &colInfo); } - SSortHandle* phandle = tsortCreateSortHandle(orderInfo, NULL, SORT_MULTISOURCE_MERGE, 1024, 5, pBlock,"test_abc"); + SSortHandle* phandle = tsortCreateSortHandle(orderInfo, SORT_MULTISOURCE_MERGE, 1024, 5, pBlock,"test_abc"); tsortSetFetchRawDataFp(phandle, getSingleColDummyBlock, NULL, NULL); tsortSetComparFp(phandle, docomp); From 45d379473fe0c089b4fde83d452fe0d701643476 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Fri, 17 Jun 2022 19:49:43 +0800 Subject: [PATCH 3/8] fix(query): fix syntax error. --- source/libs/executor/src/timewindowoperator.c | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c index 795802678f..6e9cd0453e 100644 --- a/source/libs/executor/src/timewindowoperator.c +++ b/source/libs/executor/src/timewindowoperator.c @@ -2456,10 +2456,12 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh } initDummyFunction(pInfo->pDummyCtx, pInfo->binfo.pCtx, numOfCols); - pInfo->twAggSup = (STimeWindowAggSupp) {.waterMark = pSessionNode->window.watermark, + pInfo->twAggSup = (STimeWindowAggSupp) { + .waterMark = pSessionNode->window.watermark, .calTrigger = pSessionNode->window.triggerType, .maxTs = INT64_MIN}; - initResultRowInfo(&pInfo->binfo.resultRowInfo, 8); + + initResultRowInfo(&pInfo->binfo.resultRowInfo); initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window); pInfo->primaryTsIndex = tsSlotId; From 0f92fb02f9235350b2eb7989ee41b1aba9f6397b Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Fri, 17 Jun 2022 20:07:55 +0800 Subject: [PATCH 4/8] enh(query): add new api. --- source/libs/executor/inc/executorimpl.h | 2 +- source/libs/executor/src/executorMain.c | 16 ++++++++++++++++ source/libs/executor/src/executorimpl.c | 8 +++++--- 3 files changed, 22 insertions(+), 4 deletions(-) diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index 6d4cdef06e..5efb448a38 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -868,7 +868,7 @@ int32_t encodeOperator(SOperatorInfo* ops, char** data, int32_t *length); * length: the length of data * return: result code, 0 means success */ -int32_t decodeOperator(SOperatorInfo* ops, char* data, int32_t length); +int32_t decodeOperator(SOperatorInfo* ops, const char* data, int32_t length); void setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status); int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHandle* pHandle, uint64_t taskId, diff --git a/source/libs/executor/src/executorMain.c b/source/libs/executor/src/executorMain.c index 00158d7024..9ec70f1016 100644 --- a/source/libs/executor/src/executorMain.c +++ b/source/libs/executor/src/executorMain.c @@ -219,4 +219,20 @@ int32_t qGetExplainExecInfo(qTaskInfo_t tinfo, int32_t *resNum, SExplainExecInfo return getOperatorExplainExecInfo(pTaskInfo->pRoot, pRes, &capacity, resNum); } +int32_t qSerializeTaskStatus(SExecTaskInfo* pTaskInfo, char** pOutput, int32_t* len) { + if (pTaskInfo->pRoot == NULL) { + return TSDB_CODE_INVALID_PARA; + } + + return encodeOperator(pTaskInfo->pRoot, pOutput, len); +} + +int32_t qDeserializeTaskStatus(SExecTaskInfo* pTaskInfo, const char* pInput, int32_t len) { + if (pTaskInfo == NULL || pInput == NULL || len == 0) { + return TSDB_CODE_INVALID_PARA; + } + + return decodeOperator(pTaskInfo->pRoot, pInput, len); +} + diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index fe54d0e185..7ae9f54361 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -4515,15 +4515,17 @@ int32_t encodeOperator(SOperatorInfo* ops, char** result, int32_t* length) { return TDB_CODE_SUCCESS; } -int32_t decodeOperator(SOperatorInfo* ops, char* result, int32_t length) { +int32_t decodeOperator(SOperatorInfo* ops, const char* result, int32_t length) { int32_t code = TDB_CODE_SUCCESS; if (ops->fpSet.decodeResultRow) { if (result == NULL) { return TSDB_CODE_TSC_INVALID_INPUT; } + ASSERT(length == *(int32_t*)result); - char* data = result + sizeof(int32_t); - code = ops->fpSet.decodeResultRow(ops, data); + + const char* data = result + sizeof(int32_t); + code = ops->fpSet.decodeResultRow(ops, (char*) data); if (code != TDB_CODE_SUCCESS) { return code; } From d2f8a330e1ba240780e50be2e8a7dfe1fbcdc29d Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Fri, 17 Jun 2022 20:09:49 +0800 Subject: [PATCH 5/8] refactor(query): do some internal refactor. --- include/libs/executor/executor.h | 5 ++++- source/libs/executor/src/executorMain.c | 7 +++++-- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/include/libs/executor/executor.h b/include/libs/executor/executor.h index fdedb947d7..083f6ae1b0 100644 --- a/include/libs/executor/executor.h +++ b/include/libs/executor/executor.h @@ -159,11 +159,14 @@ int64_t qGetQueriedTableUid(qTaskInfo_t tinfo); */ int32_t qGetQualifiedTableIdList(void* pTableList, const char* tagCond, int32_t tagCondLen, SArray* pTableIdList); - void qProcessFetchRsp(void* parent, struct SRpcMsg* pMsg, struct SEpSet* pEpSet); int32_t qGetExplainExecInfo(qTaskInfo_t tinfo, int32_t* resNum, SExplainExecInfo** pRes); +int32_t qSerializeTaskStatus(qTaskInfo_t tinfo, char** pOutput, int32_t* len); + +int32_t qDeserializeTaskStatus(qTaskInfo_t tinfo, const char* pInput, int32_t len); + #ifdef __cplusplus } #endif diff --git a/source/libs/executor/src/executorMain.c b/source/libs/executor/src/executorMain.c index 9ec70f1016..ff281dacbd 100644 --- a/source/libs/executor/src/executorMain.c +++ b/source/libs/executor/src/executorMain.c @@ -219,7 +219,8 @@ int32_t qGetExplainExecInfo(qTaskInfo_t tinfo, int32_t *resNum, SExplainExecInfo return getOperatorExplainExecInfo(pTaskInfo->pRoot, pRes, &capacity, resNum); } -int32_t qSerializeTaskStatus(SExecTaskInfo* pTaskInfo, char** pOutput, int32_t* len) { +int32_t qSerializeTaskStatus(qTaskInfo_t tinfo, char** pOutput, int32_t* len) { + SExecTaskInfo* pTaskInfo = (struct SExecTaskInfo*)tinfo; if (pTaskInfo->pRoot == NULL) { return TSDB_CODE_INVALID_PARA; } @@ -227,7 +228,9 @@ int32_t qSerializeTaskStatus(SExecTaskInfo* pTaskInfo, char** pOutput, int32_t* return encodeOperator(pTaskInfo->pRoot, pOutput, len); } -int32_t qDeserializeTaskStatus(SExecTaskInfo* pTaskInfo, const char* pInput, int32_t len) { +int32_t qDeserializeTaskStatus(qTaskInfo_t tinfo, const char* pInput, int32_t len) { + SExecTaskInfo* pTaskInfo = (struct SExecTaskInfo*) tinfo; + if (pTaskInfo == NULL || pInput == NULL || len == 0) { return TSDB_CODE_INVALID_PARA; } From 342a1ae42d1a6bec73a4918a21e8f0dfb95ee5b7 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Fri, 17 Jun 2022 21:00:10 +0800 Subject: [PATCH 6/8] fix(stream): build ctb name --- include/client/taos.h | 21 +++++---------------- source/common/src/tdatablock.c | 1 + source/dnode/mnode/impl/src/mndScheduler.c | 4 +++- source/libs/stream/src/streamDispatch.c | 1 + source/libs/stream/src/streamTask.c | 4 ++-- 5 files changed, 12 insertions(+), 19 deletions(-) diff --git a/include/client/taos.h b/include/client/taos.h index e9f7f88387..61538e392a 100644 --- a/include/client/taos.h +++ b/include/client/taos.h @@ -209,15 +209,6 @@ DLL_EXPORT TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLi /* --------------------------TMQ INTERFACE------------------------------- */ -#if 0 -enum { - TMQ_RESP_ERR__FAIL = -1, - TMQ_RESP_ERR__SUCCESS = 0, -}; - -typedef int32_t tmq_resp_err_t; -#endif - typedef struct tmq_t tmq_t; typedef struct tmq_conf_t tmq_conf_t; typedef struct tmq_list_t tmq_list_t; @@ -236,15 +227,13 @@ DLL_EXPORT const char *tmq_err2str(int32_t code); /* ------------------------TMQ CONSUMER INTERFACE------------------------ */ -DLL_EXPORT int32_t tmq_subscribe(tmq_t *tmq, const tmq_list_t *topic_list); -DLL_EXPORT int32_t tmq_unsubscribe(tmq_t *tmq); -DLL_EXPORT int32_t tmq_subscription(tmq_t *tmq, tmq_list_t **topics); -// timeout: -1 means infinitely waiting +DLL_EXPORT int32_t tmq_subscribe(tmq_t *tmq, const tmq_list_t *topic_list); +DLL_EXPORT int32_t tmq_unsubscribe(tmq_t *tmq); +DLL_EXPORT int32_t tmq_subscription(tmq_t *tmq, tmq_list_t **topics); DLL_EXPORT TAOS_RES *tmq_consumer_poll(tmq_t *tmq, int64_t timeout); DLL_EXPORT int32_t tmq_consumer_close(tmq_t *tmq); - -DLL_EXPORT int32_t tmq_commit_sync(tmq_t *tmq, const TAOS_RES *msg); -DLL_EXPORT void tmq_commit_async(tmq_t *tmq, const TAOS_RES *msg, tmq_commit_cb *cb, void *param); +DLL_EXPORT int32_t tmq_commit_sync(tmq_t *tmq, const TAOS_RES *msg); +DLL_EXPORT void tmq_commit_async(tmq_t *tmq, const TAOS_RES *msg, tmq_commit_cb *cb, void *param); /* ----------------------TMQ CONFIGURATION INTERFACE---------------------- */ diff --git a/source/common/src/tdatablock.c b/source/common/src/tdatablock.c index 712b4fcf42..3c3d3e953d 100644 --- a/source/common/src/tdatablock.c +++ b/source/common/src/tdatablock.c @@ -1713,6 +1713,7 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks } char* buildCtbNameByGroupId(const char* stbName, uint64_t groupId) { + ASSERT(stbName[0] != 0); SArray* tags = taosArrayInit(0, sizeof(void*)); SSmlKv* pTag = taosMemoryCalloc(1, sizeof(SSmlKv)); pTag->key = "group_id"; diff --git a/source/dnode/mnode/impl/src/mndScheduler.c b/source/dnode/mnode/impl/src/mndScheduler.c index 6f8fc748c2..39bb6798aa 100644 --- a/source/dnode/mnode/impl/src/mndScheduler.c +++ b/source/dnode/mnode/impl/src/mndScheduler.c @@ -105,7 +105,7 @@ int32_t mndPersistTaskDeployReq(STrans* pTrans, SStreamTask* pTask, const SEpSet int32_t size = encoder.pos; int32_t tlen = sizeof(SMsgHead) + size; tEncoderClear(&encoder); - void* buf = taosMemoryMalloc(tlen); + void* buf = taosMemoryCalloc(1, tlen); if (buf == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; return -1; @@ -157,6 +157,7 @@ int32_t mndAddDispatcherToInnerTask(SMnode* pMnode, STrans* pTrans, SStreamObj* } sdbRelease(pMnode->pSdb, pDb); + memcpy(pTask->shuffleDispatcher.stbFullName, pStream->targetSTbName, TSDB_TABLE_FNAME_LEN); SArray* pVgs = pTask->shuffleDispatcher.dbInfo.pVgroupInfos; int32_t sz = taosArrayGetSize(pVgs); SArray* sinkLv = taosArrayGetP(pStream->tasks, 0); @@ -166,6 +167,7 @@ int32_t mndAddDispatcherToInnerTask(SMnode* pMnode, STrans* pTrans, SStreamObj* for (int32_t j = 0; j < sinkLvSize; j++) { SStreamTask* pLastLevelTask = taosArrayGetP(sinkLv, j); if (pLastLevelTask->nodeId == pVgInfo->vgId) { + ASSERT(pVgInfo->vgId > 0); pVgInfo->taskId = pLastLevelTask->taskId; ASSERT(pVgInfo->taskId != 0); break; diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c index 59ec2b5ceb..ca10e7d956 100644 --- a/source/libs/stream/src/streamDispatch.c +++ b/source/libs/stream/src/streamDispatch.c @@ -134,6 +134,7 @@ int32_t streamBuildDispatchMsg(SStreamTask* pTask, SStreamDataBlock* data, SRpcM int32_t sz = taosArrayGetSize(vgInfo); for (int32_t i = 0; i < sz; i++) { SVgroupInfo* pVgInfo = taosArrayGet(vgInfo, i); + ASSERT(pVgInfo->vgId > 0); if (hashValue >= pVgInfo->hashBegin && hashValue <= pVgInfo->hashEnd) { vgId = pVgInfo->vgId; downstreamTaskId = pVgInfo->taskId; diff --git a/source/libs/stream/src/streamTask.c b/source/libs/stream/src/streamTask.c index 7a7d9d15ad..a35e7679a1 100644 --- a/source/libs/stream/src/streamTask.c +++ b/source/libs/stream/src/streamTask.c @@ -70,7 +70,7 @@ int32_t tEncodeSStreamTask(SEncoder* pEncoder, const SStreamTask* pTask) { if (tEncodeSEpSet(pEncoder, &pTask->fixedEpDispatcher.epSet) < 0) return -1; } else if (pTask->dispatchType == TASK_DISPATCH__SHUFFLE) { if (tSerializeSUseDbRspImp(pEncoder, &pTask->shuffleDispatcher.dbInfo) < 0) return -1; - /*if (tEncodeI8(pEncoder, pTask->shuffleDispatcher.hashMethod) < 0) return -1;*/ + if (tEncodeCStr(pEncoder, pTask->shuffleDispatcher.stbFullName) < 0) return -1; } if (tEncodeI64(pEncoder, pTask->triggerParam) < 0) return -1; @@ -119,8 +119,8 @@ int32_t tDecodeSStreamTask(SDecoder* pDecoder, SStreamTask* pTask) { if (tDecodeI32(pDecoder, &pTask->fixedEpDispatcher.nodeId) < 0) return -1; if (tDecodeSEpSet(pDecoder, &pTask->fixedEpDispatcher.epSet) < 0) return -1; } else if (pTask->dispatchType == TASK_DISPATCH__SHUFFLE) { - /*if (tDecodeI8(pDecoder, &pTask->shuffleDispatcher.hashMethod) < 0) return -1;*/ if (tDeserializeSUseDbRspImp(pDecoder, &pTask->shuffleDispatcher.dbInfo) < 0) return -1; + if (tDecodeCStrTo(pDecoder, pTask->shuffleDispatcher.stbFullName) < 0) return -1; } if (tDecodeI64(pDecoder, &pTask->triggerParam) < 0) return -1; From 384b02d2a295d2add0e21c7968f9298d7f4f96cb Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Fri, 17 Jun 2022 21:39:16 +0800 Subject: [PATCH 7/8] test(tmq): add case --- examples/c/stream_demo.c | 13 ++++++++++--- tests/system-test/fulltest.sh | 2 ++ 2 files changed, 12 insertions(+), 3 deletions(-) diff --git a/examples/c/stream_demo.c b/examples/c/stream_demo.c index ab59fa5e47..5a141867e7 100644 --- a/examples/c/stream_demo.c +++ b/examples/c/stream_demo.c @@ -32,6 +32,13 @@ int32_t init_env() { } taos_free_result(pRes); + pRes = taos_query(pConn, "create database if not exists abc2 vgroups 20"); + if (taos_errno(pRes) != 0) { + printf("error in create db, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + pRes = taos_query(pConn, "use abc1"); if (taos_errno(pRes) != 0) { printf("error in use db, reason:%s\n", taos_errstr(pRes)); @@ -81,9 +88,9 @@ int32_t create_stream() { /*const char* sql = "select min(k), max(k), sum(k) as sum_of_k from st1";*/ /*const char* sql = "select sum(k) from tu1 interval(10m)";*/ /*pRes = tmq_create_stream(pConn, "stream1", "out1", sql);*/ - pRes = taos_query( - pConn, - "create stream stream1 trigger max_delay 10s into outstb as select _wstartts, sum(k) from st1 interval(10m)"); + pRes = taos_query(pConn, + "create stream stream1 trigger at_once into abc2.outstb as select _wstartts, sum(k) from st1 " + "partition by tbname interval(10m) "); if (taos_errno(pRes) != 0) { printf("failed to create stream stream1, reason:%s\n", taos_errstr(pRes)); return -1; diff --git a/tests/system-test/fulltest.sh b/tests/system-test/fulltest.sh index a4655bddbe..78cf63de92 100755 --- a/tests/system-test/fulltest.sh +++ b/tests/system-test/fulltest.sh @@ -115,6 +115,8 @@ python3 ./test.py -f 7-tmq/basic5.py python3 ./test.py -f 7-tmq/subscribeDb.py python3 ./test.py -f 7-tmq/subscribeDb0.py python3 ./test.py -f 7-tmq/subscribeDb1.py +python3 ./test.py -f 7-tmq/subscribeDb2.py +python3 ./test.py -f 7-tmq/subscribeDb3.py python3 ./test.py -f 7-tmq/subscribeStb.py python3 ./test.py -f 7-tmq/subscribeStb0.py python3 ./test.py -f 7-tmq/subscribeStb1.py From 3d15b65cea13868134adceb6bd9c67920f3b783c Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Fri, 17 Jun 2022 21:46:09 +0800 Subject: [PATCH 8/8] fix: avoid timeout event lost --- include/os/osTime.h | 28 +++++++++++++-------------- source/libs/transport/src/transComm.c | 6 ++++-- 2 files changed, 18 insertions(+), 16 deletions(-) diff --git a/include/os/osTime.h b/include/os/osTime.h index b9e407cbcf..949c15ed0d 100644 --- a/include/os/osTime.h +++ b/include/os/osTime.h @@ -23,22 +23,22 @@ extern "C" { // If the error is in a third-party library, place this header file under the third-party library header file. // When you want to use this feature, you should find or add the same function in the following section. #ifndef ALLOW_FORBID_FUNC - #define strptime STRPTIME_FUNC_TAOS_FORBID - #define gettimeofday GETTIMEOFDAY_FUNC_TAOS_FORBID - #define localtime LOCALTIME_FUNC_TAOS_FORBID - #define localtime_s LOCALTIMES_FUNC_TAOS_FORBID - #define localtime_r LOCALTIMER_FUNC_TAOS_FORBID - #define time TIME_FUNC_TAOS_FORBID - #define mktime MKTIME_FUNC_TAOS_FORBID +#define strptime STRPTIME_FUNC_TAOS_FORBID +#define gettimeofday GETTIMEOFDAY_FUNC_TAOS_FORBID +#define localtime LOCALTIME_FUNC_TAOS_FORBID +#define localtime_s LOCALTIMES_FUNC_TAOS_FORBID +#define localtime_r LOCALTIMER_FUNC_TAOS_FORBID +#define time TIME_FUNC_TAOS_FORBID +#define mktime MKTIME_FUNC_TAOS_FORBID #endif #ifdef WINDOWS - #define CLOCK_REALTIME 0 +#define CLOCK_REALTIME 0 - #define MILLISECOND_PER_SECOND (1000i64) +#define MILLISECOND_PER_SECOND (1000i64) #else - #define MILLISECOND_PER_SECOND ((int64_t)1000L) +#define MILLISECOND_PER_SECOND ((int64_t)1000L) #endif #define MILLISECOND_PER_MINUTE (MILLISECOND_PER_SECOND * 60) @@ -82,13 +82,13 @@ static FORCE_INLINE int64_t taosGetTimestampNs() { return (int64_t)systemTime.tv_sec * 1000000000L + (int64_t)systemTime.tv_nsec; } -char *taosStrpTime(const char *buf, const char *fmt, struct tm *tm); +char * taosStrpTime(const char *buf, const char *fmt, struct tm *tm); struct tm *taosLocalTime(const time_t *timep, struct tm *result); -time_t taosTime(time_t *t); -time_t taosMktime(struct tm *timep); +time_t taosTime(time_t *t); +time_t taosMktime(struct tm *timep); #ifdef __cplusplus } #endif -#endif /*_TD_OS_TIME_H_*/ +#endif /*_TD_OS_TIME_H_*/ diff --git a/source/libs/transport/src/transComm.c b/source/libs/transport/src/transComm.c index a04e8b5fca..9191b60518 100644 --- a/source/libs/transport/src/transComm.c +++ b/source/libs/transport/src/transComm.c @@ -376,17 +376,19 @@ static void transDQTimeout(uv_timer_t* timer) { SDelayQueue* queue = timer->data; tTrace("timer %p timeout", timer); uint64_t timeout = 0; + int64_t current = taosGetTimestampMs(); do { HeapNode* minNode = heapMin(queue->heap); if (minNode == NULL) break; SDelayTask* task = container_of(minNode, SDelayTask, node); - if (task->execTime <= taosGetTimestampMs()) { + + if (task->execTime <= current) { heapRemove(queue->heap, minNode); task->func(task->arg); taosMemoryFree(task); timeout = 0; } else { - timeout = task->execTime - taosGetTimestampMs(); + timeout = task->execTime - current; break; } } while (1);