diff --git a/include/libs/function/functionMgt.h b/include/libs/function/functionMgt.h index 402b8f0309..eebb69e6ba 100644 --- a/include/libs/function/functionMgt.h +++ b/include/libs/function/functionMgt.h @@ -157,6 +157,8 @@ typedef enum EFunctionType { FUNCTION_TYPE_AVG_MERGE, FUNCTION_TYPE_STDDEV_PARTIAL, FUNCTION_TYPE_STDDEV_MERGE, + FUNCTION_TYPE_IRATE_PARTIAL, + FUNCTION_TYPE_IRATE_MERGE, // geometry functions FUNCTION_TYPE_GEOM_FROM_TEXT = 4250, diff --git a/source/libs/function/inc/builtinsimpl.h b/source/libs/function/inc/builtinsimpl.h index c3afc30a7b..d2f19ed2eb 100644 --- a/source/libs/function/inc/builtinsimpl.h +++ b/source/libs/function/inc/builtinsimpl.h @@ -127,7 +127,10 @@ int32_t derivativeFunction(SqlFunctionCtx* pCtx); bool getIrateFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv); bool irateFuncSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResInfo); int32_t irateFunction(SqlFunctionCtx* pCtx); +int32_t irateFunctionMerge(SqlFunctionCtx* pCtx); int32_t irateFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock); +int32_t iratePartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock); +int32_t getIrateInfoSize(); int32_t cachedLastRowFunction(SqlFunctionCtx* pCtx); diff --git a/source/libs/function/src/builtins.c b/source/libs/function/src/builtins.c index cee4000155..ef7c9d1442 100644 --- a/source/libs/function/src/builtins.c +++ b/source/libs/function/src/builtins.c @@ -1567,6 +1567,43 @@ static int32_t translateIrate(SFunctionNode* pFunc, char* pErrBuf, int32_t len) return TSDB_CODE_SUCCESS; } +static int32_t translateIrateImpl(SFunctionNode* pFunc, char* pErrBuf, int32_t len, bool isPartial) { + if (3 != LIST_LENGTH(pFunc->pParameterList)) { + return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName); + } + + uint8_t colType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type; + if (isPartial) { + if (!IS_NUMERIC_TYPE(colType)) { + return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName); + } + pFunc->node.resType = (SDataType){.bytes = getIrateInfoSize() + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY}; + } else { + if (TSDB_DATA_TYPE_BINARY != colType) { + return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName); + } + pFunc->node.resType = (SDataType){.bytes = tDataTypes[TSDB_DATA_TYPE_DOUBLE].bytes, .type = TSDB_DATA_TYPE_DOUBLE}; + + // add database precision as param + uint8_t dbPrec = pFunc->node.resType.precision; + int32_t code = addDbPrecisonParam(&pFunc->pParameterList, dbPrec); + if (code != TSDB_CODE_SUCCESS) { + return code; + } + } + + + return TSDB_CODE_SUCCESS; +} + +static int32_t translateIratePartial(SFunctionNode* pFunc, char* pErrBuf, int32_t len) { + return translateIrateImpl(pFunc, pErrBuf, len, true); +} + +static int32_t translateIrateMerge(SFunctionNode* pFunc, char* pErrBuf, int32_t len) { + return translateIrateImpl(pFunc, pErrBuf, len, false); +} + static int32_t translateInterp(SFunctionNode* pFunc, char* pErrBuf, int32_t len) { int32_t numOfParams = LIST_LENGTH(pFunc->pParameterList); uint8_t dbPrec = pFunc->node.resType.precision; @@ -2604,6 +2641,32 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { .initFunc = irateFuncSetup, .processFunc = irateFunction, .sprocessFunc = irateScalarFunction, + .finalizeFunc = irateFinalize, + .pPartialFunc = "_irate_partial", + .pMergeFunc = "_irate_merge" + }, + { + .name = "_irate_partial", + .type = FUNCTION_TYPE_IRATE_PARTIAL, + .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC | FUNC_MGT_FORBID_STREAM_FUNC | + FUNC_MGT_FORBID_SYSTABLE_FUNC, + .translateFunc = translateIratePartial, + .getEnvFunc = getIrateFuncEnv, + .initFunc = irateFuncSetup, + .processFunc = irateFunction, + .sprocessFunc = irateScalarFunction, + .finalizeFunc = iratePartialFinalize + }, + { + .name = "_irate_merge", + .type = FUNCTION_TYPE_IRATE_MERGE, + .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC | FUNC_MGT_FORBID_STREAM_FUNC | + FUNC_MGT_FORBID_SYSTABLE_FUNC, + .translateFunc = translateIrateMerge, + .getEnvFunc = getIrateFuncEnv, + .initFunc = irateFuncSetup, + .processFunc = irateFunctionMerge, + .sprocessFunc = irateScalarFunction, .finalizeFunc = irateFinalize }, { diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c index fad8c9ca5b..631f943915 100644 --- a/source/libs/function/src/builtinsimpl.c +++ b/source/libs/function/src/builtinsimpl.c @@ -5768,6 +5768,8 @@ int32_t derivativeFunction(SqlFunctionCtx* pCtx) { return TSDB_CODE_SUCCESS; } +int32_t getIrateInfoSize() { return (int32_t)sizeof(SRateInfo); } + bool getIrateFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv) { pEnv->calcMemSize = sizeof(SRateInfo); return true; @@ -5868,6 +5870,85 @@ static double doCalcRate(const SRateInfo* pRateInfo, double tickPerSec) { return (duration > 0) ? ((double)diff) / (duration / tickPerSec) : 0.0; } +static void irateTransferInfoImpl(TSKEY inputKey, SRateInfo* pInput, SRateInfo* pOutput) { + if (inputKey > pOutput->lastKey) { + pOutput->firstKey = pOutput->lastKey; + pOutput->lastKey = pInput->firstKey; + + pOutput->firstValue = pOutput->lastValue; + pOutput->lastValue = pInput->firstValue; + } else if ((inputKey < pOutput->lastKey) && (inputKey > pOutput->firstKey)) { + pOutput->firstKey = pOutput->lastKey; + pOutput->firstValue = pOutput->lastValue; + } else { + // inputKey < pOutput->firstKey + } +} + +static int32_t irateTransferInfo(SRateInfo* pInput, SRateInfo* pOutput) { + pOutput->hasResult = pInput->hasResult; + if (pInput->firstKey == pOutput->firstKey || pInput->firstKey == pOutput->lastKey || + pInput->lastKey == pOutput->firstKey || pInput->lastKey == pOutput->lastKey) { + return TSDB_CODE_FUNC_DUP_TIMESTAMP; + } + + if (pInput->firstKey != INT64_MIN) { + irateTransferInfoImpl(pInput->firstKey, pInput, pOutput); + } + + if (pInput->lastKey != INT64_MIN) { + irateTransferInfoImpl(pInput->lastKey, pInput, pOutput); + } + + return TSDB_CODE_SUCCESS; +} + +int32_t irateFunctionMerge(SqlFunctionCtx* pCtx) { + SInputColumnInfoData* pInput = &pCtx->input; + SColumnInfoData* pCol = pInput->pData[0]; + if (pCol->info.type != TSDB_DATA_TYPE_BINARY) { + return TSDB_CODE_FUNC_FUNTION_PARA_TYPE; + } + + SRateInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx)); + + int32_t start = pInput->startRowIndex; + for (int32_t i = start; i < start + pInput->numOfRows; ++i) { + char* data = colDataGetData(pCol, i); + SRateInfo* pInputInfo = (SRateInfo*)varDataVal(data); + if (pInputInfo->hasResult) { + int32_t code = irateTransferInfo(pInputInfo, pInfo); + if (code != TSDB_CODE_SUCCESS) { + return code; + } + } + } + + if (pInfo->hasResult) { + GET_RES_INFO(pCtx)->numOfRes = 1; + } + + return TSDB_CODE_SUCCESS; +} + +int32_t iratePartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) { + SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx); + SRateInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx)); + int32_t resultBytes = getIrateInfoSize(); + char* res = taosMemoryCalloc(resultBytes + VARSTR_HEADER_SIZE, sizeof(char)); + + memcpy(varDataVal(res), pInfo, resultBytes); + varDataSetLen(res, resultBytes); + + int32_t slotId = pCtx->pExpr->base.resSchema.slotId; + SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId); + + colDataSetVal(pCol, pBlock->info.rows, res, false); + + taosMemoryFree(res); + return pResInfo->numOfRes; +} + int32_t irateFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) { int32_t slotId = pCtx->pExpr->base.resSchema.slotId; SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);