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

[td-13039] support scalar function.

This commit is contained in:
Haojun Liao 2022-03-27 22:43:07 +08:00
parent 6cd2d69d14
commit 0daa0de073
10 changed files with 207 additions and 497 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
include/libs/function/function.h
View File

@ -27,16 +27,22 @@ extern "C" {
struct SqlFunctionCtx;
struct SResultRowEntryInfo;
typedef struct SFunctionNode SFunctionNode;
struct SFunctionNode;
typedef struct SScalarParam SScalarParam;
typedef struct SFuncExecEnv {
int32_t calcMemSize;
} SFuncExecEnv;
typedef bool (*FExecGetEnv)(SFunctionNode* pFunc, SFuncExecEnv* pEnv);
typedef bool (*FExecGetEnv)(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
typedef bool (*FExecInit)(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResultCellInfo);
typedef void (*FExecProcess)(struct SqlFunctionCtx *pCtx);
typedef void (*FExecFinalize)(struct SqlFunctionCtx *pCtx);
typedef int32_t (*FScalarExecProcess)(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
typedef struct SScalarFuncExecFuncs {
FScalarExecProcess process;
} SScalarFuncExecFuncs;
typedef struct SFuncExecFuncs {
FExecGetEnv getEnv;
@ -191,6 +197,7 @@ typedef struct SqlFunctionCtx {
SPoint1 start;
SPoint1 end;
SFuncExecFuncs fpSet;
SScalarFuncExecFuncs sfp;
} SqlFunctionCtx;
enum {
@ -219,7 +226,7 @@ typedef struct tExprNode {
char functionName[FUNCTIONS_NAME_MAX_LENGTH]; // todo refactor
int32_t functionId;
int32_t num;
SFunctionNode *pFunctNode;
struct SFunctionNode *pFunctNode;
// Note that the attribute of pChild is not the parameter of function, it is the columns that involved in the
// calculation instead.
// E.g., Cov(col1, col2), the column information, w.r.t. the col1 and col2, is kept in pChild nodes.
@ -254,18 +261,11 @@ typedef struct SAggFunctionInfo {
int32_t (*dataReqFunc)(SqlFunctionCtx *pCtx, STimeWindow* w, int32_t colId);
} SAggFunctionInfo;
typedef struct SScalarParam {
struct SScalarParam {
SColumnInfoData *columnData;
SHashObj *pHashFilter;
int32_t numOfRows;
} SScalarParam;
typedef struct SScalarFunctionInfo {
char name[FUNCTIONS_NAME_MAX_LENGTH];
int8_t type; // scalar function or aggregation function
uint32_t functionId; // index of scalar function
void (*process)(struct SScalarParam* pOutput, size_t numOfInput, const struct SScalarParam *pInput);
} SScalarFunctionInfo;
};
typedef struct SMultiFunctionsDesc {
bool stableQuery;

7
include/libs/function/functionMgt.h
View File

@ -103,13 +103,6 @@ struct SqlFunctionCtx;
struct SResultRowEntryInfo;
struct STimeWindow;
typedef int32_t (*FScalarExecProcess)(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
typedef struct SScalarFuncExecFuncs {
FScalarExecProcess process;
} SScalarFuncExecFuncs;
int32_t fmFuncMgtInit();
void fmFuncMgtDestroy();

27
source/libs/executor/src/executorimpl.c
View File

@ -1257,10 +1257,25 @@ static void projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSData
SArray* pBlockList = taosArrayInit(4, POINTER_BYTES);
taosArrayPush(pBlockList, &pSrcBlock);
SScalarParam dest = {0};
dest.columnData = taosArrayGet(pResult->pDataBlock, k);
scalarCalculate(pExpr[k].pExpr->_optrRoot.pRootNode, pBlockList, &dest);
pResult->info.rows = dest.numOfRows;
taosArrayDestroy(pBlockList);
} else if (pExpr[k].pExpr->nodeType == QUERY_NODE_FUNCTION) {
ASSERT(!fmIsAggFunc(pCtx->functionId));
SScalarParam p = {.numOfRows = pSrcBlock->info.rows};
p.columnData = taosArrayGet(pResult->pDataBlock, k);
scalarCalculate(pExpr[k].pExpr->_optrRoot.pRootNode, pBlockList, &p);
pResult->info.rows = p.numOfRows;
int32_t slotId = pExpr[k].base.pParam[0].pCol->slotId;
p.columnData = taosArrayGet(pSrcBlock->pDataBlock, slotId);
SScalarParam dest = {0};
dest.columnData = taosArrayGet(pResult->pDataBlock, k);
pCtx[k].sfp.process(&p, 1, &dest);
pResult->info.rows = dest.numOfRows;
} else {
ASSERT(0);
}
@ -2044,8 +2059,12 @@ static SqlFunctionCtx* createSqlFunctionCtx_rv(SExprInfo* pExprInfo, int32_t num
SFuncExecEnv env = {0};
pCtx->functionId = pExpr->pExpr->_function.pFunctNode->funcId;
if (fmIsAggFunc(pCtx->functionId)) {
fmGetFuncExecFuncs(pCtx->functionId, &pCtx->fpSet);
pCtx->fpSet.getEnv(pExpr->pExpr->_function.pFunctNode, &env);
} else {
fmGetScalarFuncExecFuncs(pCtx->functionId, &pCtx->sfp);
}
pCtx->resDataInfo.interBufSize = env.calcMemSize;
} else if (pExpr->pExpr->nodeType == QUERY_NODE_COLUMN) {
@ -6620,6 +6639,8 @@ static SSDataBlock* doProjectOperation(SOperatorInfo *pOperator, bool* newgroup)
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pOperator, pInfo->pCtx, pBlock, TSDB_ORDER_ASC);
blockDataEnsureCapacity(pInfo->pRes, pInfo->pRes->info.rows + pBlock->info.rows);
projectApplyFunctions(pOperator->pExpr, pInfo->pRes, pBlock, pInfo->pCtx, pOperator->numOfOutput);
if (pRes->info.rows >= pOperator->resultInfo.threshold) {
break;

10
source/libs/function/inc/builtinsimpl.h
View File

@ -25,23 +25,23 @@ extern "C" {
bool functionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
void functionFinalize(SqlFunctionCtx *pCtx);
bool getCountFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool getCountFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
void countFunction(SqlFunctionCtx *pCtx);
bool getSumFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool getSumFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
void sumFunction(SqlFunctionCtx *pCtx);
bool minFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
bool maxFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
bool getMinmaxFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool getMinmaxFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
void minFunction(SqlFunctionCtx* pCtx);
void maxFunction(SqlFunctionCtx *pCtx);
bool getStddevFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool getStddevFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
void stddevFunction(SqlFunctionCtx* pCtx);
void stddevFinalize(SqlFunctionCtx* pCtx);
bool getFirstLastFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool getFirstLastFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
void firstFunction(SqlFunctionCtx *pCtx);
void lastFunction(SqlFunctionCtx *pCtx);

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

@ -331,6 +331,14 @@ int32_t stubCheckAndGetResultType(SFunctionNode* pFunc) {
case FUNCTION_TYPE_CONCAT:
// todo
break;
case FUNCTION_TYPE_ABS: {
SColumnNode* pParam = nodesListGetNode(pFunc->pParameterList, 0);
int32_t paraType = pParam->node.resType.type;
pFunc->node.resType = (SDataType) { .bytes = tDataTypes[paraType].bytes, .type = paraType };
break;
}
default:
ASSERT(0); // to found the fault ASAP.
}

12
source/libs/scalar/inc/sclfunc.h
View File

@ -22,19 +22,7 @@ extern "C" {
#include "function.h"
#include "scalar.h"
typedef struct SScalarFunctionSupport {
struct SExprInfo *pExprInfo;
int32_t numOfCols;
SColumnInfo *colList;
void *exprList; // client side used
int32_t offset;
char** data;
} SScalarFunctionSupport;
extern struct SScalarFunctionInfo scalarFunc[8];
int32_t evaluateExprNodeTree(tExprNode* pExprs, int32_t numOfRows, SScalarParam* pOutput,
void* param, char* (*getSourceDataBlock)(void*, const char*, int32_t));
#ifdef __cplusplus

60
source/libs/scalar/inc/sclvector.h
View File

@ -20,10 +20,66 @@
extern "C" {
#endif
#include "sclfunc.h"
typedef double (*_getDoubleValue_fn_t)(void *src, int32_t index);
typedef double (*_mathFunc)(double, double, bool *);
static FORCE_INLINE double getVectorDoubleValue_TINYINT(void *src, int32_t index) {
return (double)*((int8_t *)src + index);
}
static FORCE_INLINE double getVectorDoubleValue_UTINYINT(void *src, int32_t index) {
return (double)*((uint8_t *)src + index);
}
static FORCE_INLINE double getVectorDoubleValue_SMALLINT(void *src, int32_t index) {
return (double)*((int16_t *)src + index);
}
static FORCE_INLINE double getVectorDoubleValue_USMALLINT(void *src, int32_t index) {
return (double)*((uint16_t *)src + index);
}
static FORCE_INLINE double getVectorDoubleValue_INT(void *src, int32_t index) {
return (double)*((int32_t *)src + index);
}
static FORCE_INLINE double getVectorDoubleValue_UINT(void *src, int32_t index) {
return (double)*((uint32_t *)src + index);
}
static FORCE_INLINE double getVectorDoubleValue_BIGINT(void *src, int32_t index) {
return (double)*((int64_t *)src + index);
}
static FORCE_INLINE double getVectorDoubleValue_UBIGINT(void *src, int32_t index) {
return (double)*((uint64_t *)src + index);
}
static FORCE_INLINE double getVectorDoubleValue_FLOAT(void *src, int32_t index) {
return (double)*((float *)src + index);
}
static FORCE_INLINE double getVectorDoubleValue_DOUBLE(void *src, int32_t index) {
return (double)*((double *)src + index);
}
static FORCE_INLINE _getDoubleValue_fn_t getVectorDoubleValueFn(int32_t srcType) {
_getDoubleValue_fn_t p = NULL;
if (srcType == TSDB_DATA_TYPE_TINYINT) {
p = getVectorDoubleValue_TINYINT;
} else if (srcType == TSDB_DATA_TYPE_UTINYINT) {
p = getVectorDoubleValue_UTINYINT;
} else if (srcType == TSDB_DATA_TYPE_SMALLINT) {
p = getVectorDoubleValue_SMALLINT;
} else if (srcType == TSDB_DATA_TYPE_USMALLINT) {
p = getVectorDoubleValue_USMALLINT;
} else if (srcType == TSDB_DATA_TYPE_INT) {
p = getVectorDoubleValue_INT;
} else if (srcType == TSDB_DATA_TYPE_UINT) {
p = getVectorDoubleValue_UINT;
} else if (srcType == TSDB_DATA_TYPE_BIGINT) {
p = getVectorDoubleValue_BIGINT;
} else if (srcType == TSDB_DATA_TYPE_UBIGINT) {
p = getVectorDoubleValue_UBIGINT;
} else if (srcType == TSDB_DATA_TYPE_FLOAT) {
p = getVectorDoubleValue_FLOAT;
} else if (srcType == TSDB_DATA_TYPE_DOUBLE) {
p = getVectorDoubleValue_DOUBLE;
} else {
assert(0);
}
return p;
}
typedef void (*_bufConverteFunc)(char *buf, SScalarParam* pOut, int32_t outType);
typedef void (*_bin_scalar_fn_t)(SScalarParam* pLeft, SScalarParam* pRight, SScalarParam *output, int32_t order);

19
source/libs/scalar/src/scalar.c
View File

@ -6,6 +6,7 @@
#include "sclvector.h"
#include "tcommon.h"
#include "tdatablock.h"
#include "scalar.h"
int32_t scalarGetOperatorParamNum(EOperatorType type) {
if (OP_TYPE_IS_NULL == type || OP_TYPE_IS_NOT_NULL == type || OP_TYPE_IS_TRUE == type || OP_TYPE_IS_NOT_TRUE == type
@ -168,7 +169,6 @@ int32_t sclInitParam(SNode* node, SScalarParam *param, SScalarCtx *ctx, int32_t
}
break;
}
case QUERY_NODE_NODE_LIST: {
SNodeListNode *nodeList = (SNodeListNode *)node;
if (LIST_LENGTH(nodeList->pNodeList) <= 0) {
@ -207,8 +207,9 @@ int32_t sclInitParam(SNode* node, SScalarParam *param, SScalarCtx *ctx, int32_t
param->columnData = columnData;
break;
}
case QUERY_NODE_LOGIC_CONDITION:
case QUERY_NODE_OPERATOR: {
case QUERY_NODE_FUNCTION:
case QUERY_NODE_OPERATOR:
case QUERY_NODE_LOGIC_CONDITION: {
SScalarParam *res = (SScalarParam *)taosHashGet(ctx->pRes, &node, POINTER_BYTES);
if (NULL == res) {
sclError("no result for node, type:%d, node:%p", nodeType(node), node);
@ -217,7 +218,6 @@ int32_t sclInitParam(SNode* node, SScalarParam *param, SScalarCtx *ctx, int32_t
*param = *res;
break;
}
default:
break;
}
@ -305,21 +305,18 @@ int32_t sclExecFuncion(SFunctionNode *node, SScalarCtx *ctx, SScalarParam *outpu
int32_t rowNum = 0;
SCL_ERR_RET(sclInitParamList(&params, node->pParameterList, ctx, &rowNum));
output->columnData->info.type = node->node.resType.type;
output->columnData->info.bytes = tDataTypes[node->node.resType.type].bytes;
code = blockDataEnsureColumnCapacity(output->columnData, rowNum);
if (code != TSDB_CODE_SUCCESS) {
output->columnData = createColumnInfoData(&node->node.resType, rowNum);
if (output->columnData == NULL) {
sclError("calloc %d failed", (int32_t)(rowNum * output->columnData->info.bytes));
SCL_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
for (int32_t i = 0; i < rowNum; ++i) {
// for (int32_t i = 0; i < rowNum; ++i) {
code = (*ffpSet.process)(params, node->pParameterList->length, output);
if (code) {
sclError("scalar function exec failed, funcId:%d, code:%s", node->funcId, tstrerror(code));
SCL_ERR_JRET(code);
}
// }
}
_return:

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

@ -23,9 +23,9 @@ int32_t absFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutpu
case TSDB_DATA_TYPE_FLOAT: {
float *in = (float *)pInputData->pData;
float *out = (float *)pOutputData->pData;
for (int32_t i = 0; i < pOutput->numOfRows; ++i) {
for (int32_t i = 0; i < pInput->numOfRows; ++i) {
if (colDataIsNull_f(pInputData->nullbitmap, i)) {
colDataAppendNULL(pOutputData, i);
colDataSetNull_f(pOutputData->nullbitmap, i);
continue;
}
out[i] = (in[i] > 0)? in[i] : -in[i];
@ -36,9 +36,9 @@ int32_t absFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutpu
case TSDB_DATA_TYPE_DOUBLE: {
double *in = (double *)pInputData->pData;
double *out = (double *)pOutputData->pData;
for (int32_t i = 0; i < pOutput->numOfRows; ++i) {
for (int32_t i = 0; i < pInput->numOfRows; ++i) {
if (colDataIsNull_f(pInputData->nullbitmap, i)) {
colDataAppendNULL(pOutputData, i);
colDataSetNull_f(pOutputData->nullbitmap, i);
continue;
}
out[i] = (in[i] > 0)? in[i] : -in[i];
@ -49,9 +49,9 @@ int32_t absFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutpu
case TSDB_DATA_TYPE_TINYINT: {
int8_t *in = (int8_t *)pInputData->pData;
int8_t *out = (int8_t *)pOutputData->pData;
for (int32_t i = 0; i < pOutput->numOfRows; ++i) {
for (int32_t i = 0; i < pInput->numOfRows; ++i) {
if (colDataIsNull_f(pInputData->nullbitmap, i)) {
colDataAppendNULL(pOutputData, i);
colDataSetNull_f(pOutputData->nullbitmap, i);
continue;
}
out[i] = (in[i] > 0)? in[i] : -in[i];
@ -62,9 +62,9 @@ int32_t absFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutpu
case TSDB_DATA_TYPE_SMALLINT: {
int16_t *in = (int16_t *)pInputData->pData;
int16_t *out = (int16_t *)pOutputData->pData;
for (int32_t i = 0; i < pOutput->numOfRows; ++i) {
for (int32_t i = 0; i < pInput->numOfRows; ++i) {
if (colDataIsNull_f(pInputData->nullbitmap, i)) {
colDataAppendNULL(pOutputData, i);
colDataSetNull_f(pOutputData->nullbitmap, i);
continue;
}
out[i] = (in[i] > 0)? in[i] : -in[i];
@ -75,10 +75,9 @@ int32_t absFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutpu
case TSDB_DATA_TYPE_INT: {
int32_t *in = (int32_t *)pInputData->pData;
int32_t *out = (int32_t *)pOutputData->pData;
for (int32_t i = 0; i < pOutput->numOfRows; ++i) {
for (int32_t i = 0; i < pInput->numOfRows; ++i) {
if (colDataIsNull_f(pInputData->nullbitmap, i)) {
colDataAppendNULL(pOutputData, i);
colDataSetNull_f(pOutputData->nullbitmap, i);
continue;
}
out[i] = (in[i] > 0)? in[i] : -in[i];
@ -89,9 +88,9 @@ int32_t absFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutpu
case TSDB_DATA_TYPE_BIGINT: {
int64_t *in = (int64_t *)pInputData->pData;
int64_t *out = (int64_t *)pOutputData->pData;
for (int32_t i = 0; i < pOutput->numOfRows; ++i) {
for (int32_t i = 0; i < pInput->numOfRows; ++i) {
if (colDataIsNull_f(pInputData->nullbitmap, i)) {
colDataAppendNULL(pOutputData, i);
colDataSetNull_f(pOutputData->nullbitmap, i);
continue;
}
out[i] = (in[i] > 0)? in[i] : -in[i];
@ -194,371 +193,119 @@ int32_t powFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutpu
return TSDB_CODE_SUCCESS;
}
int32_t sqrtFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
#if 0
if (inputNum != 1 || !IS_NUMERIC_TYPE(pInput->type)) {
typedef float (*_float_fn)(float);
typedef double (*_double_fn)(double);
int32_t doScalarFunctionUnique(SScalarParam *pInput, int32_t inputNum, SScalarParam* pOutput, _double_fn valFn) {
int32_t type = GET_PARAM_TYPE(pInput);
if (inputNum != 1 || !IS_NUMERIC_TYPE(type)) {
return TSDB_CODE_FAILED;
}
pOutput->type = TSDB_DATA_TYPE_DOUBLE;
pOutput->bytes = tDataTypes[TSDB_DATA_TYPE_DOUBLE].bytes;
SColumnInfoData *pInputData = pInput->columnData;
SColumnInfoData *pOutputData = pOutput->columnData;
char *input = NULL, *output = NULL;
for (int32_t i = 0; i < pOutput->num; ++i) {
if (pInput->num == 1) {
input = pInput->data;
} else {
input = pInput->data + i * pInput->bytes;
}
output = pOutput->data + i * pOutput->bytes;
_getDoubleValue_fn_t getValueFn = getVectorDoubleValueFn(type);
if (isNull(input, pInput->type)) {
setNull(output, pOutput->type, pOutput->bytes);
double *out = (double *)pOutputData->pData;
for (int32_t i = 0; i < pInput->numOfRows; ++i) {
if (colDataIsNull_f(pInputData->nullbitmap, i)) {
colDataSetNull_f(pOutputData->nullbitmap, i);
continue;
}
double v;
GET_TYPED_DATA(v, double, pInput->type, input);
double result = sqrt(v);
SET_TYPED_DATA(output, pOutput->type, result);
out[i] = valFn(getValueFn(pInputData->pData, i));
}
#endif
pOutput->numOfRows = pInput->numOfRows;
return TSDB_CODE_SUCCESS;
}
int32_t sinFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
#if 0
if (inputNum != 1 || !IS_NUMERIC_TYPE(pInput->type)) {
int32_t doScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam* pOutput, _float_fn f1, _double_fn d1) {
int32_t type = GET_PARAM_TYPE(pInput);
if (inputNum != 1 || !IS_NUMERIC_TYPE(type)) {
return TSDB_CODE_FAILED;
}
pOutput->type = TSDB_DATA_TYPE_DOUBLE;
pOutput->bytes = tDataTypes[TSDB_DATA_TYPE_DOUBLE].bytes;
SColumnInfoData *pInputData = pInput->columnData;
SColumnInfoData *pOutputData = pOutput->columnData;
char *input = NULL, *output = NULL;
for (int32_t i = 0; i < pOutput->num; ++i) {
if (pInput->num == 1) {
input = pInput->data;
} else {
input = pInput->data + i * pInput->bytes;
}
output = pOutput->data + i * pOutput->bytes;
switch (type) {
case TSDB_DATA_TYPE_FLOAT: {
float *in = (float *)pInputData->pData;
float *out = (float *)pOutputData->pData;
if (isNull(input, pInput->type)) {
setNull(output, pOutput->type, pOutput->bytes);
for (int32_t i = 0; i < pInput->numOfRows; ++i) {
if (colDataIsNull_f(pInputData->nullbitmap, i)) {
colDataSetNull_f(pOutputData->nullbitmap, i);
continue;
}
double v;
GET_TYPED_DATA(v, double, pInput->type, input);
double result = sin(v);
SET_TYPED_DATA(output, pOutput->type, result);
out[i] = f1(in[i]);
}
#endif
return TSDB_CODE_SUCCESS;
}
int32_t cosFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
#if 0
if (inputNum != 1 || !IS_NUMERIC_TYPE(pInput->type)) {
return TSDB_CODE_FAILED;
break;
}
pOutput->type = TSDB_DATA_TYPE_DOUBLE;
pOutput->bytes = tDataTypes[TSDB_DATA_TYPE_DOUBLE].bytes;
case TSDB_DATA_TYPE_DOUBLE: {
double *in = (double *)pInputData->pData;
double *out = (double *)pOutputData->pData;
char *input = NULL, *output = NULL;
for (int32_t i = 0; i < pOutput->num; ++i) {
if (pInput->num == 1) {
input = pInput->data;
} else {
input = pInput->data + i * pInput->bytes;
}
output = pOutput->data + i * pOutput->bytes;
if (isNull(input, pInput->type)) {
setNull(output, pOutput->type, pOutput->bytes);
for (int32_t i = 0; i < pInput->numOfRows; ++i) {
if (colDataIsNull_f(pInputData->nullbitmap, i)) {
colDataSetNull_f(pOutputData->nullbitmap, i);
continue;
}
double v;
GET_TYPED_DATA(v, double, pInput->type, input);
double result = cos(v);
SET_TYPED_DATA(output, pOutput->type, result);
out[i] = d1(in[i]);
}
#endif
return TSDB_CODE_SUCCESS;
}
int32_t tanFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
#if 0
if (inputNum != 1 || !IS_NUMERIC_TYPE(pInput->type)) {
return TSDB_CODE_FAILED;
break;
}
pOutput->type = TSDB_DATA_TYPE_DOUBLE;
pOutput->bytes = tDataTypes[TSDB_DATA_TYPE_DOUBLE].bytes;
char *input = NULL, *output = NULL;
for (int32_t i = 0; i < pOutput->num; ++i) {
if (pInput->num == 1) {
input = pInput->data;
} else {
input = pInput->data + i * pInput->bytes;
default: {
colDataAssign(pOutputData, pInputData, pInput->numOfRows);
}
output = pOutput->data + i * pOutput->bytes;
if (isNull(input, pInput->type)) {
setNull(output, pOutput->type, pOutput->bytes);
continue;
}
double v;
GET_TYPED_DATA(v, double, pInput->type, input);
double result = tan(v);
SET_TYPED_DATA(output, pOutput->type, result);
}
#endif
return TSDB_CODE_SUCCESS;
}
int32_t asinFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
#if 0
if (inputNum != 1 || !IS_NUMERIC_TYPE(pInput->type)) {
return TSDB_CODE_FAILED;
}
pOutput->type = TSDB_DATA_TYPE_DOUBLE;
pOutput->bytes = tDataTypes[TSDB_DATA_TYPE_DOUBLE].bytes;
char *input = NULL, *output = NULL;
for (int32_t i = 0; i < pOutput->num; ++i) {
if (pInput->num == 1) {
input = pInput->data;
} else {
input = pInput->data + i * pInput->bytes;
}
output = pOutput->data + i * pOutput->bytes;
if (isNull(input, pInput->type)) {
setNull(output, pOutput->type, pOutput->bytes);
continue;
}
double v;
GET_TYPED_DATA(v, double, pInput->type, input);
double result = asin(v);
SET_TYPED_DATA(output, pOutput->type, result);
}
#endif
return TSDB_CODE_SUCCESS;
}
int32_t acosFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
#if 0
if (inputNum != 1 || !IS_NUMERIC_TYPE(pInput->type)) {
return TSDB_CODE_FAILED;
}
pOutput->type = TSDB_DATA_TYPE_DOUBLE;
pOutput->bytes = tDataTypes[TSDB_DATA_TYPE_DOUBLE].bytes;
char *input = NULL, *output = NULL;
for (int32_t i = 0; i < pOutput->num; ++i) {
if (pInput->num == 1) {
input = pInput->data;
} else {
input = pInput->data + i * pInput->bytes;
}
output = pOutput->data + i * pOutput->bytes;
if (isNull(input, pInput->type)) {
setNull(output, pOutput->type, pOutput->bytes);
continue;
}
double v;
GET_TYPED_DATA(v, double, pInput->type, input);
double result = acos(v);
SET_TYPED_DATA(output, pOutput->type, result);
}
#endif
pOutput->numOfRows = pInput->numOfRows;
return TSDB_CODE_SUCCESS;
}
int32_t atanFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
#if 0
if (inputNum != 1 || !IS_NUMERIC_TYPE(pInput->type)) {
return TSDB_CODE_FAILED;
}
pOutput->type = TSDB_DATA_TYPE_DOUBLE;
pOutput->bytes = tDataTypes[TSDB_DATA_TYPE_DOUBLE].bytes;
char *input = NULL, *output = NULL;
for (int32_t i = 0; i < pOutput->num; ++i) {
if (pInput->num == 1) {
input = pInput->data;
} else {
input = pInput->data + i * pInput->bytes;
}
output = pOutput->data + i * pOutput->bytes;
if (isNull(input, pInput->type)) {
setNull(output, pOutput->type, pOutput->bytes);
continue;
}
double v;
GET_TYPED_DATA(v, double, pInput->type, input);
double result = atan(v);
SET_TYPED_DATA(output, pOutput->type, result);
}
#endif
return TSDB_CODE_SUCCESS;
return doScalarFunctionUnique(pInput, inputNum, pOutput, atan);
}
int32_t sinFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
return doScalarFunctionUnique(pInput, inputNum, pOutput, sin);
}
int32_t cosFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
return doScalarFunctionUnique(pInput, inputNum, pOutput, cos);
}
int32_t tanFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
return doScalarFunctionUnique(pInput, inputNum, pOutput, tan);
}
int32_t asinFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
return doScalarFunctionUnique(pInput, inputNum, pOutput, asin);
}
int32_t acosFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
return doScalarFunctionUnique(pInput, inputNum, pOutput, acos);
}
int32_t sqrtFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
return doScalarFunctionUnique(pInput, inputNum, pOutput, sqrt);
}
//TODO use callback function [ceilf, ceil]
int32_t ceilFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
int32_t type = GET_PARAM_TYPE(pInput);
if (inputNum != 1 || !IS_NUMERIC_TYPE(type)) {
return TSDB_CODE_FAILED;
}
SColumnInfoData *pInputData = pInput->columnData;
SColumnInfoData *pOutputData = pOutput->columnData;
switch (type) {
case TSDB_DATA_TYPE_FLOAT: {
float *in = (float *)pInputData->pData;
float *out = (float *)pOutputData->pData;
for (int32_t i = 0; i < pOutput->numOfRows; ++i) {
if (colDataIsNull_f(pInputData->nullbitmap, i)) {
colDataAppendNULL(pOutputData, i);
continue;
}
out[i] = ceilf(in[i]);
}
break;
}
case TSDB_DATA_TYPE_DOUBLE: {
double *in = (double *)pInputData->pData;
double *out = (double *)pOutputData->pData;
for (int32_t i = 0; i < pOutput->numOfRows; ++i) {
if (colDataIsNull_f(pInputData->nullbitmap, i)) {
colDataAppendNULL(pOutputData, i);
continue;
}
out[i] = ceil(in[i]);
}
break;
}
default: {
colDataAssign(pOutputData, pInputData, pInput->numOfRows);
}
}
pOutput->numOfRows = pInput->numOfRows;
return TSDB_CODE_SUCCESS;
return doScalarFunction(pInput, inputNum, pOutput, ceilf, ceil);
}
int32_t floorFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
int32_t type = GET_PARAM_TYPE(pInput);
if (inputNum != 1 || !IS_NUMERIC_TYPE(type)) {
return TSDB_CODE_FAILED;
}
SColumnInfoData *pInputData = pInput->columnData;
SColumnInfoData *pOutputData = pOutput->columnData;
switch (type) {
case TSDB_DATA_TYPE_FLOAT: {
float *in = (float *)pInputData->pData;
float *out = (float *)pOutputData->pData;
for (int32_t i = 0; i < pOutput->numOfRows; ++i) {
if (colDataIsNull_f(pInputData->nullbitmap, i)) {
colDataAppendNULL(pOutputData, i);
continue;
}
out[i] = floorf(in[i]);
}
break;
}
case TSDB_DATA_TYPE_DOUBLE: {
double *in = (double *)pInputData->pData;
double *out = (double *)pOutputData->pData;
for (int32_t i = 0; i < pOutput->numOfRows; ++i) {
if (colDataIsNull_f(pInputData->nullbitmap, i)) {
colDataAppendNULL(pOutputData, i);
continue;
}
out[i] = floor(in[i]);
}
break;
}
default: {
colDataAssign(pOutputData, pInputData, pInput->numOfRows);
}
}
pOutput->numOfRows = pInput->numOfRows;
return TSDB_CODE_SUCCESS;
return doScalarFunction(pInput, inputNum, pOutput, floorf, floor);
}
int32_t roundFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
int32_t type = GET_PARAM_TYPE(pInput);
if (inputNum != 1 || !IS_NUMERIC_TYPE(type)) {
return TSDB_CODE_FAILED;
}
SColumnInfoData *pInputData = pInput->columnData;
SColumnInfoData *pOutputData = pOutput->columnData;
switch (type) {
case TSDB_DATA_TYPE_FLOAT: {
float *in = (float *)pInputData->pData;
float *out = (float *)pOutputData->pData;
for (int32_t i = 0; i < pOutput->numOfRows; ++i) {
if (colDataIsNull_f(pInputData->nullbitmap, i)) {
colDataAppendNULL(pOutputData, i);
continue;
}
out[i] = roundf(in[i]);
}
break;
}
case TSDB_DATA_TYPE_DOUBLE: {
double *in = (double *)pInputData->pData;
double *out = (double *)pOutputData->pData;
for (int32_t i = 0; i < pOutput->numOfRows; ++i) {
if (colDataIsNull_f(pInputData->nullbitmap, i)) {
colDataAppendNULL(pOutputData, i);
continue;
}
out[i] = round(in[i]);
}
break;
}
default: {
colDataAssign(pOutputData, pInputData, pInput->numOfRows);
}
}
pOutput->numOfRows = pInput->numOfRows;
return TSDB_CODE_SUCCESS;
return doScalarFunction(pInput, inputNum, pOutput, roundf, round);
}
static void tlength(SScalarParam* pOutput, size_t numOfInput, const SScalarParam *pLeft) {

100
source/libs/scalar/src/sclvector.c
View File

@ -25,106 +25,6 @@
#include "tdatablock.h"
#include "ttypes.h"
//GET_TYPED_DATA(v, double, GET_PARAM_TYPE(pRight), (char *)&((right)[i]));
void calc_i32_i32_add(void *left, void *right, int32_t numLeft, int32_t numRight, void *output, int32_t order) {
int32_t *pLeft = (int32_t *)left;
int32_t *pRight = (int32_t *)right;
double * pOutput = (double *)output;
int32_t i = (order == TSDB_ORDER_ASC) ? 0 : TMAX(numLeft, numRight) - 1;
int32_t step = (order == TSDB_ORDER_ASC) ? 1 : -1;
if (numLeft == numRight) {
for (; i >= 0 && i < numRight; i += step, pOutput += 1) {
if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) {
SET_DOUBLE_NULL(pOutput);
continue;
}
*pOutput = (double)pLeft[i] + pRight[i];
}
} else if (numLeft == 1) {
for (; i >= 0 && i < numRight; i += step, pOutput += 1) {
if (isNull((char *)(pLeft), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) {
SET_DOUBLE_NULL(pOutput);
continue;
}
*pOutput = (double)pLeft[0] + pRight[i];
}
} else if (numRight == 1) {
for (; i >= 0 && i < numLeft; i += step, pOutput += 1) {
if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)(pRight), TSDB_DATA_TYPE_INT)) {
SET_DOUBLE_NULL(pOutput);
continue;
}
*pOutput = (double)pLeft[i] + pRight[0];
}
}
}
typedef double (*_getDoubleValue_fn_t)(void *src, int32_t index);
double getVectorDoubleValue_TINYINT(void *src, int32_t index) {
return (double)*((int8_t *)src + index);
}
double getVectorDoubleValue_UTINYINT(void *src, int32_t index) {
return (double)*((uint8_t *)src + index);
}
double getVectorDoubleValue_SMALLINT(void *src, int32_t index) {
return (double)*((int16_t *)src + index);
}
double getVectorDoubleValue_USMALLINT(void *src, int32_t index) {
return (double)*((uint16_t *)src + index);
}
double getVectorDoubleValue_INT(void *src, int32_t index) {
return (double)*((int32_t *)src + index);
}
double getVectorDoubleValue_UINT(void *src, int32_t index) {
return (double)*((uint32_t *)src + index);
}
double getVectorDoubleValue_BIGINT(void *src, int32_t index) {
return (double)*((int64_t *)src + index);
}
double getVectorDoubleValue_UBIGINT(void *src, int32_t index) {
return (double)*((uint64_t *)src + index);
}
double getVectorDoubleValue_FLOAT(void *src, int32_t index) {
return (double)*((float *)src + index);
}
double getVectorDoubleValue_DOUBLE(void *src, int32_t index) {
return (double)*((double *)src + index);
}
_getDoubleValue_fn_t getVectorDoubleValueFn(int32_t srcType) {
_getDoubleValue_fn_t p = NULL;
if(srcType==TSDB_DATA_TYPE_TINYINT) {
p = getVectorDoubleValue_TINYINT;
}else if(srcType==TSDB_DATA_TYPE_UTINYINT) {
p = getVectorDoubleValue_UTINYINT;
}else if(srcType==TSDB_DATA_TYPE_SMALLINT) {
p = getVectorDoubleValue_SMALLINT;
}else if(srcType==TSDB_DATA_TYPE_USMALLINT) {
p = getVectorDoubleValue_USMALLINT;
}else if(srcType==TSDB_DATA_TYPE_INT) {
p = getVectorDoubleValue_INT;
}else if(srcType==TSDB_DATA_TYPE_UINT) {
p = getVectorDoubleValue_UINT;
}else if(srcType==TSDB_DATA_TYPE_BIGINT) {
p = getVectorDoubleValue_BIGINT;
}else if(srcType==TSDB_DATA_TYPE_UBIGINT) {
p = getVectorDoubleValue_UBIGINT;
}else if(srcType==TSDB_DATA_TYPE_FLOAT) {
p = getVectorDoubleValue_FLOAT;
}else if(srcType==TSDB_DATA_TYPE_DOUBLE) {
p = getVectorDoubleValue_DOUBLE;
}else {
assert(0);
}
return p;
}
typedef int64_t (*_getBigintValue_fn_t)(void *src, int32_t index);
int64_t getVectorBigintValue_TINYINT(void *src, int32_t index) {