diff --git a/include/libs/function/function.h b/include/libs/function/function.h index 54762a0f17ba45929414adf99f1226d25cac1242..0d8ae289cc4def348556adc5549e948cff7fcda6 100644 --- a/include/libs/function/function.h +++ b/include/libs/function/function.h @@ -158,10 +158,10 @@ typedef struct tExprNode { int32_t nodeType; union { struct {// function node - char functionName[FUNCTIONS_NAME_MAX_LENGTH]; // todo refactor - int32_t functionId; - int32_t num; - struct SFunctionNode *pFunctNode; + char functionName[FUNCTIONS_NAME_MAX_LENGTH]; // todo refactor + int32_t functionId; + int32_t num; + struct SFunctionNode *pFunctNode; } _function; struct { diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index 337b6275fc8c9bb5698b22d0daf2a15a2290a289..4f9b9f5c5d6e2bddf803d7ac2468c340bb34e6e1 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -49,8 +49,6 @@ typedef int32_t (*__block_search_fn_t)(char* data, int32_t num, int64_t key, int #define Q_STATUS_EQUAL(p, s) (((p) & (s)) != 0u) #define QUERY_IS_ASC_QUERY(q) (GET_FORWARD_DIRECTION_FACTOR((q)->order.order) == QUERY_ASC_FORWARD_STEP) -//#define GET_TABLEGROUP(q, _index) ((SArray*)taosArrayGetP((q)->tableqinfoGroupInfo.pGroupList, (_index))) - #define NEEDTO_COMPRESS_QUERY(size) ((size) > tsCompressColData ? 1 : 0) enum { @@ -64,11 +62,6 @@ enum { TASK_COMPLETED = 0x2u, }; -typedef struct SResultRowCell { - uint64_t groupId; - SResultRowPosition pos; -} SResultRowCell; - /** * If the number of generated results is greater than this value, * query query will be halt and return results to client immediate. @@ -122,35 +115,6 @@ typedef struct SOperatorCostInfo { double totalCost; } SOperatorCostInfo; -// The basic query information extracted from the SQueryInfo tree to support the -// execution of query in a data node. -typedef struct STaskAttr { - SLimit limit; - SLimit slimit; - bool stableQuery; // super table query or not - bool tsCompQuery; // is tscomp query - bool diffQuery; // is diff query - bool pointInterpQuery; // point interpolation query - int32_t havingNum; // having expr number - int16_t numOfCols; - int16_t numOfTags; - STimeWindow window; - SInterval interval; - int16_t precision; - int16_t numOfOutput; - int16_t fillType; - int32_t resultRowSize; - int32_t tagLen; // tag value length of current query - - SExprInfo* pExpr1; - SColumnInfo* tagColList; - int32_t numOfFilterCols; - int64_t* fillVal; - void* tsdb; -// STableListInfo tableGroupInfo; // table list - int32_t vgId; -} STaskAttr; - struct SOperatorInfo; typedef int32_t (*__optr_encode_fn_t)(struct SOperatorInfo* pOperator, char** result, int32_t* length); @@ -189,21 +153,6 @@ typedef struct SExecTaskInfo { struct SOperatorInfo* pRoot; } SExecTaskInfo; -typedef struct STaskRuntimeEnv { - STaskAttr* pQueryAttr; - uint32_t status; // query status - uint8_t scanFlag; // denotes reversed scan of data or not - SHashObj* pResultRowHashTable; // quick locate the window object for each result - SHashObj* pResultRowListSet; // used to check if current ResultRowInfo has ResultRow object or not - char* keyBuf; // window key buffer - STSCursor cur; - char* tagVal; // tag value of current data block - struct SOperatorInfo* proot; - int64_t currentOffset; // dynamic offset value - STableQueryInfo* current; - SResultInfo resultInfo; -} STaskRuntimeEnv; - enum { OP_NOT_OPENED = 0x0, OP_OPENED = 0x1, @@ -222,14 +171,20 @@ typedef struct SOperatorFpSet { __optr_explain_fn_t getExplainFn; } SOperatorFpSet; +typedef struct SExprSupp { + SExprInfo* pExprInfo; + int32_t numOfExprs; // the number of scalar expression in group operator + SqlFunctionCtx* pCtx; + int32_t* rowEntryInfoOffset; // offset value for each row result cell info +} SExprSupp; + typedef struct SOperatorInfo { uint8_t operatorType; bool blocking; // block operator or not uint8_t status; // denote if current operator is completed - int32_t numOfExprs; // number of columns of the current operator results char* name; // name, for debug purpose void* info; // extension attribution - SExprInfo* pExpr; + SExprSupp exprSupp; SExecTaskInfo* pTaskInfo; SOperatorCostInfo cost; SResultInfo resultInfo; @@ -244,6 +199,9 @@ typedef enum { EX_SOURCE_DATA_EXHAUSTED = 0x3, } EX_SOURCE_STATUS; +#define COL_MATCH_FROM_COL_ID 0x1 +#define COL_MATCH_FROM_SLOT_ID 0x2 + typedef struct SSourceDataInfo { int32_t index; SRetrieveTableRsp* pRsp; @@ -271,9 +229,6 @@ typedef struct SExchangeInfo { uint64_t self; } SExchangeInfo; -#define COL_MATCH_FROM_COL_ID 0x1 -#define COL_MATCH_FROM_SLOT_ID 0x2 - typedef struct SColMatchInfo { int32_t srcSlotId; // source slot id int32_t colId; @@ -283,8 +238,8 @@ typedef struct SColMatchInfo { } SColMatchInfo; typedef struct SScanInfo { - int32_t numOfAsc; - int32_t numOfDesc; + int32_t numOfAsc; + int32_t numOfDesc; } SScanInfo; typedef struct SSampleExecInfo { @@ -292,13 +247,6 @@ typedef struct SSampleExecInfo { uint32_t seed; // random seed value } SSampleExecInfo; -typedef struct SExecSupp { - SExprInfo* pExprInfo; - int32_t numOfExprs; // the number of scalar expression in group operator - SqlFunctionCtx* pCtx; - int32_t* rowEntryInfoOffset; // offset value for each row result cell info -} SExecSupp; - typedef struct STableScanInfo { void* dataReader; SReadHandle readHandle; @@ -317,7 +265,7 @@ typedef struct STableScanInfo { SArray* pColMatchInfo; int32_t numOfOutput; - SExecSupp pseudoSup; + SExprSupp pseudoSup; SQueryTableDataCond cond; int32_t scanFlag; // table scan flag to denote if it is a repeat/reverse/main scan int32_t dataBlockLoadFlag; @@ -426,12 +374,9 @@ typedef struct SBlockDistInfo { // todo remove this typedef struct SOptrBasicInfo { SResultRowInfo resultRowInfo; - int32_t* rowEntryInfoOffset; // offset value for each row result cell info - SqlFunctionCtx* pCtx; SSDataBlock* pRes; } SOptrBasicInfo; -// TODO move the resultrowsiz together with SOptrBasicInfo:rowEntryInfoOffset typedef struct SAggSupporter { SHashObj* pResultRowHashTable; // quick locate the window object for each result char* keyBuf; // window key buffer @@ -516,7 +461,7 @@ typedef struct SIndefOperatorInfo { SOptrBasicInfo binfo; SAggSupporter aggSup; SArray* pPseudoColInfo; - SExecSupp scalarSup; + SExprSupp scalarSup; } SIndefOperatorInfo; typedef struct SFillOperatorInfo { @@ -540,7 +485,7 @@ typedef struct SGroupbyOperatorInfo { char* keyBuf; // group by keys for hash int32_t groupKeyLen; // total group by column width SGroupResInfo groupResInfo; - SExecSupp scalarSup; + SExprSupp scalarSup; } SGroupbyOperatorInfo; typedef struct SDataGroupInfo { @@ -564,7 +509,7 @@ typedef struct SPartitionOperatorInfo { void* pGroupIter; // group iterator int32_t pageIndex; // page index of current group SSDataBlock* pUpdateRes; - SExecSupp scalarSup; + SExprSupp scalarSup; } SPartitionOperatorInfo; typedef struct SWindowRowsSup { @@ -717,7 +662,7 @@ SOperatorFpSet createOperatorFpSet(__optr_open_fn_t openFn, __optr_fn_t nextFn, int32_t operatorDummyOpenFn(SOperatorInfo* pOperator); void operatorDummyCloseFn(void* param, int32_t numOfCols); int32_t appendDownstream(SOperatorInfo* p, SOperatorInfo** pDownstream, int32_t num); -int32_t initAggInfo(SOptrBasicInfo* pBasicInfo, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, +int32_t initAggInfo(SOptrBasicInfo* pBasicInfo, SExprSupp *pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock, size_t keyBufSize, const char* pkey); void initResultSizeInfo(SOperatorInfo* pOperator, int32_t numOfRows); void doBuildResultDatablock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo, SDiskbasedBuf* pBuf); @@ -741,7 +686,7 @@ 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); -void cleanupExecSupp(SExecSupp* pSupp); +void cleanupExecSupp(SExprSupp* pSupp); SSDataBlock* loadNextDataBlock(void* param); diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 460b1a6d2b6f08874dc6a1ead51023ef45a1613e..5b51150321dcc332d033209b765b666aa3c0ff48 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -143,9 +143,8 @@ static int32_t doCopyToSDataBlock(SExecTaskInfo* taskInfo, SSDataBlock* pBlock, SqlFunctionCtx* pCtx, int32_t numOfExprs); static void initCtxOutputBuffer(SqlFunctionCtx* pCtx, int32_t size); -static void setResultBufSize(STaskAttr* pQueryAttr, SResultInfo* pResultInfo); -static void doSetTableGroupOutputBuf(SAggOperatorInfo* pAggInfo, int32_t numOfOutput, uint64_t groupId, - SExecTaskInfo* pTaskInfo); +static void doSetTableGroupOutputBuf(SOperatorInfo* pOperator, SAggOperatorInfo* pAggInfo, int32_t numOfOutput, uint64_t groupId); + // setup the output buffer for each operator static bool hasNull(SColumn* pColumn, SColumnDataAgg* pStatis) { @@ -161,6 +160,7 @@ static bool hasNull(SColumn* pColumn, SColumnDataAgg* pStatis) { return true; } +#if 0 static bool chkResultRowFromKey(STaskRuntimeEnv* pRuntimeEnv, SResultRowInfo* pResultRowInfo, char* pData, int16_t bytes, bool masterscan, uint64_t uid) { bool existed = false; @@ -197,6 +197,7 @@ static bool chkResultRowFromKey(STaskRuntimeEnv* pRuntimeEnv, SResultRowInfo* pR return p1 != NULL; } +#endif SResultRow* getNewResultRow(SDiskbasedBuf* pResultBuf, int64_t tableGroupId, int32_t interBufSize) { SFilePage* pData = NULL; @@ -305,36 +306,6 @@ SResultRow* doSetResultOutBufByKey(SDiskbasedBuf* pResultBuf, SResultRowInfo* pR return pResult; } -// get the correct time window according to the handled timestamp -static STimeWindow getCurrentActiveTimeWindow(SResultRowInfo* pResultRowInfo, int64_t ts, STaskAttr* pQueryAttr) { - STimeWindow w = {0}; -#if 0 - if (pResultRowInfo->curPos == -1) { // the first window, from the previous stored value - // getInitialStartTimeWindow(pQueryAttr, ts, &w); - - if (pQueryAttr->interval.intervalUnit == 'n' || pQueryAttr->interval.intervalUnit == 'y') { - w.ekey = - taosTimeAdd(w.skey, pQueryAttr->interval.interval, pQueryAttr->interval.intervalUnit, pQueryAttr->precision) - - 1; - } else { - w.ekey = w.skey + pQueryAttr->interval.interval - 1; - } - } else { - w = pRow->win; - } - - /* - * query border check, skey should not be bounded by the query time range, since the value skey will - * be used as the time window index value. So we only change ekey of time window accordingly. - */ - if (w.ekey > pQueryAttr->window.ekey && QUERY_IS_ASC_QUERY(pQueryAttr)) { - w.ekey = pQueryAttr->window.ekey; - } -#endif - - return w; -} - // a new buffer page for each table. Needs to opt this design static int32_t addNewWindowResultBuf(SResultRow* pWindowRes, SDiskbasedBuf* pResultBuf, int32_t tid, uint32_t size) { if (pWindowRes->pageId != -1) { @@ -382,13 +353,6 @@ static int32_t addNewWindowResultBuf(SResultRow* pWindowRes, SDiskbasedBuf* pRes return 0; } -static bool chkWindowOutputBufByKey(STaskRuntimeEnv* pRuntimeEnv, SResultRowInfo* pResultRowInfo, STimeWindow* win, - bool masterscan, SResultRow** pResult, int64_t groupId, SqlFunctionCtx* pCtx, - int32_t numOfOutput, int32_t* rowEntryInfoOffset) { - assert(win->skey <= win->ekey); - return chkResultRowFromKey(pRuntimeEnv, pResultRowInfo, (char*)&win->skey, TSDB_KEYSIZE, masterscan, groupId); -} - // query_range_start, query_range_end, window_duration, window_start, window_end void initExecTimeWindowInfo(SColumnInfoData* pColData, STimeWindow* pQueryWindow) { pColData->info.type = TSDB_DATA_TYPE_TIMESTAMP; @@ -457,33 +421,15 @@ void doApplyFunctions(SExecTaskInfo* taskInfo, SqlFunctionCtx* pCtx, STimeWindow } } -static FORCE_INLINE TSKEY reviseWindowEkey(STaskAttr* pQueryAttr, STimeWindow* pWindow) { - TSKEY ekey = -1; - int32_t order = TSDB_ORDER_ASC; - if (order == TSDB_ORDER_ASC) { - ekey = pWindow->ekey; - if (ekey > pQueryAttr->window.ekey) { - ekey = pQueryAttr->window.ekey; - } - } else { - ekey = pWindow->skey; - if (ekey < pQueryAttr->window.ekey) { - ekey = pQueryAttr->window.ekey; - } - } - - return ekey; -} - static int32_t doSetInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order, int32_t scanFlag, bool createDummyCol); static void doSetInputDataBlockInfo(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order) { - for (int32_t i = 0; i < pOperator->numOfExprs; ++i) { + for (int32_t i = 0; i < pOperator->exprSupp.numOfExprs; ++i) { pCtx[i].order = order; pCtx[i].input.numOfRows = pBlock->info.rows; - setBlockStatisInfo(&pCtx[i], &pOperator->pExpr[i], pBlock); + setBlockStatisInfo(&pCtx[i], &pOperator->exprSupp.pExprInfo[i], pBlock); } } @@ -542,7 +488,7 @@ static int32_t doSetInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCt int32_t scanFlag, bool createDummyCol) { int32_t code = TSDB_CODE_SUCCESS; - for (int32_t i = 0; i < pOperator->numOfExprs; ++i) { + for (int32_t i = 0; i < pOperator->exprSupp.numOfExprs; ++i) { pCtx[i].order = order; pCtx[i].input.numOfRows = pBlock->info.rows; @@ -553,7 +499,7 @@ static int32_t doSetInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCt pInput->uid = pBlock->info.uid; pInput->colDataAggIsSet = false; - SExprInfo* pOneExpr = &pOperator->pExpr[i]; + SExprInfo* pOneExpr = &pOperator->exprSupp.pExprInfo[i]; for (int32_t j = 0; j < pOneExpr->base.numOfParams; ++j) { SFunctParam* pFuncParam = &pOneExpr->base.pParam[j]; if (pFuncParam->type == FUNC_PARAM_TYPE_COLUMN) { @@ -589,7 +535,7 @@ static int32_t doSetInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCt } static int32_t doAggregateImpl(SOperatorInfo* pOperator, TSKEY startTs, SqlFunctionCtx* pCtx) { - for (int32_t k = 0; k < pOperator->numOfExprs; ++k) { + for (int32_t k = 0; k < pOperator->exprSupp.numOfExprs; ++k) { if (functionNeedToExecute(&pCtx[k])) { // todo add a dummy funtion to avoid process check if (pCtx[k].fpSet.process == NULL) { @@ -870,32 +816,6 @@ bool isTaskKilled(SExecTaskInfo* pTaskInfo) { void setTaskKilled(SExecTaskInfo* pTaskInfo) { pTaskInfo->code = TSDB_CODE_TSC_QUERY_CANCELLED; } -static bool isCachedLastQuery(STaskAttr* pQueryAttr) { - for (int32_t i = 0; i < pQueryAttr->numOfOutput; ++i) { - int32_t functionId = getExprFunctionId(&pQueryAttr->pExpr1[i]); - // if (functionId == FUNCTION_LAST || functionId == FUNCTION_LAST_DST) { - // continue; - // } - - return false; - } - - int32_t order = TSDB_ORDER_ASC; - if (order != TSDB_ORDER_DESC || !TSWINDOW_IS_EQUAL(pQueryAttr->window, TSWINDOW_DESC_INITIALIZER)) { - return false; - } - - if (pQueryAttr->interval.interval > 0) { - return false; - } - - if (pQueryAttr->numOfFilterCols > 0 || pQueryAttr->havingNum > 0) { - return false; - } - - return true; -} - ///////////////////////////////////////////////////////////////////////////////////////////// // todo refactor : return window void getAlignQueryTimeWindow(SInterval* pInterval, int32_t precision, int64_t key, STimeWindow* win) { @@ -911,7 +831,9 @@ void getAlignQueryTimeWindow(SInterval* pInterval, int32_t precision, int64_t ke } } +#if 0 static int32_t updateBlockLoadStatus(STaskAttr* pQuery, int32_t status) { + bool hasFirstLastFunc = false; bool hasOtherFunc = false; @@ -921,7 +843,7 @@ static int32_t updateBlockLoadStatus(STaskAttr* pQuery, int32_t status) { for (int32_t i = 0; i < pQuery->numOfOutput; ++i) { int32_t functionId = getExprFunctionId(&pQuery->pExpr1[i]); -#if 0 + if (functionId == FUNCTION_TS || functionId == FUNCTION_TS_DUMMY || functionId == FUNCTION_TAG || functionId == FUNCTION_TAG_DUMMY) { continue; @@ -932,7 +854,7 @@ static int32_t updateBlockLoadStatus(STaskAttr* pQuery, int32_t status) { } else { hasOtherFunc = true; } -#endif + } if (hasFirstLastFunc && status == BLK_DATA_NOT_LOAD) { @@ -946,6 +868,8 @@ static int32_t updateBlockLoadStatus(STaskAttr* pQuery, int32_t status) { return status; } +#endif + // static void updateDataCheckOrder(SQInfo *pQInfo, SQueryTableReq* pQueryMsg, bool stableQuery) { // STaskAttr* pQueryAttr = pQInfo->runtimeEnv.pQueryAttr; // @@ -1057,7 +981,7 @@ static int32_t updateBlockLoadStatus(STaskAttr* pQuery, int32_t status) { // // return filterRangeExecute(pQueryAttr->pFilters, pDataStatis, pQueryAttr->numOfCols, numOfRows); // } - +#if 0 static bool overlapWithTimeWindow(STaskAttr* pQueryAttr, SDataBlockInfo* pBlockInfo) { STimeWindow w = {0}; @@ -1106,55 +1030,7 @@ static bool overlapWithTimeWindow(STaskAttr* pQueryAttr, SDataBlockInfo* pBlockI return false; } - -void doCompactSDataBlock(SSDataBlock* pBlock, int32_t numOfRows, int8_t* p) { - int32_t len = 0; - int32_t start = 0; - for (int32_t j = 0; j < numOfRows; ++j) { - if (p[j] == 1) { - len++; - } else { - if (len > 0) { - int32_t cstart = j - len; - for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) { - SColumnInfoData* pColumnInfoData = taosArrayGet(pBlock->pDataBlock, i); - - int16_t bytes = pColumnInfoData->info.bytes; - memmove(((char*)pColumnInfoData->pData) + start * bytes, pColumnInfoData->pData + cstart * bytes, - len * bytes); - } - - start += len; - len = 0; - } - } - } - - if (len > 0) { - int32_t cstart = numOfRows - len; - for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) { - SColumnInfoData* pColumnInfoData = taosArrayGet(pBlock->pDataBlock, i); - - int16_t bytes = pColumnInfoData->info.bytes; - memmove(pColumnInfoData->pData + start * bytes, pColumnInfoData->pData + cstart * bytes, len * bytes); - } - - start += len; - len = 0; - } - - pBlock->info.rows = start; - pBlock->pBlockAgg = NULL; // clean the block statistics info - - if (start > 0) { - SColumnInfoData* pColumnInfoData = taosArrayGet(pBlock->pDataBlock, 0); - if (pColumnInfoData->info.type == TSDB_DATA_TYPE_TIMESTAMP && - pColumnInfoData->info.colId == PRIMARYKEY_TIMESTAMP_COL_ID) { - pBlock->info.window.skey = *(int64_t*)pColumnInfoData->pData; - pBlock->info.window.ekey = *(int64_t*)(pColumnInfoData->pData + TSDB_KEYSIZE * (start - 1)); - } - } -} +#endif static uint32_t doFilterByBlockTimeWindow(STableScanInfo* pTableScanInfo, SSDataBlock* pBlock) { SqlFunctionCtx* pCtx = pTableScanInfo->pCtx; @@ -1356,10 +1232,10 @@ void initResultRow(SResultRow* pResultRow) { * offset[0] offset[1] offset[2] */ // TODO refactor: some function move away -void setFunctionResultOutput(SOptrBasicInfo* pInfo, SAggSupporter* pSup, int32_t stage, int32_t numOfExprs, - SExecTaskInfo* pTaskInfo) { - SqlFunctionCtx* pCtx = pInfo->pCtx; - int32_t* rowEntryInfoOffset = pInfo->rowEntryInfoOffset; +void setFunctionResultOutput(SOperatorInfo *pOperator, SOptrBasicInfo *pInfo, SAggSupporter* pSup, int32_t stage, int32_t numOfExprs) { + SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; + SqlFunctionCtx* pCtx = pOperator->exprSupp.pCtx; + int32_t* rowEntryInfoOffset = pOperator->exprSupp.rowEntryInfoOffset; SResultRowInfo* pResultRowInfo = &pInfo->resultRowInfo; initResultRowInfo(pResultRowInfo); @@ -1506,14 +1382,12 @@ void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const int8_t* rowR } } -void doSetTableGroupOutputBuf(SAggOperatorInfo* pAggInfo, int32_t numOfOutput, uint64_t groupId, - SExecTaskInfo* pTaskInfo) { +void doSetTableGroupOutputBuf(SOperatorInfo* pOperator, SAggOperatorInfo* pAggInfo, int32_t numOfOutput, uint64_t groupId) { // for simple group by query without interval, all the tables belong to one group result. - int64_t uid = 0; - + SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SResultRowInfo* pResultRowInfo = &pAggInfo->binfo.resultRowInfo; - SqlFunctionCtx* pCtx = pAggInfo->binfo.pCtx; - int32_t* rowEntryInfoOffset = pAggInfo->binfo.rowEntryInfoOffset; + SqlFunctionCtx* pCtx = pOperator->exprSupp.pCtx; + int32_t* rowEntryInfoOffset = pOperator->exprSupp.rowEntryInfoOffset; SResultRow* pResultRow = doSetResultOutBufByKey(pAggInfo->aggSup.pResultBuf, pResultRowInfo, (char*)&groupId, sizeof(groupId), true, groupId, pTaskInfo, false, &pAggInfo->aggSup); @@ -1534,12 +1408,12 @@ void doSetTableGroupOutputBuf(SAggOperatorInfo* pAggInfo, int32_t numOfOutput, u setResultRowInitCtx(pResultRow, pCtx, numOfOutput, rowEntryInfoOffset); } -void setExecutionContext(int32_t numOfOutput, uint64_t groupId, SExecTaskInfo* pTaskInfo, SAggOperatorInfo* pAggInfo) { +void setExecutionContext(SOperatorInfo* pOperator, int32_t numOfOutput, uint64_t groupId, SAggOperatorInfo* pAggInfo) { if (pAggInfo->groupId != INT32_MIN && pAggInfo->groupId == groupId) { return; } - doSetTableGroupOutputBuf(pAggInfo, numOfOutput, groupId, pTaskInfo); + doSetTableGroupOutputBuf(pOperator, pAggInfo, numOfOutput, groupId); // record the current active group id pAggInfo->groupId = groupId; @@ -1691,13 +1565,13 @@ int32_t doCopyToSDataBlock(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock, SExprI void doBuildResultDatablock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo, SDiskbasedBuf* pBuf) { - SExprInfo* pExprInfo = pOperator->pExpr; - int32_t numOfExprs = pOperator->numOfExprs; + SExprInfo* pExprInfo = pOperator->exprSupp.pExprInfo; + int32_t numOfExprs = pOperator->exprSupp.numOfExprs; SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; - int32_t* rowCellOffset = pbInfo->rowEntryInfoOffset; + int32_t* rowCellOffset = pOperator->exprSupp.rowEntryInfoOffset; SSDataBlock* pBlock = pbInfo->pRes; - SqlFunctionCtx* pCtx = pbInfo->pCtx; + SqlFunctionCtx* pCtx = pOperator->exprSupp.pCtx; blockDataCleanup(pBlock); if (!hasDataInGroupInfo(pGroupResInfo)) { @@ -2545,7 +2419,7 @@ SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, SExchangePhysiNode pOperator->blocking = false; pOperator->status = OP_NOT_OPENED; pOperator->info = pInfo; - pOperator->numOfExprs = pInfo->pResult->info.numOfCols; + pOperator->exprSupp.numOfExprs = pInfo->pResult->info.numOfCols; pOperator->pTaskInfo = pTaskInfo; pOperator->fpSet = createOperatorFpSet(prepareLoadRemoteData, doLoadRemoteData, NULL, NULL, @@ -2666,7 +2540,7 @@ static bool saveCurrentTuple(char** rowColData, SArray* pColumnList, SSDataBlock static void doMergeImpl(SOperatorInfo* pOperator, int32_t numOfExpr, SSDataBlock* pBlock) { SSortedMergeOperatorInfo* pInfo = pOperator->info; - SqlFunctionCtx* pCtx = pInfo->binfo.pCtx; + SqlFunctionCtx* pCtx = pOperator->exprSupp.pCtx; for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) { // pCtx[i].size = 1; } @@ -2681,8 +2555,8 @@ static void doMergeImpl(SOperatorInfo* pOperator, int32_t numOfExpr, SSDataBlock doMergeResultImpl(pInfo, pCtx, numOfExpr, i); } else { doFinalizeResultImpl(pCtx, numOfExpr); - int32_t numOfRows = getNumOfResult(pInfo->binfo.pCtx, pOperator->numOfExprs, NULL); - // setTagValueForMultipleRows(pCtx, pOperator->numOfExprs, numOfRows); + int32_t numOfRows = getNumOfResult(pOperator->exprSupp.pCtx, pOperator->exprSupp.numOfExprs, NULL); + // setTagValueForMultipleRows(pCtx, pOperator->exprSupp.numOfExprs, numOfRows); // TODO check for available buffer; @@ -2729,16 +2603,16 @@ static SSDataBlock* doMerge(SOperatorInfo* pOperator) { break; } - setInputDataBlock(pOperator, pInfo->binfo.pCtx, pDataBlock, TSDB_ORDER_ASC, MAIN_SCAN, true); + setInputDataBlock(pOperator, pOperator->exprSupp.pCtx, pDataBlock, TSDB_ORDER_ASC, MAIN_SCAN, true); // updateOutputBuf(&pInfo->binfo, &pAggInfo->bufCapacity, pBlock->info.rows * pAggInfo->resultRowFactor, // pOperator->pRuntimeEnv, true); - doMergeImpl(pOperator, pOperator->numOfExprs, pDataBlock); + doMergeImpl(pOperator, pOperator->exprSupp.numOfExprs, pDataBlock); // flush to tuple store, and after all data have been handled, return to upstream node or sink node } - doFinalizeResultImpl(pInfo->binfo.pCtx, pOperator->numOfExprs); - int32_t numOfRows = getNumOfResult(pInfo->binfo.pCtx, pOperator->numOfExprs, NULL); - // setTagValueForMultipleRows(pCtx, pOperator->numOfExprs, numOfRows); + doFinalizeResultImpl(pOperator->exprSupp.pCtx, pOperator->exprSupp.numOfExprs); + int32_t numOfRows = getNumOfResult(pOperator->exprSupp.pCtx, pOperator->exprSupp.numOfExprs, NULL); + // setTagValueForMultipleRows(pCtx, pOperator->exprSupp.numOfExprs, numOfRows); // TODO check for available buffer; @@ -2879,20 +2753,20 @@ SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t goto _error; } - pInfo->binfo.pCtx = createSqlFunctionCtx(pExprInfo, num, &pInfo->binfo.rowEntryInfoOffset); + pOperator->exprSupp.pCtx = createSqlFunctionCtx(pExprInfo, num, &pOperator->exprSupp.rowEntryInfoOffset); initResultRowInfo(&pInfo->binfo.resultRowInfo); - if (pInfo->binfo.pCtx == NULL || pInfo->binfo.pRes == NULL) { + if (pOperator->exprSupp.pCtx == NULL || pInfo->binfo.pRes == NULL) { goto _error; } size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES; - int32_t code = doInitAggInfoSup(&pInfo->aggSup, pInfo->binfo.pCtx, num, keyBufSize, pTaskInfo->id.str); + int32_t code = doInitAggInfoSup(&pInfo->aggSup, pOperator->exprSupp.pCtx, num, keyBufSize, pTaskInfo->id.str); if (code != TSDB_CODE_SUCCESS) { goto _error; } - setFunctionResultOutput(&pInfo->binfo, &pInfo->aggSup, MAIN_SCAN, num, pTaskInfo); + setFunctionResultOutput(pOperator, &pInfo->binfo, &pInfo->aggSup, MAIN_SCAN, num); code = initGroupCol(pExprInfo, num, pGroupInfo, pInfo); if (code != TSDB_CODE_SUCCESS) { goto _error; @@ -2911,8 +2785,8 @@ SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; pOperator->info = pInfo; - pOperator->numOfExprs = num; - pOperator->pExpr = pExprInfo; + pOperator->exprSupp.numOfExprs = num; + pOperator->exprSupp.pExprInfo = pExprInfo; pOperator->pTaskInfo = pTaskInfo; @@ -2967,6 +2841,7 @@ static int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) { SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SAggOperatorInfo* pAggInfo = pOperator->info; + SExprSupp* pSup = &pOperator->exprSupp; SOptrBasicInfo* pInfo = &pAggInfo->binfo; SOperatorInfo* downstream = pOperator->pDownstream[0]; @@ -2996,9 +2871,9 @@ static int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) { } // the pDataBlock are always the same one, no need to call this again - setExecutionContext(pOperator->numOfExprs, pBlock->info.groupId, pTaskInfo, pAggInfo); - setInputDataBlock(pOperator, pInfo->pCtx, pBlock, order, scanFlag, true); - code = doAggregateImpl(pOperator, 0, pInfo->pCtx); + setExecutionContext(pOperator, pOperator->exprSupp.numOfExprs, pBlock->info.groupId, pAggInfo); + setInputDataBlock(pOperator, pSup->pCtx, pBlock, order, scanFlag, true); + code = doAggregateImpl(pOperator, 0, pSup->pCtx); if (code != 0) { longjmp(pTaskInfo->env, code); } @@ -3257,6 +3132,7 @@ static SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) { SProjectOperatorInfo* pProjectInfo = pOperator->info; SOptrBasicInfo* pInfo = &pProjectInfo->binfo; + SExprSupp* pSup = &pOperator->exprSupp; SSDataBlock* pRes = pInfo->pRes; blockDataCleanup(pRes); @@ -3274,10 +3150,10 @@ static SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) { setInputDataBlock(pOperator, pInfo->pCtx, pBlock, TSDB_ORDER_ASC); blockDataEnsureCapacity(pInfo->pRes, pBlock->info.rows); - projectApplyFunctions(pOperator->pExpr, pInfo->pRes, pBlock, pInfo->pCtx, pOperator->numOfExprs); + projectApplyFunctions(pOperator->exprSupp.pExprInfo, pInfo->pRes, pBlock, pInfo->pCtx, pOperator->exprSupp.numOfExprs); if (pRes->info.rows >= pProjectInfo->binfo.capacity * 0.8) { - copyTsColoum(pRes, pInfo->pCtx, pOperator->numOfExprs); - resetResultRowEntryResult(pInfo->pCtx, pOperator->numOfExprs); + copyTsColoum(pRes, pInfo->pCtx, pOperator->exprSupp.numOfExprs); + resetResultRowEntryResult(pInfo->pCtx, pOperator->exprSupp.numOfExprs); return pRes; } } @@ -3309,10 +3185,10 @@ static SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) { doFilter(pProjectInfo->pFilterNode, pBlock); - setInputDataBlock(pOperator, pInfo->pCtx, pBlock, order, scanFlag, false); + setInputDataBlock(pOperator, pSup->pCtx, pBlock, order, scanFlag, false); blockDataEnsureCapacity(pInfo->pRes, pInfo->pRes->info.rows + pBlock->info.rows); - code = projectApplyFunctions(pOperator->pExpr, pInfo->pRes, pBlock, pInfo->pCtx, pOperator->numOfExprs, + code = projectApplyFunctions(pSup->pExprInfo, pInfo->pRes, pBlock, pSup->pCtx, pSup->numOfExprs, pProjectInfo->pPseudoColInfo); if (code != TSDB_CODE_SUCCESS) { longjmp(pTaskInfo->env, code); @@ -3463,7 +3339,7 @@ static void destroyOperatorInfo(SOperatorInfo* pOperator) { } if (pOperator->fpSet.closeFn != NULL) { - pOperator->fpSet.closeFn(pOperator->info, pOperator->numOfExprs); + pOperator->fpSet.closeFn(pOperator->info, pOperator->exprSupp.numOfExprs); } if (pOperator->pDownstream != NULL) { @@ -3475,11 +3351,11 @@ static void destroyOperatorInfo(SOperatorInfo* pOperator) { pOperator->numOfDownstream = 0; } - if (pOperator->pExpr != NULL) { - destroyExprInfo(pOperator->pExpr, pOperator->numOfExprs); + if (pOperator->exprSupp.pExprInfo != NULL) { + destroyExprInfo(pOperator->exprSupp.pExprInfo, pOperator->exprSupp.numOfExprs); } - taosMemoryFreeClear(pOperator->pExpr); + taosMemoryFreeClear(pOperator->exprSupp.pExprInfo); taosMemoryFreeClear(pOperator->info); taosMemoryFreeClear(pOperator); } @@ -3529,15 +3405,15 @@ void cleanupAggSup(SAggSupporter* pAggSup) { destroyDiskbasedBuf(pAggSup->pResultBuf); } -int32_t initAggInfo(SOptrBasicInfo* pBasicInfo, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, +int32_t initAggInfo(SOptrBasicInfo* pBasicInfo, SExprSupp* pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock, size_t keyBufSize, const char* pkey) { - pBasicInfo->pCtx = createSqlFunctionCtx(pExprInfo, numOfCols, &pBasicInfo->rowEntryInfoOffset); + pSup->pCtx = createSqlFunctionCtx(pExprInfo, numOfCols, &pSup->rowEntryInfoOffset); pBasicInfo->pRes = pResultBlock; - doInitAggInfoSup(pAggSup, pBasicInfo->pCtx, numOfCols, keyBufSize, pkey); + doInitAggInfoSup(pAggSup, pSup->pCtx, numOfCols, keyBufSize, pkey); for (int32_t i = 0; i < numOfCols; ++i) { - pBasicInfo->pCtx[i].pBuf = pAggSup->pResultBuf; + pSup->pCtx[i].pBuf = pAggSup->pResultBuf; } return TSDB_CODE_SUCCESS; @@ -3566,7 +3442,7 @@ SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo* initResultSizeInfo(pOperator, numOfRows); int32_t code = - initAggInfo(&pInfo->binfo, &pInfo->aggSup, pExprInfo, numOfCols, pResultBlock, keyBufSize, pTaskInfo->id.str); + initAggInfo(&pInfo->binfo, &pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, pResultBlock, keyBufSize, pTaskInfo->id.str); if (code != TSDB_CODE_SUCCESS) { goto _error; } @@ -3585,8 +3461,8 @@ SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo* pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; pOperator->info = pInfo; - pOperator->pExpr = pExprInfo; - pOperator->numOfExprs = numOfCols; + pOperator->exprSupp.pExprInfo = pExprInfo; + pOperator->exprSupp.numOfExprs = numOfCols; pOperator->pTaskInfo = pTaskInfo; pOperator->fpSet = createOperatorFpSet(doOpenAggregateOptr, getAggregateResult, NULL, NULL, destroyAggOperatorInfo, @@ -3627,10 +3503,6 @@ static void* destroySqlFunctionCtx(SqlFunctionCtx* pCtx, int32_t numOfOutput) { void doDestroyBasicInfo(SOptrBasicInfo* pInfo, int32_t numOfOutput) { assert(pInfo != NULL); - - destroySqlFunctionCtx(pInfo->pCtx, numOfOutput); - taosMemoryFreeClear(pInfo->rowEntryInfoOffset); - cleanupResultRowInfo(&pInfo->resultRowInfo); pInfo->pRes = blockDataDestroy(pInfo->pRes); } @@ -3662,7 +3534,7 @@ static void destroyProjectOperatorInfo(void* param, int32_t numOfOutput) { taosArrayDestroy(pInfo->pPseudoColInfo); } -void cleanupExecSupp(SExecSupp* pSupp) { +void cleanupExecSupp(SExprSupp* pSupp) { destroySqlFunctionCtx(pSupp->pCtx, pSupp->numOfExprs); destroyExprInfo(pSupp->pExprInfo, pSupp->numOfExprs); @@ -3737,17 +3609,17 @@ SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhys } initResultSizeInfo(pOperator, numOfRows); - initAggInfo(&pInfo->binfo, &pInfo->aggSup, pExprInfo, numOfCols, pResBlock, keyBufSize, pTaskInfo->id.str); - setFunctionResultOutput(&pInfo->binfo, &pInfo->aggSup, MAIN_SCAN, numOfCols, pTaskInfo); + initAggInfo(&pInfo->binfo, &pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, pResBlock, keyBufSize, pTaskInfo->id.str); + setFunctionResultOutput(pOperator, &pInfo->binfo, &pInfo->aggSup, MAIN_SCAN, numOfCols); - pInfo->pPseudoColInfo = setRowTsColumnOutputInfo(pInfo->binfo.pCtx, numOfCols); + pInfo->pPseudoColInfo = setRowTsColumnOutputInfo(pOperator->exprSupp.pCtx, numOfCols); 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 = numOfCols; + pOperator->exprSupp.pExprInfo = pExprInfo; + pOperator->exprSupp.numOfExprs = numOfCols; pOperator->pTaskInfo = pTaskInfo; pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doProjectOperation, NULL, NULL, @@ -3768,6 +3640,7 @@ _error: static SSDataBlock* doApplyIndefinitFunction(SOperatorInfo* pOperator) { SIndefOperatorInfo* pIndefInfo = pOperator->info; SOptrBasicInfo* pInfo = &pIndefInfo->binfo; + SExprSupp* pSup = &pOperator->exprSupp; SSDataBlock* pRes = pInfo->pRes; blockDataCleanup(pRes); @@ -3802,7 +3675,7 @@ static SSDataBlock* doApplyIndefinitFunction(SOperatorInfo* pOperator) { } // there is an scalar expression that needs to be calculated before apply the group aggregation. - SExecSupp* pScalarSup = &pIndefInfo->scalarSup; + SExprSupp* pScalarSup = &pIndefInfo->scalarSup; if (pScalarSup->pExprInfo != NULL) { code = projectApplyFunctions(pScalarSup->pExprInfo, pBlock, pBlock, pScalarSup->pCtx, pScalarSup->numOfExprs, pIndefInfo->pPseudoColInfo); @@ -3811,10 +3684,10 @@ static SSDataBlock* doApplyIndefinitFunction(SOperatorInfo* pOperator) { } } - setInputDataBlock(pOperator, pInfo->pCtx, pBlock, order, scanFlag, false); + setInputDataBlock(pOperator, pSup->pCtx, pBlock, order, scanFlag, false); blockDataEnsureCapacity(pInfo->pRes, pInfo->pRes->info.rows + pBlock->info.rows); - code = projectApplyFunctions(pOperator->pExpr, pInfo->pRes, pBlock, pInfo->pCtx, pOperator->numOfExprs, + code = projectApplyFunctions(pOperator->exprSupp.pExprInfo, pInfo->pRes, pBlock, pSup->pCtx, pOperator->exprSupp.numOfExprs, pIndefInfo->pPseudoColInfo); if (code != TSDB_CODE_SUCCESS) { longjmp(pTaskInfo->env, code); @@ -3839,6 +3712,8 @@ SOperatorInfo* createIndefinitOutputOperatorInfo(SOperatorInfo* downstream, SPhy goto _error; } + SExprSupp* pSup = &pOperator->exprSupp; + SIndefRowsFuncPhysiNode* pPhyNode = (SIndefRowsFuncPhysiNode*)pNode; int32_t numOfExpr = 0; @@ -3846,7 +3721,7 @@ SOperatorInfo* createIndefinitOutputOperatorInfo(SOperatorInfo* downstream, SPhy int32_t numOfScalarExpr = 0; if (pPhyNode->pExprs != NULL) { - SExecSupp* pSup = &pInfo->scalarSup; + SExprSupp* pSup = &pInfo->scalarSup; pSup->pExprInfo = createExprInfo(pPhyNode->pExprs, NULL, &pSup->numOfExprs); pSup->pCtx = createSqlFunctionCtx(pSup->pExprInfo, numOfScalarExpr, &pSup->rowEntryInfoOffset); } @@ -3864,19 +3739,19 @@ SOperatorInfo* createIndefinitOutputOperatorInfo(SOperatorInfo* downstream, SPhy } initResultSizeInfo(pOperator, numOfRows); - initAggInfo(&pInfo->binfo, &pInfo->aggSup, pExprInfo, numOfExpr, pResBlock, keyBufSize, pTaskInfo->id.str); - setFunctionResultOutput(&pInfo->binfo, &pInfo->aggSup, MAIN_SCAN, numOfExpr, pTaskInfo); + initAggInfo(&pInfo->binfo, &pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfExpr, pResBlock, keyBufSize, pTaskInfo->id.str); + setFunctionResultOutput(pOperator, &pInfo->binfo, &pInfo->aggSup, MAIN_SCAN, numOfExpr); pInfo->binfo.pRes = pResBlock; - pInfo->pPseudoColInfo = setRowTsColumnOutputInfo(pInfo->binfo.pCtx, numOfExpr); + pInfo->pPseudoColInfo = setRowTsColumnOutputInfo(pSup->pCtx, numOfExpr); pOperator->name = "IndefinitOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_PROJECT; pOperator->blocking = false; pOperator->status = OP_NOT_OPENED; pOperator->info = pInfo; - pOperator->pExpr = pExprInfo; - pOperator->numOfExprs = numOfExpr; + pOperator->exprSupp.pExprInfo = pExprInfo; + pOperator->exprSupp.numOfExprs = numOfExpr; pOperator->pTaskInfo = pTaskInfo; pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doApplyIndefinitFunction, NULL, NULL, @@ -3973,8 +3848,8 @@ SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode* pOperator->blocking = false; pOperator->status = OP_NOT_OPENED; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_FILL; - pOperator->pExpr = pExprInfo; - pOperator->numOfExprs = num; + pOperator->exprSupp.pExprInfo = pExprInfo; + pOperator->exprSupp.numOfExprs = num; pOperator->info = pInfo; pOperator->pTaskInfo = pTaskInfo; diff --git a/source/libs/executor/src/groupoperator.c b/source/libs/executor/src/groupoperator.c index c46c1dc3ef4536dff556e0f27d5020e1037ed430..d0d76d7da7aca760dde95731ba232b2a014432fd 100644 --- a/source/libs/executor/src/groupoperator.c +++ b/source/libs/executor/src/groupoperator.c @@ -28,8 +28,8 @@ static void* getCurrentDataGroupInfo(const SPartitionOperatorInfo* pInfo, SDataGroupInfo** pGroupInfo, int32_t len); static int32_t* setupColumnOffset(const SSDataBlock* pBlock, int32_t rowCapacity); -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 int32_t setGroupResultOutputBuf(SOperatorInfo* pOperator, SOptrBasicInfo* binfo, int32_t numOfCols, char* pData, int16_t bytes, + int32_t groupId, SDiskbasedBuf* pBuf, SAggSupporter* pAggSup); static void destroyGroupOperatorInfo(void* param, int32_t numOfOutput) { SGroupbyOperatorInfo* pInfo = (SGroupbyOperatorInfo*)param; @@ -217,7 +217,7 @@ static void doHashGroupbyAgg(SOperatorInfo* pOperator, SSDataBlock* pBlock) { SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SGroupbyOperatorInfo* pInfo = pOperator->info; - SqlFunctionCtx* pCtx = pInfo->binfo.pCtx; + SqlFunctionCtx* pCtx = pOperator->exprSupp.pCtx; int32_t numOfGroupCols = taosArrayGetSize(pInfo->pGroupCols); // if (type == TSDB_DATA_TYPE_FLOAT || type == TSDB_DATA_TYPE_DOUBLE) { // qError("QInfo:0x%"PRIx64" group by not supported on double/float columns, abort", GET_TASKID(pRuntimeEnv)); @@ -251,16 +251,16 @@ static void doHashGroupbyAgg(SOperatorInfo* pOperator, SSDataBlock* pBlock) { } len = buildGroupKeys(pInfo->keyBuf, pInfo->pGroupColVals); - int32_t ret = setGroupResultOutputBuf(&(pInfo->binfo), pOperator->numOfExprs, pInfo->keyBuf, TSDB_DATA_TYPE_VARCHAR, len, 0, pInfo->aggSup.pResultBuf, pTaskInfo, &pInfo->aggSup); + int32_t ret = setGroupResultOutputBuf(pOperator, &(pInfo->binfo), pOperator->exprSupp.numOfExprs, pInfo->keyBuf, len, 0, pInfo->aggSup.pResultBuf, &pInfo->aggSup); if (ret != TSDB_CODE_SUCCESS) { // null data, too many state code longjmp(pTaskInfo->env, TSDB_CODE_QRY_APP_ERROR); } int32_t rowIndex = j - num; - doApplyFunctions(pTaskInfo, pCtx, &w, NULL, rowIndex, num, NULL, pBlock->info.rows, pOperator->numOfExprs, TSDB_ORDER_ASC); + doApplyFunctions(pTaskInfo, pCtx, &w, NULL, rowIndex, num, NULL, pBlock->info.rows, pOperator->exprSupp.numOfExprs, TSDB_ORDER_ASC); // assign the group keys or user input constant values if required - doAssignGroupKeys(pCtx, pOperator->numOfExprs, pBlock->info.rows, rowIndex); + doAssignGroupKeys(pCtx, pOperator->exprSupp.numOfExprs, pBlock->info.rows, rowIndex); recordNewGroupKeys(pInfo->pGroupCols, pInfo->pGroupColVals, pBlock, j); num = 1; } @@ -268,15 +268,15 @@ static void doHashGroupbyAgg(SOperatorInfo* pOperator, SSDataBlock* pBlock) { if (num > 0) { len = buildGroupKeys(pInfo->keyBuf, pInfo->pGroupColVals); int32_t ret = - setGroupResultOutputBuf(&(pInfo->binfo), pOperator->numOfExprs, pInfo->keyBuf, TSDB_DATA_TYPE_VARCHAR, len, - 0, pInfo->aggSup.pResultBuf, pTaskInfo, &pInfo->aggSup); + setGroupResultOutputBuf(pOperator, &(pInfo->binfo), pOperator->exprSupp.numOfExprs, pInfo->keyBuf, len, + 0, pInfo->aggSup.pResultBuf, &pInfo->aggSup); if (ret != TSDB_CODE_SUCCESS) { longjmp(pTaskInfo->env, TSDB_CODE_QRY_APP_ERROR); } int32_t rowIndex = pBlock->info.rows - num; - doApplyFunctions(pTaskInfo, pCtx, &w, NULL, rowIndex, num, NULL, pBlock->info.rows, pOperator->numOfExprs, TSDB_ORDER_ASC); - doAssignGroupKeys(pCtx, pOperator->numOfExprs, pBlock->info.rows, rowIndex); + doApplyFunctions(pTaskInfo, pCtx, &w, NULL, rowIndex, num, NULL, pBlock->info.rows, pOperator->exprSupp.numOfExprs, TSDB_ORDER_ASC); + doAssignGroupKeys(pCtx, pOperator->exprSupp.numOfExprs, pBlock->info.rows, rowIndex); } } @@ -320,7 +320,7 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator) { } // the pDataBlock are always the same one, no need to call this again - setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, order, scanFlag, true); + setInputDataBlock(pOperator, pOperator->exprSupp.pCtx, pBlock, order, scanFlag, true); // there is an scalar expression that needs to be calculated right before apply the group aggregation. if (pInfo->scalarSup.pExprInfo != NULL) { @@ -397,15 +397,15 @@ SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pEx } initResultSizeInfo(pOperator, 4096); - initAggInfo(&pInfo->binfo, &pInfo->aggSup, pExprInfo, numOfCols, pResultBlock, pInfo->groupKeyLen, pTaskInfo->id.str); + initAggInfo(&pInfo->binfo, &pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, pResultBlock, pInfo->groupKeyLen, pTaskInfo->id.str); initResultRowInfo(&pInfo->binfo.resultRowInfo); pOperator->name = "GroupbyAggOperator"; pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; // pOperator->operatorType = OP_Groupby; - pOperator->pExpr = pExprInfo; - pOperator->numOfExprs = numOfCols; + pOperator->exprSupp.pExprInfo = pExprInfo; + pOperator->exprSupp.numOfExprs = numOfCols; pOperator->info = pInfo; pOperator->pTaskInfo = pTaskInfo; @@ -442,9 +442,9 @@ static void doHashPartition(SOperatorInfo* pOperator, SSDataBlock* pBlock) { // group id - size_t numOfCols = pOperator->numOfExprs; + size_t numOfCols = pOperator->exprSupp.numOfExprs; for(int32_t i = 0; i < numOfCols; ++i) { - SExprInfo* pExpr = &pOperator->pExpr[i]; + SExprInfo* pExpr = &pOperator->exprSupp.pExprInfo[i]; int32_t slotId = pExpr->base.pParam[0].pCol->slotId; SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, slotId); @@ -729,8 +729,8 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition pOperator->status = OP_NOT_OPENED; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_PARTITION; pInfo->binfo.pRes = pResBlock; - pOperator->numOfExprs = numOfCols; - pOperator->pExpr = pExprInfo; + pOperator->exprSupp.numOfExprs = numOfCols; + pOperator->exprSupp.pExprInfo = pExprInfo; pOperator->info = pInfo; pOperator->pTaskInfo = pTaskInfo; @@ -747,16 +747,16 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition 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) { +int32_t setGroupResultOutputBuf(SOperatorInfo* pOperator, SOptrBasicInfo* binfo, int32_t numOfCols, char* pData, int16_t bytes, + int32_t groupId, SDiskbasedBuf* pBuf, SAggSupporter* pAggSup) { + SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SResultRowInfo* pResultRowInfo = &binfo->resultRowInfo; - SqlFunctionCtx* pCtx = binfo->pCtx; + SqlFunctionCtx* pCtx = pOperator->exprSupp.pCtx; SResultRow* pResultRow = doSetResultOutBufByKey(pBuf, pResultRowInfo, (char*)pData, bytes, true, groupId, pTaskInfo, false, pAggSup); assert(pResultRow != NULL); - setResultRowInitCtx(pResultRow, pCtx, numOfCols, binfo->rowEntryInfoOffset); + setResultRowInitCtx(pResultRow, pCtx, numOfCols, pOperator->exprSupp.rowEntryInfoOffset); 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 6ac3f1a16cff899dfb354018590a4646a67ed646..6fbda778087b345c7aeef19e068f582451eab006 100644 --- a/source/libs/executor/src/joinoperator.c +++ b/source/libs/executor/src/joinoperator.c @@ -48,8 +48,8 @@ SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN; pOperator->blocking = false; pOperator->status = OP_NOT_OPENED; - pOperator->pExpr = pExprInfo; - pOperator->numOfExprs = numOfCols; + pOperator->exprSupp.pExprInfo = pExprInfo; + pOperator->exprSupp.numOfExprs = numOfCols; pOperator->info = pInfo; pOperator->pTaskInfo = pTaskInfo; @@ -132,10 +132,10 @@ SSDataBlock* doMergeJoin(struct SOperatorInfo* pOperator) { // only the timestamp match support for ordinary table ASSERT(pLeftCol->info.type == TSDB_DATA_TYPE_TIMESTAMP); if (*(int64_t*)pLeftVal == *(int64_t*)pRightVal) { - for (int32_t i = 0; i < pOperator->numOfExprs; ++i) { + for (int32_t i = 0; i < pOperator->exprSupp.numOfExprs; ++i) { SColumnInfoData* pDst = taosArrayGet(pRes->pDataBlock, i); - SExprInfo* pExprInfo = &pOperator->pExpr[i]; + SExprInfo* pExprInfo = &pOperator->exprSupp.pExprInfo[i]; int32_t blockId = pExprInfo->base.pParam[0].pCol->dataBlockId; int32_t slotId = pExprInfo->base.pParam[0].pCol->slotId; diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 73d6421ca3d16c45ce55688155a7af3e53501520..812272c5596c727255a8006378ac055b516b198b 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -262,7 +262,7 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableSca // currently only the tbname pseudo column if (pTableScanInfo->pseudoSup.numOfExprs > 0) { - SExecSupp* pSup = &pTableScanInfo->pseudoSup; + SExprSupp* pSup = &pTableScanInfo->pseudoSup; addTagPseudoColumnData(&pTableScanInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pBlock); } @@ -538,7 +538,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, } if (pTableScanNode->scan.pScanPseudoCols != NULL) { - SExecSupp* pSup = &pInfo->pseudoSup; + SExprSupp* pSup = &pInfo->pseudoSup; pSup->pExprInfo = createExprInfo(pTableScanNode->scan.pScanPseudoCols, NULL, &pSup->numOfExprs); pSup->pCtx = createSqlFunctionCtx(pSup->pExprInfo, pSup->numOfExprs, &pSup->rowEntryInfoOffset); } @@ -564,7 +564,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, pOperator->blocking = false; pOperator->status = OP_NOT_OPENED; pOperator->info = pInfo; - pOperator->numOfExprs = numOfCols; + pOperator->exprSupp.numOfExprs = numOfCols; pOperator->pTaskInfo = pTaskInfo; pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableScan, NULL, NULL, destroyTableScanOperatorInfo, @@ -1118,7 +1118,7 @@ SOperatorInfo* createStreamScanOperatorInfo(void* pDataReader, SReadHandle* pHan pOperator->blocking = false; pOperator->status = OP_NOT_OPENED; pOperator->info = pInfo; - pOperator->numOfExprs = pInfo->pRes->info.numOfCols; + pOperator->exprSupp.numOfExprs = pInfo->pRes->info.numOfCols; pOperator->pTaskInfo = pTaskInfo; pOperator->fpSet = @@ -1513,7 +1513,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) { } extractDataBlockFromFetchRsp(pInfo->pRes, &pInfo->loadInfo, pRsp->numOfRows, pRsp->data, pRsp->compLen, - pOperator->numOfExprs, startTs, NULL, pInfo->scanCols); + pOperator->exprSupp.numOfExprs, startTs, NULL, pInfo->scanCols); // todo log the filter info doFilterResult(pInfo); @@ -1628,7 +1628,7 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScan pOperator->blocking = false; pOperator->status = OP_NOT_OPENED; pOperator->info = pInfo; - pOperator->numOfExprs = pResBlock->info.numOfCols; + pOperator->exprSupp.numOfExprs = pResBlock->info.numOfCols; pOperator->pTaskInfo = pTaskInfo; pOperator->fpSet = @@ -1659,11 +1659,11 @@ static SSDataBlock* doTagScan(SOperatorInfo* pOperator) { int32_t count = 0; SArray* pa = GET_TABLEGROUP(pRuntimeEnv, 0); - int32_t functionId = getExprFunctionId(&pOperator->pExpr[0]); + int32_t functionId = getExprFunctionId(&pOperator->exprSupp.pExprInfo[0]); if (functionId == FUNCTION_TID_TAG) { // return the tags & table Id assert(pQueryAttr->numOfOutput == 1); - SExprInfo* pExprInfo = &pOperator->pExpr[0]; + SExprInfo* pExprInfo = &pOperator->exprSupp.pExprInfo[0]; int32_t rsize = pExprInfo->base.resSchema.bytes; count = 0; @@ -1726,7 +1726,7 @@ static SSDataBlock* doTagScan(SOperatorInfo* pOperator) { #endif STagScanInfo* pInfo = pOperator->info; - SExprInfo* pExprInfo = &pOperator->pExpr[0]; + SExprInfo* pExprInfo = &pOperator->exprSupp.pExprInfo[0]; SSDataBlock* pRes = pInfo->pRes; int32_t size = taosArrayGetSize(pInfo->pTableList->pTableList); @@ -1744,7 +1744,7 @@ static SSDataBlock* doTagScan(SOperatorInfo* pOperator) { STableKeyInfo* item = taosArrayGet(pInfo->pTableList->pTableList, pInfo->curPos); metaGetTableEntryByUid(&mr, item->uid); - for (int32_t j = 0; j < pOperator->numOfExprs; ++j) { + for (int32_t j = 0; j < pOperator->exprSupp.numOfExprs; ++j) { SColumnInfoData* pDst = taosArrayGet(pRes->pDataBlock, pExprInfo[j].base.resSchema.slotId); // refactor later @@ -1824,8 +1824,8 @@ SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysi pOperator->blocking = false; pOperator->status = OP_NOT_OPENED; pOperator->info = pInfo; - pOperator->pExpr = pExprInfo; - pOperator->numOfExprs = numOfExprs; + pOperator->exprSupp.pExprInfo = pExprInfo; + pOperator->exprSupp.numOfExprs = numOfExprs; pOperator->pTaskInfo = pTaskInfo; initResultSizeInfo(pOperator, 4096); @@ -2304,7 +2304,7 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN pOperator->blocking = false; pOperator->status = OP_NOT_OPENED; pOperator->info = pInfo; - pOperator->numOfExprs = numOfCols; + pOperator->exprSupp.numOfExprs = numOfCols; pOperator->pTaskInfo = pTaskInfo; initResultSizeInfo(pOperator, 1024); diff --git a/source/libs/executor/src/sortoperator.c b/source/libs/executor/src/sortoperator.c index a16bceabfc7cb1a9226b2938edbf60562e590ebb..9d13276e6d68d2aaf380d0ac3111c2214c6f0e75 100644 --- a/source/libs/executor/src/sortoperator.c +++ b/source/libs/executor/src/sortoperator.c @@ -39,7 +39,7 @@ SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* SArray* pColMatchColInfo = extractColMatchInfo(pSortPhyNode->pTargets, pDescNode, &numOfOutputCols, COL_MATCH_FROM_SLOT_ID); - pInfo->binfo.pCtx = createSqlFunctionCtx(pExprInfo, numOfCols, &pInfo->binfo.rowEntryInfoOffset); + pOperator->exprSupp.pCtx = createSqlFunctionCtx(pExprInfo, numOfCols, &pOperator->exprSupp.rowEntryInfoOffset); pInfo->binfo.pRes = pResBlock; initResultSizeInfo(pOperator, 1024); @@ -51,8 +51,8 @@ SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; pOperator->info = pInfo; - pOperator->pExpr = pExprInfo; - pOperator->numOfExprs = numOfCols; + pOperator->exprSupp.pExprInfo = pExprInfo; + pOperator->exprSupp.numOfExprs = numOfCols; pOperator->pTaskInfo = pTaskInfo; // lazy evaluation for the following parameter since the input datablock is not known till now. @@ -145,9 +145,9 @@ SSDataBlock* loadNextDataBlock(void* param) { void applyScalarFunction(SSDataBlock* pBlock, void* param) { SOperatorInfo* pOperator = param; SSortOperatorInfo* pSort = pOperator->info; - if (pOperator->pExpr != NULL) { + if (pOperator->exprSupp.pExprInfo != NULL) { int32_t code = - projectApplyFunctions(pOperator->pExpr, pBlock, pBlock, pSort->binfo.pCtx, pOperator->numOfExprs, NULL); + projectApplyFunctions(pOperator->exprSupp.pExprInfo, pBlock, pBlock, pOperator->exprSupp.pCtx, pOperator->exprSupp.numOfExprs, NULL); if (code != TSDB_CODE_SUCCESS) { longjmp(pOperator->pTaskInfo->env, code); } diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c index 0c269b11fb177f25f7f57e7e2617b2eccb3bc88f..e580bf6304ea16bb5084ac626a497badd2bf6983 100644 --- a/source/libs/executor/src/timewindowoperator.c +++ b/source/libs/executor/src/timewindowoperator.c @@ -322,12 +322,12 @@ static void getNextTimeWindow(SInterval* pInterval, int32_t precision, int32_t o tw->ekey -= 1; } -void doTimeWindowInterpolation(SIntervalAggOperatorInfo* pInfo, int32_t numOfExprs, SArray* pDataBlock, TSKEY prevTs, - int32_t prevRowIndex, TSKEY curTs, int32_t curRowIndex, TSKEY windowKey, int32_t type) { - SqlFunctionCtx* pCtx = pInfo->binfo.pCtx; +void doTimeWindowInterpolation(SArray* pPrevValues, SArray* pDataBlock, TSKEY prevTs, int32_t prevRowIndex, + TSKEY curTs, int32_t curRowIndex, TSKEY windowKey, int32_t type, SExprSupp* pSup) { + SqlFunctionCtx* pCtx = pSup->pCtx; int32_t index = 1; - for (int32_t k = 0; k < numOfExprs; ++k) { + for (int32_t k = 0; k < pSup->numOfExprs; ++k) { if (!fmIsIntervalInterpoFunc(pCtx[k].functionId)) { pCtx[k].start.key = INT64_MIN; continue; @@ -340,7 +340,7 @@ void doTimeWindowInterpolation(SIntervalAggOperatorInfo* pInfo, int32_t numOfExp double v1 = 0, v2 = 0, v = 0; if (prevRowIndex == -1) { - SGroupKeys* p = taosArrayGet(pInfo->pPrevValues, index); + SGroupKeys* p = taosArrayGet(pPrevValues, index); GET_TYPED_DATA(v1, double, pColInfo->info.type, p->pData); } else { GET_TYPED_DATA(v1, double, pColInfo->info.type, colDataGetData(pColInfo, prevRowIndex)); @@ -403,8 +403,8 @@ static void setNotInterpoWindowKey(SqlFunctionCtx* pCtx, int32_t numOfOutput, in } } -static bool setTimeWindowInterpolationStartTs(SIntervalAggOperatorInfo* pInfo, SqlFunctionCtx* pCtx, int32_t numOfExprs, - int32_t pos, SSDataBlock* pBlock, const TSKEY* tsCols, STimeWindow* win) { +static bool setTimeWindowInterpolationStartTs(SIntervalAggOperatorInfo* pInfo, int32_t pos, SSDataBlock* pBlock, const TSKEY* tsCols, + STimeWindow* win, SExprSupp* pSup) { bool ascQuery = (pInfo->order == TSDB_ORDER_ASC); TSKEY curTs = tsCols[pos]; @@ -416,23 +416,23 @@ static bool setTimeWindowInterpolationStartTs(SIntervalAggOperatorInfo* pInfo, S // start exactly from this point, no need to do interpolation TSKEY key = ascQuery ? win->skey : win->ekey; if (key == curTs) { - setNotInterpoWindowKey(pCtx, numOfExprs, RESULT_ROW_START_INTERP); + setNotInterpoWindowKey(pSup->pCtx, pSup->numOfExprs, RESULT_ROW_START_INTERP); return true; } // it is the first time window, no need to do interpolation if (pTsKey->isNull && pos == 0) { - setNotInterpoWindowKey(pCtx, numOfExprs, RESULT_ROW_START_INTERP); + setNotInterpoWindowKey(pSup->pCtx, pSup->numOfExprs, RESULT_ROW_START_INTERP); } else { TSKEY prevTs = ((pos == 0) ? lastTs : tsCols[pos - 1]); - doTimeWindowInterpolation(pInfo, numOfExprs, pBlock->pDataBlock, prevTs, pos - 1, curTs, pos, key, - RESULT_ROW_START_INTERP); + doTimeWindowInterpolation(pInfo->pPrevValues, pBlock->pDataBlock, prevTs, pos - 1, curTs, pos, key, + RESULT_ROW_START_INTERP, pSup); } return true; } -static bool setTimeWindowInterpolationEndTs(SIntervalAggOperatorInfo* pInfo, SqlFunctionCtx* pCtx, int32_t numOfExprs, +static bool setTimeWindowInterpolationEndTs(SIntervalAggOperatorInfo* pInfo, SExprSupp* pSup, int32_t endRowIndex, SArray* pDataBlock, const TSKEY* tsCols, TSKEY blockEkey, STimeWindow* win) { int32_t order = pInfo->order; @@ -442,13 +442,13 @@ static bool setTimeWindowInterpolationEndTs(SIntervalAggOperatorInfo* pInfo, Sql // not ended in current data block, do not invoke interpolation if ((key > blockEkey && (order == TSDB_ORDER_ASC)) || (key < blockEkey && (order == TSDB_ORDER_DESC))) { - setNotInterpoWindowKey(pCtx, numOfExprs, RESULT_ROW_END_INTERP); + setNotInterpoWindowKey(pSup->pCtx, pSup->numOfExprs, RESULT_ROW_END_INTERP); return false; } // there is actual end point of current time window, no interpolation needs if (key == actualEndKey) { - setNotInterpoWindowKey(pCtx, numOfExprs, RESULT_ROW_END_INTERP); + setNotInterpoWindowKey(pSup->pCtx, pSup->numOfExprs, RESULT_ROW_END_INTERP); return true; } @@ -456,8 +456,8 @@ static bool setTimeWindowInterpolationEndTs(SIntervalAggOperatorInfo* pInfo, Sql assert(nextRowIndex >= 0); TSKEY nextKey = tsCols[nextRowIndex]; - doTimeWindowInterpolation(pInfo, numOfExprs, pDataBlock, actualEndKey, endRowIndex, nextKey, nextRowIndex, key, - RESULT_ROW_END_INTERP); + doTimeWindowInterpolation(pInfo->pPrevValues, pDataBlock, actualEndKey, endRowIndex, nextKey, nextRowIndex, key, + RESULT_ROW_END_INTERP, pSup); return true; } @@ -546,9 +546,8 @@ static void setResultRowInterpo(SResultRow* pResult, SResultTsInterpType type) { } } -static void doWindowBorderInterpolation(SIntervalAggOperatorInfo* pInfo, SSDataBlock* pBlock, int32_t numOfExprs, - SqlFunctionCtx* pCtx, SResultRow* pResult, STimeWindow* win, int32_t startPos, - int32_t forwardRows) { +static void doWindowBorderInterpolation(SIntervalAggOperatorInfo* pInfo, SSDataBlock* pBlock, SResultRow* pResult, STimeWindow* win, int32_t startPos, + int32_t forwardRows, SExprSupp* pSup) { if (!pInfo->timeWindowInterpo) { return; } @@ -564,12 +563,12 @@ static void doWindowBorderInterpolation(SIntervalAggOperatorInfo* pInfo, SSDataB TSKEY* tsCols = (TSKEY*)(pColInfo->pData); bool done = isResultRowInterpolated(pResult, RESULT_ROW_START_INTERP); if (!done) { // it is not interpolated, now start to generated the interpolated value - bool interp = setTimeWindowInterpolationStartTs(pInfo, pCtx, numOfExprs, startPos, pBlock, tsCols, win); + bool interp = setTimeWindowInterpolationStartTs(pInfo, startPos, pBlock, tsCols, win, pSup); if (interp) { setResultRowInterpo(pResult, RESULT_ROW_START_INTERP); } } else { - setNotInterpoWindowKey(pCtx, numOfExprs, RESULT_ROW_START_INTERP); + setNotInterpoWindowKey(pSup->pCtx, pSup->numOfExprs, RESULT_ROW_START_INTERP); } // point interpolation does not require the end key time window interpolation. @@ -583,13 +582,12 @@ static void doWindowBorderInterpolation(SIntervalAggOperatorInfo* pInfo, SSDataB int32_t endRowIndex = startPos + forwardRows - 1; TSKEY endKey = (pInfo->order == TSDB_ORDER_ASC) ? pBlock->info.window.ekey : pBlock->info.window.skey; - bool interp = - setTimeWindowInterpolationEndTs(pInfo, pCtx, numOfExprs, endRowIndex, pBlock->pDataBlock, tsCols, endKey, win); + bool interp = setTimeWindowInterpolationEndTs(pInfo, pSup, endRowIndex, pBlock->pDataBlock, tsCols, endKey, win); if (interp) { setResultRowInterpo(pResult, RESULT_ROW_END_INTERP); } } else { - setNotInterpoWindowKey(pCtx, numOfExprs, RESULT_ROW_END_INTERP); + setNotInterpoWindowKey(pSup->pCtx, pSup->numOfExprs, RESULT_ROW_END_INTERP); } } @@ -628,9 +626,10 @@ static void doInterpUnclosedTimeWindow(SOperatorInfo* pOperatorInfo, int32_t num SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo; SIntervalAggOperatorInfo* pInfo = (SIntervalAggOperatorInfo*)pOperatorInfo->info; + SExprSupp* pSup = &pOperatorInfo->exprSupp; int32_t startPos = 0; - int32_t numOfOutput = pOperatorInfo->numOfExprs; + int32_t numOfOutput = pSup->numOfExprs; uint64_t groupId = pBlock->info.groupId; SResultRow* pResult = NULL; @@ -655,8 +654,8 @@ static void doInterpUnclosedTimeWindow(SOperatorInfo* pOperatorInfo, int32_t num STimeWindow w = pr->win; int32_t ret = - setTimeWindowOutputBuf(pResultRowInfo, &w, (scanFlag == MAIN_SCAN), &pResult, groupId, pInfo->binfo.pCtx, - numOfOutput, pInfo->binfo.rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo); + setTimeWindowOutputBuf(pResultRowInfo, &w, (scanFlag == MAIN_SCAN), &pResult, groupId, pSup->pCtx, + numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo); if (ret != TSDB_CODE_SUCCESS) { longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY); } @@ -665,13 +664,13 @@ static void doInterpUnclosedTimeWindow(SOperatorInfo* pOperatorInfo, int32_t num SGroupKeys* pTsKey = taosArrayGet(pInfo->pPrevValues, 0); int64_t prevTs = *(int64_t*)pTsKey->pData; - doTimeWindowInterpolation(pInfo, numOfOutput, pBlock->pDataBlock, prevTs, -1, tsCols[startPos], startPos, w.ekey, - RESULT_ROW_END_INTERP); + doTimeWindowInterpolation(pInfo->pPrevValues, pBlock->pDataBlock, prevTs, -1, tsCols[startPos], startPos, w.ekey, + RESULT_ROW_END_INTERP, pSup); setResultRowInterpo(pResult, RESULT_ROW_END_INTERP); - setNotInterpoWindowKey(pInfo->binfo.pCtx, numOfExprs, RESULT_ROW_START_INTERP); + setNotInterpoWindowKey(pSup->pCtx, numOfExprs, RESULT_ROW_START_INTERP); - doApplyFunctions(pTaskInfo, pInfo->binfo.pCtx, &w, &pInfo->twAggSup.timeWindowData, startPos, 0, tsCols, + doApplyFunctions(pTaskInfo, pSup->pCtx, &w, &pInfo->twAggSup.timeWindowData, startPos, 0, tsCols, pBlock->info.rows, numOfExprs, pInfo->order); if (isResultRowInterpolated(pResult, RESULT_ROW_END_INTERP)) { @@ -798,9 +797,10 @@ static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResul SIntervalAggOperatorInfo* pInfo = (SIntervalAggOperatorInfo*)pOperatorInfo->info; SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo; + SExprSupp* pSup = &pOperatorInfo->exprSupp; int32_t startPos = 0; - int32_t numOfOutput = pOperatorInfo->numOfExprs; + int32_t numOfOutput = pSup->numOfExprs; int64_t* tsCols = extractTsCol(pBlock, pInfo); uint64_t tableGroupId = pBlock->info.groupId; bool ascScan = (pInfo->order == TSDB_ORDER_ASC); @@ -811,8 +811,8 @@ static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResul pInfo->interval.precision, &pInfo->win); int32_t ret = - setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, pInfo->binfo.pCtx, - numOfOutput, pInfo->binfo.rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo); + setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, pSup->pCtx, + numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo); if (ret != TSDB_CODE_SUCCESS || pResult == NULL) { longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY); } @@ -835,18 +835,18 @@ static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResul // restore current time window ret = - setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, pInfo->binfo.pCtx, - numOfOutput, pInfo->binfo.rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo); + setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, pSup->pCtx, + numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo); if (ret != TSDB_CODE_SUCCESS) { longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY); } // window start key interpolation - doWindowBorderInterpolation(pInfo, pBlock, numOfOutput, pInfo->binfo.pCtx, pResult, &win, startPos, forwardRows); + doWindowBorderInterpolation(pInfo, pBlock, pResult, &win, startPos, forwardRows, pSup); } updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &win, true); - doApplyFunctions(pTaskInfo, pInfo->binfo.pCtx, &win, &pInfo->twAggSup.timeWindowData, startPos, forwardRows, tsCols, + doApplyFunctions(pTaskInfo, pSup->pCtx, &win, &pInfo->twAggSup.timeWindowData, startPos, forwardRows, tsCols, pBlock->info.rows, numOfOutput, pInfo->order); doCloseWindow(pResultRowInfo, pInfo, pResult); @@ -861,7 +861,7 @@ static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResul // null data, failed to allocate more memory buffer int32_t code = setTimeWindowOutputBuf(pResultRowInfo, &nextWin, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, - pInfo->binfo.pCtx, numOfOutput, pInfo->binfo.rowEntryInfoOffset, + pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo); if (code != TSDB_CODE_SUCCESS || pResult == NULL) { longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY); @@ -878,11 +878,10 @@ static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResul getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, pInfo->order); // window start(end) key interpolation - doWindowBorderInterpolation(pInfo, pBlock, numOfOutput, pInfo->binfo.pCtx, pResult, &nextWin, startPos, - forwardRows); + doWindowBorderInterpolation(pInfo, pBlock, pResult, &nextWin, startPos, forwardRows, pSup); updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &nextWin, true); - doApplyFunctions(pTaskInfo, pInfo->binfo.pCtx, &nextWin, &pInfo->twAggSup.timeWindowData, startPos, forwardRows, + doApplyFunctions(pTaskInfo, pSup->pCtx, &nextWin, &pInfo->twAggSup.timeWindowData, startPos, forwardRows, tsCols, pBlock->info.rows, numOfOutput, pInfo->order); doCloseWindow(pResultRowInfo, pInfo, pResult); } @@ -937,6 +936,7 @@ static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) { SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SIntervalAggOperatorInfo* pInfo = pOperator->info; + SExprSupp* pSup = &pOperator->exprSupp; int32_t scanFlag = MAIN_SCAN; @@ -952,7 +952,7 @@ static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) { getTableScanInfo(pOperator, &pInfo->order, &scanFlag); // the pDataBlock are always the same one, no need to call this again - setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, pInfo->order, scanFlag, true); + setInputDataBlock(pOperator, pSup->pCtx, pBlock, pInfo->order, scanFlag, true); hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, scanFlag, NULL); #if 0 // test for encode/decode result info @@ -993,12 +993,13 @@ static bool compareVal(const char* v, const SStateKeys* pKey) { static void doStateWindowAggImpl(SOperatorInfo* pOperator, SStateWindowOperatorInfo* pInfo, SSDataBlock* pBlock) { SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; + SExprSupp* pSup = &pOperator->exprSupp; SColumnInfoData* pStateColInfoData = taosArrayGet(pBlock->pDataBlock, pInfo->stateCol.slotId); int64_t gid = pBlock->info.groupId; bool masterScan = true; - int32_t numOfOutput = pOperator->numOfExprs; + int32_t numOfOutput = pOperator->exprSupp.numOfExprs; int16_t bytes = pStateColInfoData->info.bytes; SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pInfo->tsSlotId); @@ -1041,14 +1042,14 @@ static void doStateWindowAggImpl(SOperatorInfo* pOperator, SStateWindowOperatorI pRowSup->win.ekey = pRowSup->win.skey; int32_t ret = - setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &window, masterScan, &pResult, gid, pInfo->binfo.pCtx, - numOfOutput, pInfo->binfo.rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo); + setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &window, masterScan, &pResult, gid, pSup->pCtx, + numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo); if (ret != TSDB_CODE_SUCCESS) { // null data, too many state code longjmp(pTaskInfo->env, TSDB_CODE_QRY_APP_ERROR); } updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &window, false); - doApplyFunctions(pTaskInfo, pInfo->binfo.pCtx, &window, &pInfo->twAggSup.timeWindowData, pRowSup->startRowIndex, + doApplyFunctions(pTaskInfo, pSup->pCtx, &window, &pInfo->twAggSup.timeWindowData, pRowSup->startRowIndex, pRowSup->numOfRows, NULL, pBlock->info.rows, numOfOutput, TSDB_ORDER_ASC); // here we start a new session window @@ -1067,14 +1068,14 @@ static void doStateWindowAggImpl(SOperatorInfo* pOperator, SStateWindowOperatorI SResultRow* pResult = NULL; pRowSup->win.ekey = tsList[pBlock->info.rows - 1]; int32_t ret = - setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &pRowSup->win, masterScan, &pResult, gid, pInfo->binfo.pCtx, - numOfOutput, pInfo->binfo.rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo); + setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &pRowSup->win, masterScan, &pResult, gid, pSup->pCtx, + numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo); if (ret != TSDB_CODE_SUCCESS) { // null data, too many state code longjmp(pTaskInfo->env, TSDB_CODE_QRY_APP_ERROR); } updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pRowSup->win, false); - doApplyFunctions(pTaskInfo, pInfo->binfo.pCtx, &pRowSup->win, &pInfo->twAggSup.timeWindowData, pRowSup->startRowIndex, + doApplyFunctions(pTaskInfo, pSup->pCtx, &pRowSup->win, &pInfo->twAggSup.timeWindowData, pRowSup->startRowIndex, pRowSup->numOfRows, NULL, pBlock->info.rows, numOfOutput, TSDB_ORDER_ASC); } @@ -1086,6 +1087,8 @@ static SSDataBlock* doStateWindowAgg(SOperatorInfo* pOperator) { SStateWindowOperatorInfo* pInfo = pOperator->info; SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; + SExprSupp* pSup = &pOperator->exprSupp; + SOptrBasicInfo* pBInfo = &pInfo->binfo; if (pOperator->status == OP_RES_TO_RETURN) { @@ -1108,7 +1111,7 @@ static SSDataBlock* doStateWindowAgg(SOperatorInfo* pOperator) { break; } - setInputDataBlock(pOperator, pBInfo->pCtx, pBlock, order, MAIN_SCAN, true); + setInputDataBlock(pOperator, pSup->pCtx, pBlock, order, MAIN_SCAN, true); blockDataUpdateTsWindow(pBlock, pInfo->tsSlotId); doStateWindowAggImpl(pOperator, pInfo, pBlock); @@ -1195,11 +1198,11 @@ static void setInverFunction(SqlFunctionCtx* pCtx, int32_t num, EStreamType type } } -void doClearWindowImpl(SResultRowPosition* p1, SDiskbasedBuf* pResultBuf, SOptrBasicInfo* pBinfo, int32_t numOfOutput) { +void doClearWindowImpl(SResultRowPosition* p1, SDiskbasedBuf* pResultBuf, SExprSupp *pSup, int32_t numOfOutput) { SResultRow* pResult = getResultRowByPos(pResultBuf, p1); - SqlFunctionCtx* pCtx = pBinfo->pCtx; + SqlFunctionCtx* pCtx = pSup->pCtx; for (int32_t i = 0; i < numOfOutput; ++i) { - pCtx[i].resultInfo = getResultEntryInfo(pResult, i, pBinfo->rowEntryInfoOffset); + pCtx[i].resultInfo = getResultEntryInfo(pResult, i, pSup->rowEntryInfoOffset); struct SResultRowEntryInfo* pResInfo = pCtx[i].resultInfo; if (fmIsWindowPseudoColumnFunc(pCtx[i].functionId)) { continue; @@ -1211,19 +1214,19 @@ void doClearWindowImpl(SResultRowPosition* p1, SDiskbasedBuf* pResultBuf, SOptrB } } -void doClearWindow(SAggSupporter* pSup, SOptrBasicInfo* pBinfo, char* pData, int16_t bytes, uint64_t groupId, +void doClearWindow(SAggSupporter* pAggSup, SExprSupp *pSup, char* pData, int16_t bytes, uint64_t groupId, int32_t numOfOutput) { - SET_RES_WINDOW_KEY(pSup->keyBuf, pData, bytes, groupId); + SET_RES_WINDOW_KEY(pAggSup->keyBuf, pData, bytes, groupId); SResultRowPosition* p1 = - (SResultRowPosition*)taosHashGet(pSup->pResultRowHashTable, pSup->keyBuf, GET_RES_WINDOW_KEY_LEN(bytes)); + (SResultRowPosition*)taosHashGet(pAggSup->pResultRowHashTable, pAggSup->keyBuf, GET_RES_WINDOW_KEY_LEN(bytes)); if (!p1) { // window has been closed return; } - doClearWindowImpl(p1, pSup->pResultBuf, pBinfo, numOfOutput); + doClearWindowImpl(p1, pAggSup->pResultBuf, pSup, numOfOutput); } -static void doClearWindows(SAggSupporter* pSup, SOptrBasicInfo* pBinfo, SInterval* pInterval, int32_t tsIndex, +static void doClearWindows(SAggSupporter* pAggSup, SExprSupp* pSup1, SInterval* pInterval, int32_t tsIndex, int32_t numOfOutput, SSDataBlock* pBlock, SArray* pUpWins) { SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, tsIndex); TSKEY* tsCols = (TSKEY*)pColDataInfo->pData; @@ -1233,7 +1236,7 @@ static void doClearWindows(SAggSupporter* pSup, SOptrBasicInfo* pBinfo, SInterva dumyInfo.cur.pageId = -1; STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, tsCols[i], pInterval, pInterval->precision, NULL); step = getNumOfRowsInTimeWindow(&pBlock->info, tsCols, i, win.ekey, binarySearchForKey, NULL, TSDB_ORDER_ASC); - doClearWindow(pSup, pBinfo, (char*)&win.skey, sizeof(TKEY), pBlock->info.groupId, numOfOutput); + doClearWindow(pAggSup, pSup1, (char*)&win.skey, sizeof(TKEY), pBlock->info.groupId, numOfOutput); if (pUpWins) { taosArrayPush(pUpWins, &win); } @@ -1294,6 +1297,7 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) { SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; pInfo->order = TSDB_ORDER_ASC; + SExprSupp* pSup = &pOperator->exprSupp; if (pOperator->status == OP_EXEC_DONE) { return NULL; @@ -1317,7 +1321,7 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) { } if (pBlock->info.type == STREAM_REPROCESS) { - doClearWindows(&pInfo->aggSup, &pInfo->binfo, &pInfo->interval, 0, pOperator->numOfExprs, pBlock, NULL); + doClearWindows(&pInfo->aggSup, &pOperator->exprSupp, &pInfo->interval, 0, pOperator->exprSupp.numOfExprs, pBlock, NULL); qDebug("%s clear existed time window results for updates checked", GET_TASKID(pTaskInfo)); continue; } else if (pBlock->info.type == STREAM_GET_ALL) { @@ -1328,9 +1332,9 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) { // The timewindow that overlaps the timestamps of the input pBlock need to be recalculated and return to the // caller. Note that all the time window are not close till now. // the pDataBlock are always the same one, no need to call this again - setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, pInfo->order, MAIN_SCAN, true); + setInputDataBlock(pOperator, pSup->pCtx, pBlock, pInfo->order, MAIN_SCAN, true); if (pInfo->invertible) { - setInverFunction(pInfo->binfo.pCtx, pOperator->numOfExprs, pBlock->info.type); + setInverFunction(pSup->pCtx, pOperator->exprSupp.numOfExprs, pBlock->info.type); } pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.window.ekey); @@ -1338,7 +1342,7 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) { } closeIntervalWindow(pInfo->aggSup.pResultRowHashTable, &pInfo->twAggSup, &pInfo->interval, pUpdated); - finalizeUpdatedResult(pOperator->numOfExprs, pInfo->aggSup.pResultBuf, pUpdated, pInfo->binfo.rowEntryInfoOffset); + finalizeUpdatedResult(pOperator->exprSupp.numOfExprs, pInfo->aggSup.pResultBuf, pUpdated, pSup->rowEntryInfoOffset); initMultiResInfoFromArrayList(&pInfo->groupResInfo, pUpdated); blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity); doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf); @@ -1452,23 +1456,25 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pInfo->primaryTsIndex = primaryTsSlotId; + SExprSupp* pSup = &pOperator->exprSupp; + size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES; initResultSizeInfo(pOperator, 4096); int32_t code = - initAggInfo(&pInfo->binfo, &pInfo->aggSup, pExprInfo, numOfCols, pResBlock, keyBufSize, pTaskInfo->id.str); + initAggInfo(&pInfo->binfo, pSup, &pInfo->aggSup, pExprInfo, numOfCols, pResBlock, keyBufSize, pTaskInfo->id.str); if (isStream) { ASSERT(numOfCols > 0); - increaseTs(pInfo->binfo.pCtx); + increaseTs(pSup->pCtx); } initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pInfo->win); - pInfo->invertible = allInvertible(pInfo->binfo.pCtx, numOfCols); + pInfo->invertible = allInvertible(pSup->pCtx, numOfCols); pInfo->invertible = false; // Todo(liuyao): Dependent TSDB API - pInfo->timeWindowInterpo = timeWindowinterpNeeded(pInfo->binfo.pCtx, numOfCols, pInfo); + pInfo->timeWindowInterpo = timeWindowinterpNeeded(pSup->pCtx, numOfCols, pInfo); if (pInfo->timeWindowInterpo) { pInfo->binfo.resultRowInfo.openWindow = tdListNew(sizeof(SResultRowPosition)); if (pInfo->binfo.resultRowInfo.openWindow == NULL) { @@ -1482,9 +1488,9 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL; pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; - pOperator->pExpr = pExprInfo; + pOperator->exprSupp.pExprInfo = pExprInfo; pOperator->pTaskInfo = pTaskInfo; - pOperator->numOfExprs = numOfCols; + pOperator->exprSupp.numOfExprs = numOfCols; pOperator->info = pInfo; pOperator->fpSet = createOperatorFpSet(doOpenIntervalAgg, doBuildIntervalResult, doStreamIntervalAgg, NULL, @@ -1526,7 +1532,7 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SExpr initResultSizeInfo(pOperator, numOfRows); int32_t code = - initAggInfo(&pInfo->binfo, &pInfo->aggSup, pExprInfo, numOfCols, pResBlock, keyBufSize, pTaskInfo->id.str); + initAggInfo(&pInfo->binfo, &pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, pResBlock, keyBufSize, pTaskInfo->id.str); initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pInfo->win); if (code != TSDB_CODE_SUCCESS) { @@ -1539,9 +1545,9 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SExpr pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL; pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; - pOperator->pExpr = pExprInfo; + pOperator->exprSupp.pExprInfo = pExprInfo; pOperator->pTaskInfo = pTaskInfo; - pOperator->numOfExprs = numOfCols; + pOperator->exprSupp.numOfExprs = numOfCols; pOperator->info = pInfo; pOperator->fpSet = createOperatorFpSet(doOpenIntervalAgg, doStreamIntervalAgg, doStreamIntervalAgg, NULL, @@ -1565,11 +1571,12 @@ _error: // todo handle multiple tables cases. static void doSessionWindowAggImpl(SOperatorInfo* pOperator, SSessionAggOperatorInfo* pInfo, SSDataBlock* pBlock) { SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; + SExprSupp* pSup = &pOperator->exprSupp; SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pInfo->tsSlotId); bool masterScan = true; - int32_t numOfOutput = pOperator->numOfExprs; + int32_t numOfOutput = pOperator->exprSupp.numOfExprs; int64_t gid = pBlock->info.groupId; int64_t gap = pInfo->gap; @@ -1602,15 +1609,15 @@ static void doSessionWindowAggImpl(SOperatorInfo* pOperator, SSessionAggOperator pRowSup->win.ekey = pRowSup->win.skey; int32_t ret = - setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &window, masterScan, &pResult, gid, pInfo->binfo.pCtx, - numOfOutput, pInfo->binfo.rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo); + setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &window, masterScan, &pResult, gid, pSup->pCtx, + numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo); if (ret != TSDB_CODE_SUCCESS) { // null data, too many state code longjmp(pTaskInfo->env, TSDB_CODE_QRY_APP_ERROR); } // pInfo->numOfRows data belong to the current session window updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &window, false); - doApplyFunctions(pTaskInfo, pInfo->binfo.pCtx, &window, &pInfo->twAggSup.timeWindowData, pRowSup->startRowIndex, + doApplyFunctions(pTaskInfo, pSup->pCtx, &window, &pInfo->twAggSup.timeWindowData, pRowSup->startRowIndex, pRowSup->numOfRows, NULL, pBlock->info.rows, numOfOutput, TSDB_ORDER_ASC); // here we start a new session window @@ -1622,14 +1629,14 @@ static void doSessionWindowAggImpl(SOperatorInfo* pOperator, SSessionAggOperator SResultRow* pResult = NULL; pRowSup->win.ekey = tsList[pBlock->info.rows - 1]; int32_t ret = - setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &pRowSup->win, masterScan, &pResult, gid, pInfo->binfo.pCtx, - numOfOutput, pInfo->binfo.rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo); + setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &pRowSup->win, masterScan, &pResult, gid, pSup->pCtx, + numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo); if (ret != TSDB_CODE_SUCCESS) { // null data, too many state code longjmp(pTaskInfo->env, TSDB_CODE_QRY_APP_ERROR); } updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pRowSup->win, false); - doApplyFunctions(pTaskInfo, pInfo->binfo.pCtx, &pRowSup->win, &pInfo->twAggSup.timeWindowData, pRowSup->startRowIndex, + doApplyFunctions(pTaskInfo, pSup->pCtx, &pRowSup->win, &pInfo->twAggSup.timeWindowData, pRowSup->startRowIndex, pRowSup->numOfRows, NULL, pBlock->info.rows, numOfOutput, TSDB_ORDER_ASC); } @@ -1640,6 +1647,7 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) { SSessionAggOperatorInfo* pInfo = pOperator->info; SOptrBasicInfo* pBInfo = &pInfo->binfo; + SExprSupp* pSup = &pOperator->exprSupp; if (pOperator->status == OP_RES_TO_RETURN) { doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf); @@ -1663,7 +1671,7 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) { } // the pDataBlock are always the same one, no need to call this again - setInputDataBlock(pOperator, pBInfo->pCtx, pBlock, order, MAIN_SCAN, true); + setInputDataBlock(pOperator, pSup->pCtx, pBlock, order, MAIN_SCAN, true); blockDataUpdateTsWindow(pBlock, pInfo->tsSlotId); doSessionWindowAggImpl(pOperator, pInfo, pBlock); @@ -1711,6 +1719,7 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) { STimeSliceOperatorInfo* pSliceInfo = pOperator->info; SSDataBlock* pResBlock = pSliceInfo->binfo.pRes; + SExprSupp* pSup = &pOperator->exprSupp; // if (pOperator->status == OP_RES_TO_RETURN) { // // doBuildResultDatablock(&pRuntimeEnv->groupResInfo, pRuntimeEnv, pIntervalInfo->pRes); @@ -1733,15 +1742,15 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) { } // the pDataBlock are always the same one, no need to call this again - setInputDataBlock(pOperator, pSliceInfo->binfo.pCtx, pBlock, order, MAIN_SCAN, true); + setInputDataBlock(pOperator, pSup->pCtx, pBlock, order, MAIN_SCAN, true); SColumnInfoData* pTsCol = taosArrayGet(pBlock->pDataBlock, 0); for(int32_t i = 0; i < pBlock->info.rows; ++i) { int64_t ts = *(int64_t*) colDataGetData(pTsCol, i); if (ts == pSliceInfo->current) { - for(int32_t j = 0; j < pOperator->numOfExprs; ++j) { - SExprInfo* pExprInfo = &pOperator->pExpr[j]; + for(int32_t j = 0; j < pOperator->exprSupp.numOfExprs; ++j) { + SExprInfo* pExprInfo = &pOperator->exprSupp.pExprInfo[j]; int32_t dstSlot = pExprInfo->base.resSchema.slotId; int32_t srcSlot = pExprInfo->base.pParam[0].pCol->slotId; @@ -1764,8 +1773,8 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) { int64_t nextTs = *(int64_t*) colDataGetData(pTsCol, i + 1); if (nextTs > pSliceInfo->current) { // output the result - for (int32_t j = 0; j < pOperator->numOfExprs; ++j) { - SExprInfo* pExprInfo = &pOperator->pExpr[j]; + for (int32_t j = 0; j < pOperator->exprSupp.numOfExprs; ++j) { + SExprInfo* pExprInfo = &pOperator->exprSupp.pExprInfo[j]; int32_t dstSlot = pExprInfo->base.resSchema.slotId; int32_t srcSlot = pExprInfo->base.pParam[0].pCol->slotId; @@ -1903,7 +1912,9 @@ SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SExprInfo* goto _error; } - int32_t code = initTimesliceInfo(pInfo, pInfo->binfo.pCtx, numOfCols); + SExprSupp* pSup = &pOperator->exprSupp; + + int32_t code = initTimesliceInfo(pInfo, pSup->pCtx, numOfCols); if (code != TSDB_CODE_SUCCESS) { goto _error; } @@ -1917,8 +1928,8 @@ SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SExprInfo* // pOperator->operatorType = OP_AllTimeWindow; pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; - pOperator->pExpr = pExprInfo; - pOperator->numOfExprs = numOfCols; + pOperator->exprSupp.pExprInfo = pExprInfo; + pOperator->exprSupp.numOfExprs = numOfCols; pOperator->info = pInfo; pOperator->pTaskInfo = pTaskInfo; @@ -1955,7 +1966,7 @@ SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SExprInf size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES; initResultSizeInfo(pOperator, 4096); - initAggInfo(&pInfo->binfo, &pInfo->aggSup, pExpr, numOfCols, pResBlock, keyBufSize, pTaskInfo->id.str); + initAggInfo(&pInfo->binfo, &pOperator->exprSupp, &pInfo->aggSup, pExpr, numOfCols, pResBlock, keyBufSize, pTaskInfo->id.str); initResultRowInfo(&pInfo->binfo.resultRowInfo); pInfo->twAggSup = *pTwAggSup; @@ -1966,8 +1977,8 @@ SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SExprInf pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_MERGE_STATE; pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; - pOperator->pExpr = pExpr; - pOperator->numOfExprs = numOfCols; + pOperator->exprSupp.pExprInfo = pExpr; + pOperator->exprSupp.numOfExprs = numOfCols; pOperator->pTaskInfo = pTaskInfo; pOperator->info = pInfo; @@ -2000,7 +2011,7 @@ SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo initResultSizeInfo(pOperator, 4096); int32_t code = - initAggInfo(&pInfo->binfo, &pInfo->aggSup, pExprInfo, numOfCols, pResBlock, keyBufSize, pTaskInfo->id.str); + initAggInfo(&pInfo->binfo, &pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, pResBlock, keyBufSize, pTaskInfo->id.str); if (code != TSDB_CODE_SUCCESS) { goto _error; } @@ -2018,8 +2029,8 @@ SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION; pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; - pOperator->pExpr = pExprInfo; - pOperator->numOfExprs = numOfCols; + pOperator->exprSupp.pExprInfo = pExprInfo; + pOperator->exprSupp.numOfExprs = numOfCols; pOperator->info = pInfo; pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doSessionWindowAgg, NULL, NULL, @@ -2060,23 +2071,25 @@ void compactFunctions(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx, int3 } } -static void rebuildIntervalWindow(SStreamFinalIntervalOperatorInfo* pInfo, SArray* pWinArray, int32_t groupId, +static void rebuildIntervalWindow(SStreamFinalIntervalOperatorInfo* pInfo, SExprSupp* pSup, SArray* pWinArray, int32_t groupId, int32_t numOfOutput, SExecTaskInfo* pTaskInfo) { int32_t size = taosArrayGetSize(pWinArray); ASSERT(pInfo->pChildren); for (int32_t i = 0; i < size; i++) { STimeWindow* pParentWin = taosArrayGet(pWinArray, i); SResultRow* pCurResult = NULL; - setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, pParentWin, true, &pCurResult, 0, pInfo->binfo.pCtx, - numOfOutput, pInfo->binfo.rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo); + setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, pParentWin, true, &pCurResult, 0, pSup->pCtx, + numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo); int32_t numOfChildren = taosArrayGetSize(pInfo->pChildren); for (int32_t j = 0; j < numOfChildren; j++) { SOperatorInfo* pChildOp = taosArrayGetP(pInfo->pChildren, j); SIntervalAggOperatorInfo* pChInfo = pChildOp->info; + SExprSupp* pChildSup = &pChildOp->exprSupp; + SResultRow* pChResult = NULL; - setTimeWindowOutputBuf(&pChInfo->binfo.resultRowInfo, pParentWin, true, &pChResult, 0, pChInfo->binfo.pCtx, - pChildOp->numOfExprs, pChInfo->binfo.rowEntryInfoOffset, &pChInfo->aggSup, pTaskInfo); - compactFunctions(pInfo->binfo.pCtx, pChInfo->binfo.pCtx, numOfOutput, pTaskInfo); + setTimeWindowOutputBuf(&pChInfo->binfo.resultRowInfo, pParentWin, true, &pChResult, 0, pChildSup->pCtx, + pChildSup->numOfExprs, pChildSup->rowEntryInfoOffset, &pChInfo->aggSup, pTaskInfo); + compactFunctions(pSup->pCtx, pChildSup->pCtx, numOfOutput, pTaskInfo); } } } @@ -2093,7 +2106,8 @@ static void doHashInterval(SOperatorInfo* pOperatorInfo, SSDataBlock* pSDataBloc SStreamFinalIntervalOperatorInfo* pInfo = (SStreamFinalIntervalOperatorInfo*)pOperatorInfo->info; SResultRowInfo* pResultRowInfo = &(pInfo->binfo.resultRowInfo); SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo; - int32_t numOfOutput = pOperatorInfo->numOfExprs; + SExprSupp* pSup = &pOperatorInfo->exprSupp; + int32_t numOfOutput = pSup->numOfExprs; int32_t step = 1; bool ascScan = true; TSKEY* tsCols = NULL; @@ -2116,12 +2130,11 @@ static void doHashInterval(SOperatorInfo* pOperatorInfo, SSDataBlock* pSDataBloc isDeletedWindow(&nextWin, tableGroupId, &pInfo->aggSup)) { SArray* pUpWins = taosArrayInit(8, sizeof(STimeWindow)); taosArrayPush(pUpWins, &nextWin); - rebuildIntervalWindow(pInfo, pUpWins, pInfo->binfo.pRes->info.groupId, - pOperatorInfo->numOfExprs, pOperatorInfo->pTaskInfo); + rebuildIntervalWindow(pInfo, pSup, pUpWins, pInfo->binfo.pRes->info.groupId, pSup->numOfExprs, pOperatorInfo->pTaskInfo); taosArrayDestroy(pUpWins); } - int32_t code = setTimeWindowOutputBuf(pResultRowInfo, &nextWin, true, &pResult, tableGroupId, pInfo->binfo.pCtx, - numOfOutput, pInfo->binfo.rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo); + int32_t code = setTimeWindowOutputBuf(pResultRowInfo, &nextWin, true, &pResult, tableGroupId, pSup->pCtx, + numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo); if (code != TSDB_CODE_SUCCESS || pResult == NULL) { longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY); } @@ -2135,10 +2148,10 @@ static void doHashInterval(SOperatorInfo* pOperatorInfo, SSDataBlock* pSDataBloc saveResultRow(pResult, tableGroupId, pUpdated); } // window start(end) key interpolation - // doWindowBorderInterpolation(pInfo, pSDataBlock, numOfOutput, pInfo->binfo.pCtx, pResult, &nextWin, startPos, + // doWindowBorderInterpolation(pInfo, pSDataBlock, numOfOutput, pSup->pCtx, pResult, &nextWin, startPos, // forwardRows); updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &nextWin, true); - doApplyFunctions(pTaskInfo, pInfo->binfo.pCtx, &nextWin, &pInfo->twAggSup.timeWindowData, startPos, forwardRows, + doApplyFunctions(pTaskInfo, pSup->pCtx, &nextWin, &pInfo->twAggSup.timeWindowData, startPos, forwardRows, tsCols, pSDataBlock->info.rows, numOfOutput, TSDB_ORDER_ASC); int32_t prevEndPos = (forwardRows - 1) * step + startPos; ASSERT(pSDataBlock->info.window.skey > 0 && pSDataBlock->info.window.ekey > 0); @@ -2190,6 +2203,8 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) { SArray* pUpdated = taosArrayInit(4, POINTER_BYTES); TSKEY maxTs = INT64_MIN; + SExprSupp* pSup = &pOperator->exprSupp; + if (pOperator->status == OP_EXEC_DONE) { return NULL; } else if (pOperator->status == OP_RES_TO_RETURN) { @@ -2219,15 +2234,17 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) { if (pBlock->info.type == STREAM_REPROCESS) { SArray* pUpWins = taosArrayInit(8, sizeof(STimeWindow)); - doClearWindows(&pInfo->aggSup, &pInfo->binfo, &pInfo->interval, pInfo->primaryTsIndex, pOperator->numOfExprs, + doClearWindows(&pInfo->aggSup, pSup, &pInfo->interval, pInfo->primaryTsIndex, pOperator->exprSupp.numOfExprs, pBlock, pUpWins); if (isFinalInterval(pInfo)) { int32_t childIndex = getChildIndex(pBlock); SOperatorInfo* pChildOp = taosArrayGetP(pInfo->pChildren, childIndex); SIntervalAggOperatorInfo* pChildInfo = pChildOp->info; - doClearWindows(&pChildInfo->aggSup, &pChildInfo->binfo, &pChildInfo->interval, pChildInfo->primaryTsIndex, - pChildOp->numOfExprs, pBlock, NULL); - rebuildIntervalWindow(pInfo, pUpWins, pInfo->binfo.pRes->info.groupId, pOperator->numOfExprs, + SExprSupp* pChildSup = &pChildOp->exprSupp; + + doClearWindows(&pChildInfo->aggSup, pChildSup, &pChildInfo->interval, pChildInfo->primaryTsIndex, + pChildSup->numOfExprs, pBlock, NULL); + rebuildIntervalWindow(pInfo, pSup, pUpWins, pInfo->binfo.pRes->info.groupId, pOperator->exprSupp.numOfExprs, pOperator->pTaskInfo); taosArrayDestroy(pUpWins); continue; @@ -2241,7 +2258,7 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) { continue; } - setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, pInfo->order, MAIN_SCAN, true); + setInputDataBlock(pOperator, pSup->pCtx, pBlock, pInfo->order, MAIN_SCAN, true); doHashInterval(pOperator, pBlock, pBlock->info.groupId, pUpdated); if (isFinalInterval(pInfo)) { int32_t chIndex = getChildIndex(pBlock); @@ -2256,7 +2273,7 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) { } SOperatorInfo* pChildOp = taosArrayGetP(pInfo->pChildren, chIndex); SStreamFinalIntervalOperatorInfo* pChInfo = pChildOp->info; - setInputDataBlock(pChildOp, pChInfo->binfo.pCtx, pBlock, pChInfo->order, MAIN_SCAN, true); + setInputDataBlock(pChildOp, pChildOp->exprSupp.pCtx, pBlock, pChInfo->order, MAIN_SCAN, true); doHashInterval(pChildOp, pBlock, pBlock->info.groupId, NULL); } maxTs = TMAX(maxTs, pBlock->info.window.ekey); @@ -2267,7 +2284,7 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) { closeIntervalWindow(pInfo->aggSup.pResultRowHashTable, &pInfo->twAggSup, &pInfo->interval, pUpdated); } - finalizeUpdatedResult(pOperator->numOfExprs, pInfo->aggSup.pResultBuf, pUpdated, pInfo->binfo.rowEntryInfoOffset); + finalizeUpdatedResult(pOperator->exprSupp.numOfExprs, pInfo->aggSup.pResultBuf, pUpdated, pSup->rowEntryInfoOffset); initMultiResInfoFromArrayList(&pInfo->groupResInfo, pUpdated); blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity); doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf); @@ -2292,6 +2309,7 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, if (pInfo == NULL || pOperator == NULL) { goto _error; } + pInfo->order = TSDB_ORDER_ASC; pInfo->interval = (SInterval){.interval = pIntervalPhyNode->interval, .sliding = pIntervalPhyNode->sliding, @@ -2311,10 +2329,10 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, int32_t numOfCols = 0; SExprInfo* pExprInfo = createExprInfo(pIntervalPhyNode->window.pFuncs, NULL, &numOfCols); SSDataBlock* pResBlock = createResDataBlock(pPhyNode->pOutputDataBlockDesc); - int32_t code = initAggInfo(&pInfo->binfo, &pInfo->aggSup, pExprInfo, numOfCols, + int32_t code = initAggInfo(&pInfo->binfo, &pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, pResBlock, keyBufSize, pTaskInfo->id.str); ASSERT(numOfCols > 0); - increaseTs(pInfo->binfo.pCtx); + increaseTs(pOperator->exprSupp.pCtx); initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window); if (code != TSDB_CODE_SUCCESS) { goto _error; @@ -2345,9 +2363,9 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL; pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; - pOperator->pExpr = pExprInfo; + pOperator->exprSupp.pExprInfo = pExprInfo; pOperator->pTaskInfo = pTaskInfo; - pOperator->numOfExprs = numOfCols; + pOperator->exprSupp.numOfExprs = numOfCols; pOperator->info = pInfo; pOperator->fpSet = @@ -2397,14 +2415,14 @@ void destroyStreamSessionAggOperatorInfo(void* param, int32_t numOfOutput) { } } -int32_t initBiasicInfo(SOptrBasicInfo* pBasicInfo, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock) { - pBasicInfo->pCtx = createSqlFunctionCtx(pExprInfo, numOfCols, &pBasicInfo->rowEntryInfoOffset); +int32_t initBasicInfo(SOptrBasicInfo* pBasicInfo, SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock) { + pSup->pCtx = createSqlFunctionCtx(pExprInfo, numOfCols, &pSup->rowEntryInfoOffset); pBasicInfo->pRes = pResultBlock; for (int32_t i = 0; i < numOfCols; ++i) { - pBasicInfo->pCtx[i].pBuf = NULL; + pSup->pCtx[i].pBuf = NULL; } ASSERT(numOfCols > 0); - increaseTs(pBasicInfo->pCtx); + increaseTs(pSup->pCtx); return TSDB_CODE_SUCCESS; } @@ -2439,13 +2457,14 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh } initResultSizeInfo(pOperator, 4096); + SExprSupp* pSup = &pOperator->exprSupp; - code = initBiasicInfo(&pInfo->binfo, pExprInfo, numOfCols, pResBlock); + code = initBasicInfo(&pInfo->binfo, pSup, pExprInfo, numOfCols, pResBlock); if (code != TSDB_CODE_SUCCESS) { goto _error; } - code = initSessionAggSupporter(&pInfo->streamAggSup, "StreamSessionAggOperatorInfo", pInfo->binfo.pCtx, numOfCols); + code = initSessionAggSupporter(&pInfo->streamAggSup, "StreamSessionAggOperatorInfo", pSup->pCtx, numOfCols); if (code != TSDB_CODE_SUCCESS) { goto _error; } @@ -2454,7 +2473,7 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh if (pInfo->pDummyCtx == NULL) { goto _error; } - initDummyFunction(pInfo->pDummyCtx, pInfo->binfo.pCtx, numOfCols); + initDummyFunction(pInfo->pDummyCtx, pSup->pCtx, numOfCols); pInfo->twAggSup = (STimeWindowAggSupp) { .waterMark = pSessionNode->window.watermark, @@ -2479,8 +2498,8 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION; pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; - pOperator->pExpr = pExprInfo; - pOperator->numOfExprs = numOfCols; + pOperator->exprSupp.pExprInfo = pExprInfo; + pOperator->exprSupp.numOfExprs = numOfCols; pOperator->info = pInfo; pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamSessionAgg, NULL, NULL, destroyStreamSessionAggOperatorInfo, @@ -2628,32 +2647,35 @@ static int32_t setWindowOutputBuf(SResultWindowInfo* pWinInfo, SResultRow** pRes static int32_t doOneWindowAggImpl(int32_t tsColId, SOptrBasicInfo* pBinfo, SStreamAggSupporter* pAggSup, SColumnInfoData* pTimeWindowData, SSDataBlock* pSDataBlock, SResultWindowInfo* pCurWin, SResultRow** pResult, int32_t startIndex, int32_t winRows, - int32_t numOutput, SExecTaskInfo* pTaskInfo) { + int32_t numOutput, SOperatorInfo* pOperator) { + SExprSupp* pSup = &pOperator->exprSupp; + SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; + SColumnInfoData* pColDataInfo = taosArrayGet(pSDataBlock->pDataBlock, tsColId); TSKEY* tsCols = (int64_t*)pColDataInfo->pData; - int32_t code = setWindowOutputBuf(pCurWin, pResult, pBinfo->pCtx, pSDataBlock->info.groupId, numOutput, - pBinfo->rowEntryInfoOffset, pAggSup, pTaskInfo); + int32_t code = setWindowOutputBuf(pCurWin, pResult, pSup->pCtx, pSDataBlock->info.groupId, numOutput, + pSup->rowEntryInfoOffset, pAggSup, pTaskInfo); if (code != TSDB_CODE_SUCCESS || (*pResult) == NULL) { return TSDB_CODE_QRY_OUT_OF_MEMORY; } updateTimeWindowInfo(pTimeWindowData, &pCurWin->win, true); - doApplyFunctions(pTaskInfo, pBinfo->pCtx, &pCurWin->win, pTimeWindowData, startIndex, winRows, tsCols, + doApplyFunctions(pTaskInfo, pSup->pCtx, &pCurWin->win, pTimeWindowData, startIndex, winRows, tsCols, pSDataBlock->info.rows, numOutput, TSDB_ORDER_ASC); return TSDB_CODE_SUCCESS; } static int32_t doOneWindowAgg(SStreamSessionAggOperatorInfo* pInfo, SSDataBlock* pSDataBlock, SResultWindowInfo* pCurWin, SResultRow** pResult, int32_t startIndex, int32_t winRows, - int32_t numOutput, SExecTaskInfo* pTaskInfo) { + int32_t numOutput, SOperatorInfo * pOperator) { return doOneWindowAggImpl(pInfo->primaryTsIndex, &pInfo->binfo, &pInfo->streamAggSup, &pInfo->twAggSup.timeWindowData, - pSDataBlock, pCurWin, pResult, startIndex, winRows, numOutput, pTaskInfo); + pSDataBlock, pCurWin, pResult, startIndex, winRows, numOutput, pOperator); } static int32_t doOneStateWindowAgg(SStreamStateAggOperatorInfo* pInfo, SSDataBlock* pSDataBlock, SResultWindowInfo* pCurWin, SResultRow** pResult, int32_t startIndex, - int32_t winRows, int32_t numOutput, SExecTaskInfo* pTaskInfo) { + int32_t winRows, int32_t numOutput, SOperatorInfo * pOperator) { return doOneWindowAggImpl(pInfo->primaryTsIndex, &pInfo->binfo, &pInfo->streamAggSup, &pInfo->twAggSup.timeWindowData, - pSDataBlock, pCurWin, pResult, startIndex, winRows, numOutput, pTaskInfo); + pSDataBlock, pCurWin, pResult, startIndex, winRows, numOutput, pOperator); } int32_t getNumCompactWindow(SArray* pWinInfos, int32_t startIndex, int64_t gap) { @@ -2671,10 +2693,13 @@ int32_t getNumCompactWindow(SArray* pWinInfos, int32_t startIndex, int64_t gap) } void compactTimeWindow(SStreamSessionAggOperatorInfo* pInfo, int32_t startIndex, int32_t num, uint64_t groupId, - int32_t numOfOutput, SExecTaskInfo* pTaskInfo, SHashObj* pStUpdated, SHashObj* pStDeleted) { + int32_t numOfOutput, SHashObj* pStUpdated, SHashObj* pStDeleted, SOperatorInfo* pOperator) { + SExprSupp* pSup = &pOperator->exprSupp; + SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; + SResultWindowInfo* pCurWin = taosArrayGet(pInfo->streamAggSup.pCurWins, startIndex); SResultRow* pCurResult = NULL; - setWindowOutputBuf(pCurWin, &pCurResult, pInfo->binfo.pCtx, groupId, numOfOutput, pInfo->binfo.rowEntryInfoOffset, + setWindowOutputBuf(pCurWin, &pCurResult, pSup->pCtx, groupId, numOfOutput, pSup->rowEntryInfoOffset, &pInfo->streamAggSup, pTaskInfo); num += startIndex + 1; ASSERT(num <= taosArrayGetSize(pInfo->streamAggSup.pCurWins)); @@ -2682,10 +2707,10 @@ void compactTimeWindow(SStreamSessionAggOperatorInfo* pInfo, int32_t startIndex, for (int32_t i = startIndex + 1; i < num; i++) { SResultWindowInfo* pWinInfo = taosArrayGet(pInfo->streamAggSup.pCurWins, i); SResultRow* pWinResult = NULL; - setWindowOutputBuf(pWinInfo, &pWinResult, pInfo->pDummyCtx, groupId, numOfOutput, pInfo->binfo.rowEntryInfoOffset, + setWindowOutputBuf(pWinInfo, &pWinResult, pInfo->pDummyCtx, groupId, numOfOutput, pSup->rowEntryInfoOffset, &pInfo->streamAggSup, pTaskInfo); pCurWin->win.ekey = TMAX(pCurWin->win.ekey, pWinInfo->win.ekey); - compactFunctions(pInfo->binfo.pCtx, pInfo->pDummyCtx, numOfOutput, pTaskInfo); + compactFunctions(pSup->pCtx, pInfo->pDummyCtx, numOfOutput, pTaskInfo); taosHashRemove(pStUpdated, &pWinInfo->pos, sizeof(SResultRowPosition)); if (pWinInfo->isOutput) { taosHashPut(pStDeleted, &pWinInfo->pos, sizeof(SResultRowPosition), &pWinInfo->win.skey, sizeof(TSKEY)); @@ -2705,7 +2730,7 @@ static void doStreamSessionAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSData SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SStreamSessionAggOperatorInfo* pInfo = pOperator->info; bool masterScan = true; - int32_t numOfOutput = pOperator->numOfExprs; + int32_t numOfOutput = pOperator->exprSupp.numOfExprs; uint64_t groupId = pSDataBlock->info.groupId; int64_t gap = pInfo->gap; int64_t code = TSDB_CODE_SUCCESS; @@ -2728,17 +2753,17 @@ static void doStreamSessionAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSData int32_t winIndex = 0; SResultWindowInfo* pCurWin = getSessionTimeWindow(pAggSup, tsCols[i], groupId, gap, &winIndex); winRows = updateSessionWindowInfo(pCurWin, tsCols, pSDataBlock->info.rows, i, pInfo->gap, pStDeleted); - code = doOneWindowAgg(pInfo, pSDataBlock, pCurWin, &pResult, i, winRows, numOfOutput, pTaskInfo); + code = doOneWindowAgg(pInfo, pSDataBlock, pCurWin, &pResult, i, winRows, numOfOutput, pOperator); if (code != TSDB_CODE_SUCCESS || pResult == NULL) { longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY); } // window start(end) key interpolation - // doWindowBorderInterpolation(pOperatorInfo, pSDataBlock, pInfo->binfo.pCtx, pResult, &nextWin, startPos, + // doWindowBorderInterpolation(pOperatorInfo, pSDataBlock, pSup->pCtx, pResult, &nextWin, startPos, // forwardRows, // pInfo->order, false); int32_t winNum = getNumCompactWindow(pAggSup->pCurWins, winIndex, gap); if (winNum > 0) { - compactTimeWindow(pInfo, winIndex, winNum, groupId, numOfOutput, pTaskInfo, pStUpdated, pStDeleted); + compactTimeWindow(pInfo, winIndex, winNum, groupId, numOfOutput, pStUpdated, pStDeleted, pOperator); } pCurWin->isClosed = false; if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE) { @@ -2753,7 +2778,7 @@ static void doStreamSessionAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSData } } -static void doClearSessionWindows(SStreamAggSupporter* pAggSup, SOptrBasicInfo* pBinfo, SSDataBlock* pBlock, +static void doClearSessionWindows(SStreamAggSupporter* pAggSup, SExprSupp* pSup, SSDataBlock* pBlock, int32_t tsIndex, int32_t numOfOutput, int64_t gap, SArray* result) { SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, tsIndex); TSKEY* tsCols = (TSKEY*)pColDataInfo->pData; @@ -2763,7 +2788,7 @@ static void doClearSessionWindows(SStreamAggSupporter* pAggSup, SOptrBasicInfo* SResultWindowInfo* pCurWin = getSessionTimeWindow(pAggSup, tsCols[i], pBlock->info.groupId, gap, &winIndex); step = updateSessionWindowInfo(pCurWin, tsCols, pBlock->info.rows, i, gap, NULL); ASSERT(isInWindow(pCurWin, tsCols[i], gap)); - doClearWindowImpl(&pCurWin->pos, pAggSup->pResultBuf, pBinfo, numOfOutput); + doClearWindowImpl(&pCurWin->pos, pAggSup->pResultBuf, pSup, numOfOutput); if (result) { taosArrayPush(result, pCurWin); } @@ -2809,13 +2834,17 @@ void doBuildDeleteDataBlock(SHashObj* pStDeleted, SSDataBlock* pBlock, void** It } static void rebuildTimeWindow(SStreamSessionAggOperatorInfo* pInfo, SArray* pWinArray, int32_t groupId, - int32_t numOfOutput, SExecTaskInfo* pTaskInfo) { + int32_t numOfOutput, SOperatorInfo* pOperator) { + SExprSupp* pSup = &pOperator->exprSupp; + SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; + int32_t size = taosArrayGetSize(pWinArray); ASSERT(pInfo->pChildren); + for (int32_t i = 0; i < size; i++) { SResultWindowInfo* pParentWin = taosArrayGet(pWinArray, i); SResultRow* pCurResult = NULL; - setWindowOutputBuf(pParentWin, &pCurResult, pInfo->binfo.pCtx, groupId, numOfOutput, pInfo->binfo.rowEntryInfoOffset, + setWindowOutputBuf(pParentWin, &pCurResult, pSup->pCtx, groupId, numOfOutput, pSup->rowEntryInfoOffset, &pInfo->streamAggSup, pTaskInfo); int32_t numOfChildren = taosArrayGetSize(pInfo->pChildren); for (int32_t j = 0; j < numOfChildren; j++) { @@ -2828,9 +2857,9 @@ static void rebuildTimeWindow(SStreamSessionAggOperatorInfo* pInfo, SArray* pWin SResultWindowInfo* pcw = taosArrayGet(pChWins, k); if (pParentWin->win.skey <= pcw->win.skey && pcw->win.ekey <= pParentWin->win.ekey) { SResultRow* pChResult = NULL; - setWindowOutputBuf(pcw, &pChResult, pChInfo->binfo.pCtx, groupId, numOfOutput, - pChInfo->binfo.rowEntryInfoOffset, &pChInfo->streamAggSup, pTaskInfo); - compactFunctions(pInfo->binfo.pCtx, pChInfo->binfo.pCtx, numOfOutput, pTaskInfo); + setWindowOutputBuf(pcw, &pChResult, pChild->exprSupp.pCtx, groupId, numOfOutput, + pChild->exprSupp.rowEntryInfoOffset, &pChInfo->streamAggSup, pTaskInfo); + compactFunctions(pSup->pCtx, pChild->exprSupp.pCtx, numOfOutput, pTaskInfo); continue; } break; @@ -2895,6 +2924,7 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) { return NULL; } + SExprSupp* pSup = &pOperator->exprSupp; SStreamSessionAggOperatorInfo* pInfo = pOperator->info; SOptrBasicInfo* pBInfo = &pInfo->binfo; if (pOperator->status == OP_RES_TO_RETURN) { @@ -2921,14 +2951,14 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) { if (pBlock->info.type == STREAM_REPROCESS) { SArray* pWins = taosArrayInit(16, sizeof(SResultWindowInfo)); - doClearSessionWindows(&pInfo->streamAggSup, &pInfo->binfo, pBlock, 0, pOperator->numOfExprs, pInfo->gap, pWins); + doClearSessionWindows(&pInfo->streamAggSup, &pOperator->exprSupp, pBlock, 0, pOperator->exprSupp.numOfExprs, pInfo->gap, pWins); if (isFinalSession(pInfo)) { int32_t childIndex = 0; // Todo(liuyao) get child id from SSDataBlock SOperatorInfo* pChildOp = taosArrayGetP(pInfo->pChildren, childIndex); SStreamSessionAggOperatorInfo* pChildInfo = pChildOp->info; - doClearSessionWindows(&pChildInfo->streamAggSup, &pChildInfo->binfo, pBlock, 0, pChildOp->numOfExprs, + doClearSessionWindows(&pChildInfo->streamAggSup, &pChildOp->exprSupp, pBlock, 0, pChildOp->exprSupp.numOfExprs, pChildInfo->gap, NULL); - rebuildTimeWindow(pInfo, pWins, pBlock->info.groupId, pOperator->numOfExprs, pOperator->pTaskInfo); + rebuildTimeWindow(pInfo, pWins, pBlock->info.groupId, pOperator->exprSupp.numOfExprs, pOperator); } taosArrayDestroy(pWins); continue; @@ -2938,7 +2968,7 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) { } // the pDataBlock are always the same one, no need to call this again - setInputDataBlock(pOperator, pBInfo->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true); + setInputDataBlock(pOperator, pSup->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true); if (isFinalSession(pInfo)) { int32_t childIndex = 0; // Todo(liuyao) get child id from SSDataBlock SOptrBasicInfo* pChildOp = taosArrayGetP(pInfo->pChildren, childIndex); @@ -2955,8 +2985,8 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) { copyUpdateResult(pStUpdated, pUpdated); taosHashCleanup(pStUpdated); - finalizeUpdatedResult(pOperator->numOfExprs, pInfo->streamAggSup.pResultBuf, pUpdated, - pInfo->binfo.rowEntryInfoOffset); + finalizeUpdatedResult(pOperator->exprSupp.numOfExprs, pInfo->streamAggSup.pResultBuf, pUpdated, + pSup->rowEntryInfoOffset); initMultiResInfoFromArrayList(&pInfo->groupResInfo, pUpdated); blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity); doBuildDeleteDataBlock(pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator); @@ -2990,7 +3020,7 @@ SOperatorInfo* createStreamFinalSessionAggOperatorInfo(SOperatorInfo* downstream _error: if (pInfo != NULL) { - destroyStreamSessionAggOperatorInfo(pInfo, pOperator->numOfExprs); + destroyStreamSessionAggOperatorInfo(pInfo, pOperator->exprSupp.numOfExprs); } taosMemoryFreeClear(pInfo); @@ -3209,7 +3239,7 @@ static void doStreamStateAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SStreamStateAggOperatorInfo* pInfo = pOperator->info; bool masterScan = true; - int32_t numOfOutput = pOperator->numOfExprs; + int32_t numOfOutput = pOperator->exprSupp.numOfExprs; int64_t groupId = pSDataBlock->info.groupId; int64_t code = TSDB_CODE_SUCCESS; int32_t step = 1; @@ -3241,7 +3271,7 @@ static void doStreamStateAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl deleteWindow(pAggSup->pCurWins, winIndex); continue; } - code = doOneStateWindowAgg(pInfo, pSDataBlock, &pCurWin->winInfo, &pResult, i, winRows, numOfOutput, pTaskInfo); + code = doOneStateWindowAgg(pInfo, pSDataBlock, &pCurWin->winInfo, &pResult, i, winRows, numOfOutput, pOperator); if (code != TSDB_CODE_SUCCESS || pResult == NULL) { longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY); } @@ -3263,6 +3293,7 @@ static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) { return NULL; } + SExprSupp* pSup = &pOperator->exprSupp; SStreamStateAggOperatorInfo* pInfo = pOperator->info; SOptrBasicInfo* pBInfo = &pInfo->binfo; if (pOperator->status == OP_RES_TO_RETURN) { @@ -3297,7 +3328,7 @@ static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) { } // the pDataBlock are always the same one, no need to call this again - setInputDataBlock(pOperator, pBInfo->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true); + setInputDataBlock(pOperator, pSup->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true); doStreamStateAggImpl(pOperator, pBlock, pSeUpdated, pInfo->pSeDeleted); pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.window.ekey); } @@ -3309,8 +3340,8 @@ static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) { copyUpdateResult(pSeUpdated, pUpdated); taosHashCleanup(pSeUpdated); - finalizeUpdatedResult(pOperator->numOfExprs, pInfo->streamAggSup.pResultBuf, pUpdated, - pInfo->binfo.rowEntryInfoOffset); + finalizeUpdatedResult(pOperator->exprSupp.numOfExprs, pInfo->streamAggSup.pResultBuf, pUpdated, + pSup->rowEntryInfoOffset); initMultiResInfoFromArrayList(&pInfo->groupResInfo, pUpdated); blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity); doBuildDeleteDataBlock(pInfo->pSeDeleted, pInfo->pDelRes, &pInfo->pDelIterator); @@ -3339,6 +3370,8 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys goto _error; } + SExprSupp* pSup = &pOperator->exprSupp; + int32_t numOfCols = 0; SExprInfo* pExprInfo = createExprInfo(pStateNode->window.pFuncs, NULL, &numOfCols); @@ -3352,12 +3385,12 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys }; initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window); - code = initBiasicInfo(&pInfo->binfo, pExprInfo, numOfCols, pResBlock); + code = initBasicInfo(&pInfo->binfo, pSup, pExprInfo, numOfCols, pResBlock); if (code != TSDB_CODE_SUCCESS) { goto _error; } - code = initStateAggSupporter(&pInfo->streamAggSup, "StreamStateAggOperatorInfo", pInfo->binfo.pCtx, numOfCols); + code = initStateAggSupporter(&pInfo->streamAggSup, "StreamStateAggOperatorInfo", pSup->pCtx, numOfCols); if (code != TSDB_CODE_SUCCESS) { goto _error; } @@ -3367,7 +3400,7 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys goto _error; } - initDummyFunction(pInfo->pDummyCtx, pInfo->binfo.pCtx, numOfCols); + initDummyFunction(pInfo->pDummyCtx, pSup->pCtx, numOfCols); pInfo->primaryTsIndex = tsSlotId; pInfo->order = TSDB_ORDER_ASC; _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY); @@ -3381,8 +3414,8 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE; pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; - pOperator->numOfExprs = numOfCols; - pOperator->pExpr = pExprInfo; + pOperator->exprSupp.numOfExprs = numOfCols; + pOperator->exprSupp.pExprInfo = pExprInfo; pOperator->pTaskInfo = pTaskInfo; pOperator->info = pInfo; pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamStateAgg, NULL, NULL, @@ -3420,6 +3453,8 @@ static int32_t outputMergeIntervalResult(SOperatorInfo* pOperatorInfo, uint64_t SMergeIntervalAggOperatorInfo* miaInfo = pOperatorInfo->info; SIntervalAggOperatorInfo* iaInfo = &miaInfo->intervalAggOperatorInfo; SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo; + + SExprSupp* pSup = &pOperatorInfo->exprSupp; bool ascScan = (iaInfo->order == TSDB_ORDER_ASC); SET_RES_WINDOW_KEY(iaInfo->aggSup.keyBuf, &wstartTs, TSDB_KEYSIZE, tableGroupId); @@ -3427,8 +3462,8 @@ static int32_t outputMergeIntervalResult(SOperatorInfo* pOperatorInfo, uint64_t GET_RES_WINDOW_KEY_LEN(TSDB_KEYSIZE)); ASSERT(p1 != NULL); - finalizeResultRowIntoResultDataBlock(iaInfo->aggSup.pResultBuf, p1, iaInfo->binfo.pCtx, pOperatorInfo->pExpr, - pOperatorInfo->numOfExprs, iaInfo->binfo.rowEntryInfoOffset, pResultBlock, + finalizeResultRowIntoResultDataBlock(iaInfo->aggSup.pResultBuf, p1, pSup->pCtx, pSup->pExprInfo, + pSup->numOfExprs, pSup->rowEntryInfoOffset, pResultBlock, pTaskInfo); taosHashRemove(iaInfo->aggSup.pResultRowHashTable, iaInfo->aggSup.keyBuf, GET_RES_WINDOW_KEY_LEN(TSDB_KEYSIZE)); @@ -3441,9 +3476,10 @@ static void doMergeIntervalAggImpl(SOperatorInfo* pOperatorInfo, SResultRowInfo* SIntervalAggOperatorInfo* iaInfo = &miaInfo->intervalAggOperatorInfo; SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo; + SExprSupp* pSup = &pOperatorInfo->exprSupp; int32_t startPos = 0; - int32_t numOfOutput = pOperatorInfo->numOfExprs; + int32_t numOfOutput = pSup->numOfExprs; int64_t* tsCols = extractTsCol(pBlock, iaInfo); uint64_t tableGroupId = pBlock->info.groupId; TSKEY blockStartTs = getStartTsKey(&pBlock->info.window, tsCols); @@ -3455,8 +3491,8 @@ static void doMergeIntervalAggImpl(SOperatorInfo* pOperatorInfo, SResultRowInfo* //TODO: remove the hash table usage (groupid + winkey => result row position) int32_t ret = - setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, iaInfo->binfo.pCtx, - numOfOutput, iaInfo->binfo.rowEntryInfoOffset, &iaInfo->aggSup, pTaskInfo); + setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, pSup->pCtx, + numOfOutput, pSup->rowEntryInfoOffset, &iaInfo->aggSup, pTaskInfo); if (ret != TSDB_CODE_SUCCESS || pResult == NULL) { longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY); } @@ -3473,7 +3509,7 @@ static void doMergeIntervalAggImpl(SOperatorInfo* pOperatorInfo, SResultRowInfo* continue; } else { updateTimeWindowInfo(&iaInfo->twAggSup.timeWindowData, &currWin, true); - doApplyFunctions(pTaskInfo, iaInfo->binfo.pCtx, &currWin, &iaInfo->twAggSup.timeWindowData, startPos, + doApplyFunctions(pTaskInfo, pSup->pCtx, &currWin, &iaInfo->twAggSup.timeWindowData, startPos, currPos - startPos, tsCols, pBlock->info.rows, numOfOutput, iaInfo->order); outputMergeIntervalResult(pOperatorInfo, tableGroupId, pResultBlock, currTs); @@ -3482,15 +3518,15 @@ static void doMergeIntervalAggImpl(SOperatorInfo* pOperatorInfo, SResultRowInfo* currWin.skey = currTs; currWin.ekey = taosTimeAdd(currWin.skey, iaInfo->interval.interval, iaInfo->interval.intervalUnit, iaInfo->interval.precision) - 1; startPos = currPos; - ret = setTimeWindowOutputBuf(pResultRowInfo, &currWin, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, iaInfo->binfo.pCtx, - numOfOutput, iaInfo->binfo.rowEntryInfoOffset, &iaInfo->aggSup, pTaskInfo); + ret = setTimeWindowOutputBuf(pResultRowInfo, &currWin, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, pSup->pCtx, + numOfOutput, pSup->rowEntryInfoOffset, &iaInfo->aggSup, pTaskInfo); if (ret != TSDB_CODE_SUCCESS || pResult == NULL) { longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY); } } } updateTimeWindowInfo(&iaInfo->twAggSup.timeWindowData, &currWin, true); - doApplyFunctions(pTaskInfo, iaInfo->binfo.pCtx, &currWin, &iaInfo->twAggSup.timeWindowData, startPos, + doApplyFunctions(pTaskInfo, pSup->pCtx, &currWin, &iaInfo->twAggSup.timeWindowData, startPos, currPos - startPos, tsCols, pBlock->info.rows, numOfOutput, iaInfo->order); outputMergeIntervalResult(pOperatorInfo, tableGroupId, pResultBlock, currTs); @@ -3505,6 +3541,7 @@ static SSDataBlock* doMergeIntervalAgg(SOperatorInfo* pOperator) { return NULL; } + SExprSupp* pSup = &pOperator->exprSupp; SSDataBlock* pRes = iaInfo->binfo.pRes; blockDataCleanup(pRes); blockDataEnsureCapacity(pRes, pOperator->resultInfo.capacity); @@ -3535,7 +3572,7 @@ static SSDataBlock* doMergeIntervalAgg(SOperatorInfo* pOperator) { } getTableScanInfo(pOperator, &iaInfo->order, &scanFlag); - setInputDataBlock(pOperator, iaInfo->binfo.pCtx, pBlock, iaInfo->order, scanFlag, true); + setInputDataBlock(pOperator, pSup->pCtx, pBlock, iaInfo->order, scanFlag, true); doMergeIntervalAggImpl(pOperator, &iaInfo->binfo.resultRowInfo, pBlock, scanFlag, pRes); if (pRes->info.rows >= pOperator->resultInfo.threshold) { @@ -3565,6 +3602,7 @@ SOperatorInfo* createMergeIntervalOperatorInfo(SOperatorInfo* downstream, SExprI } SIntervalAggOperatorInfo* iaInfo = &miaInfo->intervalAggOperatorInfo; + SExprSupp* pSup = &pOperator->exprSupp; iaInfo->win = pTaskInfo->window; iaInfo->order = TSDB_ORDER_ASC; @@ -3578,11 +3616,11 @@ SOperatorInfo* createMergeIntervalOperatorInfo(SOperatorInfo* downstream, SExprI initResultSizeInfo(pOperator, 4096); int32_t code = - initAggInfo(&iaInfo->binfo, &iaInfo->aggSup, pExprInfo, numOfCols, pResBlock, keyBufSize, pTaskInfo->id.str); + initAggInfo(&iaInfo->binfo, &pOperator->exprSupp, &iaInfo->aggSup, pExprInfo, numOfCols, pResBlock, keyBufSize, pTaskInfo->id.str); initExecTimeWindowInfo(&iaInfo->twAggSup.timeWindowData, &iaInfo->win); - iaInfo->timeWindowInterpo = timeWindowinterpNeeded(iaInfo->binfo.pCtx, numOfCols, iaInfo); + iaInfo->timeWindowInterpo = timeWindowinterpNeeded(pSup->pCtx, numOfCols, iaInfo); if (iaInfo->timeWindowInterpo) { iaInfo->binfo.resultRowInfo.openWindow = tdListNew(sizeof(SResultRowPosition)); } @@ -3598,9 +3636,9 @@ SOperatorInfo* createMergeIntervalOperatorInfo(SOperatorInfo* downstream, SExprI pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_MERGE_INTERVAL; pOperator->blocking = false; pOperator->status = OP_NOT_OPENED; - pOperator->pExpr = pExprInfo; + pOperator->exprSupp.pExprInfo = pExprInfo; pOperator->pTaskInfo = pTaskInfo; - pOperator->numOfExprs = numOfCols; + pOperator->exprSupp.numOfExprs = numOfCols; pOperator->info = miaInfo; pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doMergeIntervalAgg, NULL, NULL, diff --git a/tests/script/tsim/testsuit.sim b/tests/script/tsim/testsuit.sim index 90210fd436f70ce7d6188b9fd78a0d970e1764c9..1636bb38c800350a2b39a26d5dcd9a25ab56dc23 100644 --- a/tests/script/tsim/testsuit.sim +++ b/tests/script/tsim/testsuit.sim @@ -1,35 +1,35 @@ -#run tsim/user/pass_alter.sim -#run tsim/user/basic1.sim -#run tsim/user/privilege2.sim -#run tsim/user/user_len.sim -#run tsim/user/privilege1.sim -#run tsim/user/pass_len.sim -#run tsim/table/basic1.sim -#run tsim/trans/lossdata1.sim -#run tsim/trans/create_db.sim -#run tsim/stable/alter_metrics.sim -#run tsim/stable/tag_modify.sim -#run tsim/stable/alter_comment.sim -#run tsim/stable/column_drop.sim -#run tsim/stable/column_modify.sim -#run tsim/stable/tag_rename.sim -#run tsim/stable/vnode3.sim -#run tsim/stable/metrics.sim -#run tsim/stable/alter_insert2.sim -#run tsim/stable/show.sim -#run tsim/stable/alter_import.sim -#run tsim/stable/tag_add.sim -#run tsim/stable/tag_drop.sim -#run tsim/stable/column_add.sim -#run tsim/stable/alter_count.sim -#run tsim/stable/values.sim +run tsim/user/pass_alter.sim +run tsim/user/basic1.sim +run tsim/user/privilege2.sim +run tsim/user/user_len.sim +run tsim/user/privilege1.sim +run tsim/user/pass_len.sim +run tsim/table/basic1.sim +run tsim/trans/lossdata1.sim +run tsim/trans/create_db.sim +run tsim/stable/alter_metrics.sim +run tsim/stable/tag_modify.sim +run tsim/stable/alter_comment.sim +run tsim/stable/column_drop.sim +run tsim/stable/column_modify.sim +run tsim/stable/tag_rename.sim +run tsim/stable/vnode3.sim +run tsim/stable/metrics.sim +run tsim/stable/alter_insert2.sim +run tsim/stable/show.sim +run tsim/stable/alter_import.sim +run tsim/stable/tag_add.sim +run tsim/stable/tag_drop.sim +run tsim/stable/column_add.sim +run tsim/stable/alter_count.sim +run tsim/stable/values.sim run tsim/stable/dnode3.sim -#run tsim/stable/alter_insert1.sim -#run tsim/stable/refcount.sim -#run tsim/stable/disk.sim +run tsim/stable/alter_insert1.sim +run tsim/stable/refcount.sim +run tsim/stable/disk.sim run tsim/db/basic1.sim run tsim/db/basic3.sim -#run tsim/db/basic7.sim +run tsim/db/basic7.sim run tsim/db/basic6.sim run tsim/db/create_all_options.sim run tsim/db/basic2.sim