xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(query): add avg function scalar version 

TD-17344
This commit is contained in:
Ganlin Zhao 2022-07-14 20:02:21 +08:00
parent 256c4aa25b
commit 7e661a9e9c
3 changed files with 98 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/libs/scalar/scalar.h
View File

@ -101,6 +101,7 @@ int32_t countScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam
int32_t sumScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput); int32_t sumScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
int32_t minScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput); int32_t minScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
int32_t maxScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput); int32_t maxScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
int32_t avgScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
int32_t stddevScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput); int32_t stddevScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
#ifdef __cplusplus #ifdef __cplusplus

1
source/libs/function/src/builtins.c
View File

@ -1999,6 +1999,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
.getEnvFunc = getAvgFuncEnv, .getEnvFunc = getAvgFuncEnv,
.initFunc = avgFunctionSetup, .initFunc = avgFunctionSetup,
.processFunc = avgFunction, .processFunc = avgFunction,
.sprocessFunc = avgScalarFunction,
.finalizeFunc = avgFinalize, .finalizeFunc = avgFinalize,
.invertFunc = avgInvertFunction, .invertFunc = avgInvertFunction,
.combineFunc = avgCombine, .combineFunc = avgCombine,

96
source/libs/scalar/src/sclfunc.c
View File

@ -1919,6 +1919,101 @@ int32_t maxScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *
return doMinMaxScalarFunction(pInput, inputNum, pOutput, false); return doMinMaxScalarFunction(pInput, inputNum, pOutput, false);
} }
int32_t avgScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
SColumnInfoData *pInputData = pInput->columnData;
SColumnInfoData *pOutputData = pOutput->columnData;
int32_t type = GET_PARAM_TYPE(pInput);
int64_t count = 0, sum = 0;
bool hasNull = false;
for (int32_t i = 0; i < pInput->numOfRows; ++i) {
if (colDataIsNull_s(pInputData, i)) {
hasNull = true;
break;
}
switch(type) {
case TSDB_DATA_TYPE_TINYINT: {
int8_t *in = (int8_t *)pInputData->pData;
sum += in[i];
count++;
break;
}
case TSDB_DATA_TYPE_SMALLINT: {
int16_t *in = (int16_t *)pInputData->pData;
sum += in[i];
count++;
break;
}
case TSDB_DATA_TYPE_INT: {
int32_t *in = (int32_t *)pInputData->pData;
sum += in[i];
count++;
break;
}
case TSDB_DATA_TYPE_BIGINT: {
int64_t *in = (int64_t *)pInputData->pData;
sum += in[i];
count++;
break;
}
case TSDB_DATA_TYPE_UTINYINT: {
uint8_t *in = (uint8_t *)pInputData->pData;
sum += in[i];
count++;
break;
}
case TSDB_DATA_TYPE_USMALLINT: {
uint16_t *in = (uint16_t *)pInputData->pData;
sum += in[i];
count++;
break;
}
case TSDB_DATA_TYPE_UINT: {
uint32_t *in = (uint32_t *)pInputData->pData;
sum += in[i];
count++;
break;
}
case TSDB_DATA_TYPE_UBIGINT: {
uint64_t *in = (uint64_t *)pInputData->pData;
sum += in[i];
count++;
break;
}
case TSDB_DATA_TYPE_FLOAT: {
float *in = (float *)pInputData->pData;
sum += in[i];
count++;
break;
}
case TSDB_DATA_TYPE_DOUBLE: {
double *in = (double *)pInputData->pData;
sum += in[i];
count++;
break;
}
}
}
double *out = (double *)pOutputData->pData;
if (hasNull) {
colDataAppendNULL(pOutputData, 0);
} else {
if (IS_SIGNED_NUMERIC_TYPE(type)) {
*out = (int64_t)sum / (double)count;
} else if (IS_UNSIGNED_NUMERIC_TYPE(type)) {
*out = (uint64_t)sum / (double)count;
} else if (IS_FLOAT_TYPE(type)) {
*out = (double)sum / (double)count;
}
}
pOutput->numOfRows = 1;
return TSDB_CODE_SUCCESS;
}
int32_t stddevScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) { int32_t stddevScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
SColumnInfoData *pInputData = pInput->columnData; SColumnInfoData *pInputData = pInput->columnData;
SColumnInfoData *pOutputData = pOutput->columnData; SColumnInfoData *pOutputData = pOutput->columnData;
@ -2031,3 +2126,4 @@ int32_t stddevScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarPara
pOutput->numOfRows = 1; pOutput->numOfRows = 1;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }