From 62506bac334ec4e8fdf4d4f9f8f1abe7c9655438 Mon Sep 17 00:00:00 2001 From: Ganlin Zhao Date: Thu, 9 Jun 2022 19:15:09 +0800 Subject: [PATCH 1/3] add elapsed function partial/merge translate --- include/libs/function/functionMgt.h | 2 + source/libs/function/inc/builtinsimpl.h | 1 + source/libs/function/src/builtins.c | 80 +++++++++++++++++++++++++ source/libs/function/src/builtinsimpl.c | 4 ++ 4 files changed, 87 insertions(+) diff --git a/include/libs/function/functionMgt.h b/include/libs/function/functionMgt.h index d77b914fae..03e11b0502 100644 --- a/include/libs/function/functionMgt.h +++ b/include/libs/function/functionMgt.h @@ -131,6 +131,8 @@ typedef enum EFunctionType { FUNCTION_TYPE_HISTOGRAM_MERGE, FUNCTION_TYPE_HYPERLOGLOG_PARTIAL, FUNCTION_TYPE_HYPERLOGLOG_MERGE, + FUNCTION_TYPE_ELAPSED_PARTIAL, + FUNCTION_TYPE_ELAPSED_MERGE, // user defined funcion FUNCTION_TYPE_UDF = 10000 diff --git a/source/libs/function/inc/builtinsimpl.h b/source/libs/function/inc/builtinsimpl.h index 614be17e08..c4f6f0cc45 100644 --- a/source/libs/function/inc/builtinsimpl.h +++ b/source/libs/function/inc/builtinsimpl.h @@ -115,6 +115,7 @@ bool getElapsedFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv); bool elapsedFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo); int32_t elapsedFunction(SqlFunctionCtx* pCtx); int32_t elapsedFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock); +int32_t getElapsedInfoSize(); bool getHistogramFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv); bool histogramFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo); diff --git a/source/libs/function/src/builtins.c b/source/libs/function/src/builtins.c index e858fb3159..04cfa9d360 100644 --- a/source/libs/function/src/builtins.c +++ b/source/libs/function/src/builtins.c @@ -444,6 +444,64 @@ static int32_t translateElapsed(SFunctionNode* pFunc, char* pErrBuf, int32_t len return TSDB_CODE_SUCCESS; } +static int32_t translateElapsedImpl(SFunctionNode* pFunc, char* pErrBuf, int32_t len, bool isPartial) { + int32_t numOfParams = LIST_LENGTH(pFunc->pParameterList); + + if (isPartial) { + if (1 != numOfParams && 2 != numOfParams) { + return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName); + } + + uint8_t paraType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type; + if (TSDB_DATA_TYPE_TIMESTAMP != paraType) { + return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName); + } + + // param1 + if (2 == numOfParams) { + SNode* pParamNode1 = nodesListGetNode(pFunc->pParameterList, 1); + if (QUERY_NODE_VALUE != nodeType(pParamNode1)) { + return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName); + } + + SValueNode* pValue = (SValueNode*)pParamNode1; + + pValue->notReserved = true; + + paraType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 1))->resType.type; + if (!IS_INTEGER_TYPE(paraType)) { + return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName); + } + + if (pValue->datum.i == 0) { + return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR, + "ELAPSED function time unit parameter should be greater than db precision"); + } + } + + pFunc->node.resType = (SDataType){.bytes = getElapsedInfoSize() + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY}; + } else { + if (1 != numOfParams) { + return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName); + } + + uint8_t paraType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type; + if (TSDB_DATA_TYPE_BINARY != paraType) { + return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName); + } + pFunc->node.resType = (SDataType){.bytes = tDataTypes[TSDB_DATA_TYPE_DOUBLE].bytes, .type = TSDB_DATA_TYPE_DOUBLE}; + } + return TSDB_CODE_SUCCESS; +} + +static int32_t translateElapsedPartial(SFunctionNode* pFunc, char* pErrBuf, int32_t len) { + return translateElapsedImpl(pFunc, pErrBuf, len, true); +} + +static int32_t translateElapsedMerge(SFunctionNode* pFunc, char* pErrBuf, int32_t len) { + return translateElapsedImpl(pFunc, pErrBuf, len, false); +} + static int32_t translateLeastSQR(SFunctionNode* pFunc, char* pErrBuf, int32_t len) { int32_t numOfParams = LIST_LENGTH(pFunc->pParameterList); if (3 != numOfParams) { @@ -1421,6 +1479,28 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { .processFunc = elapsedFunction, .finalizeFunc = elapsedFinalize }, + { + .name = "_elapsed_partial", + .type = FUNCTION_TYPE_ELAPSED, + .classification = FUNC_MGT_AGG_FUNC, + .dataRequiredFunc = statisDataRequired, + .translateFunc = translateElapsedPartial, + .getEnvFunc = getElapsedFuncEnv, + .initFunc = elapsedFunctionSetup, + .processFunc = elapsedFunction, + .finalizeFunc = elapsedFinalize + }, + { + .name = "_elapsed_merge", + .type = FUNCTION_TYPE_ELAPSED, + .classification = FUNC_MGT_AGG_FUNC, + .dataRequiredFunc = statisDataRequired, + .translateFunc = translateElapsedMerge, + .getEnvFunc = getElapsedFuncEnv, + .initFunc = elapsedFunctionSetup, + .processFunc = elapsedFunction, + .finalizeFunc = elapsedFinalize + }, { .name = "last_row", .type = FUNCTION_TYPE_LAST_ROW, diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c index 6e969749c6..d202c8f565 100644 --- a/source/libs/function/src/builtinsimpl.c +++ b/source/libs/function/src/builtinsimpl.c @@ -3027,6 +3027,10 @@ int32_t spreadPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) { return pResInfo->numOfRes; } +int32_t getElapsedInfoSize() { + return (int32_t)sizeof(SElapsedInfo); +} + bool getElapsedFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) { pEnv->calcMemSize = sizeof(SElapsedInfo); return true; From d2efa06742194b13a3b94384bafe7bf93fda8e31 Mon Sep 17 00:00:00 2001 From: Ganlin Zhao Date: Thu, 9 Jun 2022 19:49:29 +0800 Subject: [PATCH 2/3] add elapsed function merge/partial functions --- source/libs/function/inc/builtinsimpl.h | 2 ++ source/libs/function/src/builtins.c | 4 +-- source/libs/function/src/builtinsimpl.c | 42 +++++++++++++++++++++++++ 3 files changed, 46 insertions(+), 2 deletions(-) diff --git a/source/libs/function/inc/builtinsimpl.h b/source/libs/function/inc/builtinsimpl.h index c4f6f0cc45..65d2457e5a 100644 --- a/source/libs/function/inc/builtinsimpl.h +++ b/source/libs/function/inc/builtinsimpl.h @@ -114,7 +114,9 @@ int32_t getSpreadInfoSize(); bool getElapsedFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv); bool elapsedFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo); int32_t elapsedFunction(SqlFunctionCtx* pCtx); +int32_t elapsedFunctionMerge(SqlFunctionCtx* pCtx); int32_t elapsedFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock); +int32_t elapsedPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock); int32_t getElapsedInfoSize(); bool getHistogramFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv); diff --git a/source/libs/function/src/builtins.c b/source/libs/function/src/builtins.c index 04cfa9d360..299e45b4b2 100644 --- a/source/libs/function/src/builtins.c +++ b/source/libs/function/src/builtins.c @@ -1488,7 +1488,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { .getEnvFunc = getElapsedFuncEnv, .initFunc = elapsedFunctionSetup, .processFunc = elapsedFunction, - .finalizeFunc = elapsedFinalize + .finalizeFunc = elapsedPartialFinalize }, { .name = "_elapsed_merge", @@ -1498,7 +1498,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { .translateFunc = translateElapsedMerge, .getEnvFunc = getElapsedFuncEnv, .initFunc = elapsedFunctionSetup, - .processFunc = elapsedFunction, + .processFunc = elapsedFunctionMerge, .finalizeFunc = elapsedFinalize }, { diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c index d202c8f565..bd8b6d0582 100644 --- a/source/libs/function/src/builtinsimpl.c +++ b/source/libs/function/src/builtinsimpl.c @@ -3132,6 +3132,30 @@ _elapsed_over: return TSDB_CODE_SUCCESS; } +int32_t elapsedFunctionMerge(SqlFunctionCtx *pCtx) { + SInputColumnInfoData* pInput = &pCtx->input; + SColumnInfoData* pCol = pInput->pData[0]; + ASSERT(pCol->info.type == TSDB_DATA_TYPE_BINARY); + + SElapsedInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx)); + + int32_t start = pInput->startRowIndex; + char* data = colDataGetData(pCol, start); + SElapsedInfo* pInputInfo = (SElapsedInfo *)varDataVal(data); + + pInfo->timeUnit = pInputInfo->timeUnit; + if (pInfo->min > pInputInfo->min) { + pInfo->min = pInputInfo->min; + } + + if (pInfo->max < pInputInfo->max) { + pInfo->max = pInputInfo->max; + } + + SET_VAL(GET_RES_INFO(pCtx), 1, 1); + return TSDB_CODE_SUCCESS; +} + int32_t elapsedFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) { SElapsedInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx)); double result = (double)pInfo->max - (double)pInfo->min; @@ -3140,6 +3164,24 @@ int32_t elapsedFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) { return functionFinalize(pCtx, pBlock); } +int32_t elapsedPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) { + SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx); + SElapsedInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx)); + int32_t resultBytes = getElapsedInfoSize(); + 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); + + colDataAppend(pCol, pBlock->info.rows, res, false); + + taosMemoryFree(res); + return pResInfo->numOfRes; +} + int32_t getHistogramInfoSize() { return (int32_t)sizeof(SHistoFuncInfo) + HISTOGRAM_MAX_BINS_NUM * sizeof(SHistoFuncBin); } From 2efbb569978bc85b94676297925ca7e8b7949758 Mon Sep 17 00:00:00 2001 From: Ganlin Zhao Date: Thu, 9 Jun 2022 20:09:37 +0800 Subject: [PATCH 3/3] enable elapsed function splitting --- source/libs/function/src/builtins.c | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/source/libs/function/src/builtins.c b/source/libs/function/src/builtins.c index 299e45b4b2..86b141ac54 100644 --- a/source/libs/function/src/builtins.c +++ b/source/libs/function/src/builtins.c @@ -1477,7 +1477,9 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { .getEnvFunc = getElapsedFuncEnv, .initFunc = elapsedFunctionSetup, .processFunc = elapsedFunction, - .finalizeFunc = elapsedFinalize + .finalizeFunc = elapsedFinalize, + .pPartialFunc = "_elapsed_partial", + .pMergeFunc = "_elapsed_merge" }, { .name = "_elapsed_partial",