From 84818987b3873a697dee8f959f03066e4ec6f77d Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Mon, 21 Mar 2022 15:25:13 +0800 Subject: [PATCH] [td-13039] fix bug in show tables; --- include/libs/executor/executor.h | 6 ++- source/libs/executor/src/executor.c | 19 ++++--- source/libs/function/inc/builtinsimpl.h | 6 ++- source/libs/function/src/builtins.c | 24 ++++----- source/libs/function/src/builtinsimpl.c | 71 ++++++++++++++++++++++++- 5 files changed, 103 insertions(+), 23 deletions(-) diff --git a/include/libs/executor/executor.h b/include/libs/executor/executor.h index d4af51fc21..6091456780 100644 --- a/include/libs/executor/executor.h +++ b/include/libs/executor/executor.h @@ -32,6 +32,9 @@ typedef struct SReadHandle { void* meta; } SReadHandle; +#define STREAM_DATA_TYPE_SUBMITBLK 0x1u +#define STREAM_DATA_TYPE_SSDATABLK 0x2u + /** * Create the exec task for streaming mode * @param pMsg @@ -44,9 +47,10 @@ qTaskInfo_t qCreateStreamExecTaskInfo(void *msg, void* streamReadHandle); * Set the input data block for the stream scan. * @param tinfo * @param input + * @param type * @return */ -int32_t qSetStreamInput(qTaskInfo_t tinfo, const void* input); +int32_t qSetStreamInput(qTaskInfo_t tinfo, const void* input, int32_t type); /** * Update the table id list, add or remove. diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c index a8602b7c77..3720e1c179 100644 --- a/source/libs/executor/src/executor.c +++ b/source/libs/executor/src/executor.c @@ -18,7 +18,7 @@ #include "planner.h" #include "tq.h" -static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, char* id) { +static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, int32_t type, char* id) { ASSERT(pOperator != NULL); if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) { if (pOperator->numOfDownstream == 0) { @@ -31,18 +31,23 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, char* id) return TSDB_CODE_QRY_APP_ERROR; } - return doSetStreamBlock(pOperator->pDownstream[0], input, id); + // TODO handle the join case + return doSetStreamBlock(pOperator->pDownstream[0], input, type, id); } else { SStreamBlockScanInfo* pInfo = pOperator->info; - if (tqReadHandleSetMsg(pInfo->readerHandle, input, 0) < 0) { - qError("submit msg messed up when initing stream block, %s" PRIx64, id); - return TSDB_CODE_QRY_APP_ERROR; + if (type == STREAM_DATA_TYPE_SUBMITBLK) { + if (tqReadHandleSetMsg(pInfo->readerHandle, input, 0) < 0) { + qError("submit msg messed up when initing stream block, %s" PRIx64, id); + return TSDB_CODE_QRY_APP_ERROR; + } + } else { // TODO + } return TSDB_CODE_SUCCESS; } } -int32_t qSetStreamInput(qTaskInfo_t tinfo, const void* input) { +int32_t qSetStreamInput(qTaskInfo_t tinfo, const void* input, int32_t type) { if (tinfo == NULL) { return TSDB_CODE_QRY_APP_ERROR; } @@ -53,7 +58,7 @@ int32_t qSetStreamInput(qTaskInfo_t tinfo, const void* input) { SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo; - int32_t code = doSetStreamBlock(pTaskInfo->pRoot, (void*)input, GET_TASKID(pTaskInfo)); + int32_t code = doSetStreamBlock(pTaskInfo->pRoot, (void*)input, type, GET_TASKID(pTaskInfo)); if (code != TSDB_CODE_SUCCESS) { qError("%s failed to set the stream block data", GET_TASKID(pTaskInfo)); } else { diff --git a/source/libs/function/inc/builtinsimpl.h b/source/libs/function/inc/builtinsimpl.h index 7ba7d7bdcc..e67215dfa5 100644 --- a/source/libs/function/inc/builtinsimpl.h +++ b/source/libs/function/inc/builtinsimpl.h @@ -23,7 +23,7 @@ extern "C" { #include "function.h" bool functionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo); -void functionFinalizer(SqlFunctionCtx *pCtx); +void functionFinalize(SqlFunctionCtx *pCtx); bool getCountFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv); void countFunction(SqlFunctionCtx *pCtx); @@ -37,6 +37,10 @@ bool getMinmaxFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv); void minFunction(SqlFunctionCtx* pCtx); void maxFunction(SqlFunctionCtx *pCtx); +bool getStddevFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv); +void stddevFunction(SqlFunctionCtx* pCtx); +void stddevFinalize(SqlFunctionCtx* pCtx); + #ifdef __cplusplus } #endif diff --git a/source/libs/function/src/builtins.c b/source/libs/function/src/builtins.c index 9f0a03b647..15f8ca2d10 100644 --- a/source/libs/function/src/builtins.c +++ b/source/libs/function/src/builtins.c @@ -29,7 +29,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { .getEnvFunc = getCountFuncEnv, .initFunc = functionSetup, .processFunc = countFunction, - .finalizeFunc = functionFinalizer + .finalizeFunc = functionFinalize }, { .name = "sum", @@ -39,7 +39,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { .getEnvFunc = getSumFuncEnv, .initFunc = functionSetup, .processFunc = sumFunction, - .finalizeFunc = functionFinalizer + .finalizeFunc = functionFinalize }, { .name = "min", @@ -49,7 +49,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { .getEnvFunc = getMinmaxFuncEnv, .initFunc = minFunctionSetup, .processFunc = minFunction, - .finalizeFunc = functionFinalizer + .finalizeFunc = functionFinalize }, { .name = "max", @@ -59,17 +59,17 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { .getEnvFunc = getMinmaxFuncEnv, .initFunc = maxFunctionSetup, .processFunc = maxFunction, - .finalizeFunc = functionFinalizer + .finalizeFunc = functionFinalize }, { .name = "stddev", .type = FUNCTION_TYPE_STDDEV, .classification = FUNC_MGT_AGG_FUNC, .checkFunc = stubCheckAndGetResultType, - .getEnvFunc = getMinmaxFuncEnv, + .getEnvFunc = getStddevFuncEnv, .initFunc = maxFunctionSetup, .processFunc = maxFunction, - .finalizeFunc = functionFinalizer + .finalizeFunc = functionFinalize }, { .name = "percentile", @@ -79,7 +79,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { .getEnvFunc = getMinmaxFuncEnv, .initFunc = maxFunctionSetup, .processFunc = maxFunction, - .finalizeFunc = functionFinalizer + .finalizeFunc = functionFinalize }, { .name = "apercentile", @@ -89,7 +89,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { .getEnvFunc = getMinmaxFuncEnv, .initFunc = maxFunctionSetup, .processFunc = maxFunction, - .finalizeFunc = functionFinalizer + .finalizeFunc = functionFinalize }, { .name = "top", @@ -99,7 +99,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { .getEnvFunc = getMinmaxFuncEnv, .initFunc = maxFunctionSetup, .processFunc = maxFunction, - .finalizeFunc = functionFinalizer + .finalizeFunc = functionFinalize }, { .name = "bottom", @@ -109,7 +109,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { .getEnvFunc = getMinmaxFuncEnv, .initFunc = maxFunctionSetup, .processFunc = maxFunction, - .finalizeFunc = functionFinalizer + .finalizeFunc = functionFinalize }, { .name = "spread", @@ -119,7 +119,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { .getEnvFunc = getMinmaxFuncEnv, .initFunc = maxFunctionSetup, .processFunc = maxFunction, - .finalizeFunc = functionFinalizer + .finalizeFunc = functionFinalize }, { .name = "last_row", @@ -129,7 +129,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { .getEnvFunc = getMinmaxFuncEnv, .initFunc = maxFunctionSetup, .processFunc = maxFunction, - .finalizeFunc = functionFinalizer + .finalizeFunc = functionFinalize }, { .name = "concat", diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c index f0f00434f0..fd3c022739 100644 --- a/source/libs/function/src/builtinsimpl.c +++ b/source/libs/function/src/builtinsimpl.c @@ -50,7 +50,7 @@ bool functionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo) { static void doFinalizer(SResultRowEntryInfo* pResInfo) { cleanupResultRowEntry(pResInfo); } -void functionFinalizer(SqlFunctionCtx *pCtx) { +void functionFinalize(SqlFunctionCtx *pCtx) { SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx); doFinalizer(pResInfo); } @@ -441,4 +441,71 @@ void minFunction(SqlFunctionCtx *pCtx) { void maxFunction(SqlFunctionCtx *pCtx) { int32_t numOfElems = doMinMaxHelper(pCtx, 0); SET_VAL(GET_RES_INFO(pCtx), numOfElems, 1); -} \ No newline at end of file +} + +typedef struct STopBotRes { + int32_t num; +} STopBotRes; + +bool getTopBotFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) { + SColumnNode* pColNode = (SColumnNode*) nodesListGetNode(pFunc->pParameterList, 0); + int32_t bytes = pColNode->node.resType.bytes; + SValueNode* pkNode = (SValueNode*) nodesListGetNode(pFunc->pParameterList, 1); + return true; +} + +typedef struct SStddevRes { + int64_t count; + union {double quadraticDSum; int64_t quadraticISum;}; + union {double dsum; int64_t isum;}; +} SStddevRes; + +bool getStddevFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) { + pEnv->calcMemSize = sizeof(SStddevRes); + return true; +} + +void stddevFunction(SqlFunctionCtx* pCtx) { + int32_t numOfElem = 0; + + // Only the pre-computing information loaded and actual data does not loaded + SInputColumnInfoData* pInput = &pCtx->input; + SColumnDataAgg *pAgg = pInput->pColumnDataAgg[0]; + int32_t type = pInput->pData[0]->info.type; + + SStddevRes* pStddevRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx)); + +// } else { // computing based on the true data block + SColumnInfoData* pCol = pInput->pData[0]; + + int32_t start = pInput->startRowIndex; + int32_t numOfRows = pInput->numOfRows; + + switch(type) { + case TSDB_DATA_TYPE_INT: { + int32_t* plist = (int32_t*)pCol->pData; + for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) { + if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) { + continue; + } + + pStddevRes->count += 1; + pStddevRes->isum += plist[i]; + pStddevRes->quadraticISum += plist[i] * plist[i]; + } + } + break; + } + + // data in the check operation are all null, not output + SET_VAL(GET_RES_INFO(pCtx), numOfElem, 1); +} + +void stddevFinalize(SqlFunctionCtx* pCtx) { + functionFinalize(pCtx); + + SStddevRes* pStddevRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx)); + double res = pStddevRes->quadraticISum/pStddevRes->count - (pStddevRes->isum / pStddevRes->count) * (pStddevRes->isum / pStddevRes->count); +} + +