Skip to content

TDengine
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
TDengine
TDengine
提交 a591f8d0 编写于 作者: D dapan1121
浏览文件
操作

support udf and add case

上级 3e890bac
隐藏空白更改
内联 并排
Showing with 863 addition and 156 deletion
src/client/src/tscSQLParser.c
浏览文件 @ a591f8d0
......@@ -81,7 +81,7 @@ static int32_t parseSelectClause(SSqlCmd* pCmd, int32_t clauseIndex, SArray* pSe
static bool validateIpAddress(const char* ip, size_t size);
static bool hasUnsupportFunctionsForSTableQuery(SSqlCmd* pCmd, SQueryInfo* pQueryInfo);
static bool functionCompatibleCheck(SQueryInfo* pQueryInfo, bool joinQuery, bool twQuery);
static bool functionCompatibleCheck(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, bool joinQuery, bool twQuery);
static int32_t parseGroupbyClause(SQueryInfo* pQueryInfo, SArray* pList, SSqlCmd* pCmd);
......@@ -1740,6 +1740,9 @@ void genUdfList(SArray* pUdfInfo, tSqlExpr *pNode) {
if (pNode->functionId < 0) { // extract all possible user defined function
struct SUdfInfo info = {0};
info.name = strndup(pNode->operand.z, pNode->operand.n);
int32_t functionId = taosArrayGetSize(pUdfInfo) * (-1) - 1;
info.functionId = functionId;
taosArrayPush(pUdfInfo, &info);
}
}
......@@ -1815,6 +1818,8 @@ int32_t parseSelectClause(SSqlCmd* pCmd, int32_t clauseIndex, SArray* pSelectLis
if (pUdfInfo == NULL) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg5);
}
pItem->pNode->functionId = pUdfInfo->functionId;
}
// sql function in selection clause, append sql function info in pSqlCmd structure sequentially
......@@ -1854,7 +1859,7 @@ int32_t parseSelectClause(SSqlCmd* pCmd, int32_t clauseIndex, SArray* pSelectLis
addPrimaryTsColIntoResult(pQueryInfo);
}
if (!functionCompatibleCheck(pQueryInfo, joinQuery, timeWindowQuery)) {
if (!functionCompatibleCheck(pCmd, pQueryInfo, joinQuery, timeWindowQuery)) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg2);
}
......@@ -3131,47 +3136,57 @@ static bool groupbyTagsOrNull(SQueryInfo* pQueryInfo) {
return true;
}
static bool functionCompatibleCheck(SQueryInfo* pQueryInfo, bool joinQuery, bool twQuery) {
static bool functionCompatibleCheck(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, bool joinQuery, bool twQuery) {
int32_t startIdx = 0;
int32_t aggUdf = 0;
int32_t scalarUdf = 0;
int32_t prjNum = 0;
int32_t aggNum = 0;
size_t numOfExpr = tscSqlExprNumOfExprs(pQueryInfo);
assert(numOfExpr > 0);
SSqlExpr* pExpr = tscSqlExprGet(pQueryInfo, startIdx);
// ts function can be simultaneously used with any other functions.
int32_t functionID = pExpr->functionId;
if (functionID == TSDB_FUNC_TS || functionID == TSDB_FUNC_TS_DUMMY) {
startIdx++;
}
int32_t factor = functionCompatList[tscSqlExprGet(pQueryInfo, startIdx)->functionId];
if (tscSqlExprGet(pQueryInfo, 0)->functionId == TSDB_FUNC_LAST_ROW && (joinQuery || twQuery || !groupbyTagsOrNull(pQueryInfo))) {
return false;
}
int32_t factor = INT32_MAX;
// diff function cannot be executed with other function
// arithmetic function can be executed with other arithmetic functions
size_t size = tscSqlExprNumOfExprs(pQueryInfo);
for (int32_t i = startIdx + 1; i < size; ++i) {
for (int32_t i = startIdx; i < size; ++i) {
SSqlExpr* pExpr1 = tscSqlExprGet(pQueryInfo, i);
int16_t functionId = pExpr1->functionId;
if (functionId == TSDB_FUNC_TAGPRJ || functionId == TSDB_FUNC_TAG || functionId == TSDB_FUNC_TS) {
if (functionId < 0) {
SUdfInfo* pUdfInfo = taosArrayGet(pCmd->pUdfInfo, -1 * functionId - 1);
pUdfInfo->funcType == TSDB_UDF_TYPE_AGGREGATE ? ++aggUdf : ++scalarUdf;
continue;
}
if (functionId == TSDB_FUNC_TAGPRJ || functionId == TSDB_FUNC_TAG || functionId == TSDB_FUNC_TS || functionId == TSDB_FUNC_TS_DUMMY) {
++prjNum;
continue;
}
if (functionId == TSDB_FUNC_PRJ) {
++prjNum;
}
if (functionId == TSDB_FUNC_PRJ && (pExpr1->colInfo.colId == PRIMARYKEY_TIMESTAMP_COL_INDEX || TSDB_COL_IS_UD_COL(pExpr1->colInfo.flag))) {
continue;
}
if (functionCompatList[functionId] != factor) {
return false;
if (factor == INT32_MAX) {
factor = functionCompatList[functionId];
} else {
if (factor == -1) { // two functions with the same -1 flag
if (functionCompatList[functionId] != factor) {
return false;
} else {
if (factor == -1) { // two functions with the same -1 flag
return false;
}
}
}
......@@ -3180,6 +3195,18 @@ static bool functionCompatibleCheck(SQueryInfo* pQueryInfo, bool joinQuery, bool
}
}
aggNum = size - prjNum - aggUdf - scalarUdf;
assert(aggNum >= 0);
if (aggUdf > 0 && (prjNum > 0 || aggNum > 0 || scalarUdf > 0)) {
return false;
}
if (scalarUdf > 0 && aggNum > 0) {
return false;
}
return true;
}
......@@ -5276,7 +5303,7 @@ int32_t validateSqlFunctionInStreamSql(SSqlCmd* pCmd, SQueryInfo* pQueryInfo) {
int32_t validateFunctionsInIntervalOrGroupbyQuery(SSqlCmd* pCmd, SQueryInfo* pQueryInfo) {
bool isProjectionFunction = false;
const char* msg1 = "column projection is not compatible with interval";
const char* msg1 = "column projection or function not compatible with interval";
// multi-output set/ todo refactor
size_t size = taosArrayGetSize(pQueryInfo->exprList);
......@@ -5284,6 +5311,16 @@ int32_t validateFunctionsInIntervalOrGroupbyQuery(SSqlCmd* pCmd, SQueryInfo* pQu
for (int32_t k = 0; k < size; ++k) {
SSqlExpr* pExpr = tscSqlExprGet(pQueryInfo, k);
if (pExpr->functionId < 0) {
SUdfInfo* pUdfInfo = taosArrayGet(pCmd->pUdfInfo, -1 * pExpr->functionId - 1);
if (pUdfInfo->funcType == TSDB_UDF_TYPE_SCALAR) {
isProjectionFunction = true;
break;
} else {
continue;
}
}
// projection query on primary timestamp, the selectivity function needs to be present.
if (pExpr->functionId == TSDB_FUNC_PRJ && pExpr->colInfo.colId == PRIMARYKEY_TIMESTAMP_COL_INDEX) {
bool hasSelectivity = false;
......@@ -5303,6 +5340,7 @@ int32_t validateFunctionsInIntervalOrGroupbyQuery(SSqlCmd* pCmd, SQueryInfo* pQu
if ((pExpr->functionId == TSDB_FUNC_PRJ && pExpr->numOfParams == 0) || pExpr->functionId == TSDB_FUNC_DIFF ||
pExpr->functionId == TSDB_FUNC_ARITHM) {
isProjectionFunction = true;
break;
}
}
......@@ -5910,6 +5948,15 @@ static int32_t checkUpdateTagPrjFunctions(SQueryInfo* pQueryInfo, SSqlCmd* pCmd)
continue;
}
if (functionId < 0) {
SUdfInfo* pUdfInfo = taosArrayGet(pCmd->pUdfInfo, -1 * functionId - 1);
if (pUdfInfo->funcType == TSDB_UDF_TYPE_AGGREGATE) {
++numOfAggregation;
}
continue;
}
if ((aAggs[functionId].status & TSDB_FUNCSTATE_SELECTIVITY) != 0) {
numOfSelectivity++;
} else {
......@@ -6129,6 +6176,10 @@ int32_t doFunctionsCompatibleCheck(SSqlCmd* pCmd, SQueryInfo* pQueryInfo) {
}
}
if (functId < 0) {
continue;
}
if (IS_MULTIOUTPUT(aAggs[functId].status) && functId != TSDB_FUNC_TOP && functId != TSDB_FUNC_BOTTOM &&
functId != TSDB_FUNC_TAGPRJ && functId != TSDB_FUNC_PRJ) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg1);
......@@ -6896,10 +6947,6 @@ int32_t doValidateSqlNode(SSqlObj* pSql, SQuerySqlNode* pQuerySqlNode, int32_t i
}
pQuerySqlNode->pWhere = NULL;
if (tinfo.precision == TSDB_TIME_PRECISION_MILLI) {
pQueryInfo->window.skey = pQueryInfo->window.skey / 1000;
pQueryInfo->window.ekey = pQueryInfo->window.ekey / 1000;
}
} else { // set the time rang
if (taosArrayGetSize(pQuerySqlNode->from->tableList) > 1) { // it is a join query, no where clause is not allowed.
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), "condition missing for join query ");
......@@ -6916,6 +6963,11 @@ int32_t doValidateSqlNode(SSqlObj* pSql, SQuerySqlNode* pQuerySqlNode, int32_t i
return code;
}
if (tinfo.precision == TSDB_TIME_PRECISION_MILLI) {
pQueryInfo->window.skey = pQueryInfo->window.skey / 1000;
pQueryInfo->window.ekey = pQueryInfo->window.ekey / 1000;
}
if (parseSelectClause(pCmd, index, pQuerySqlNode->pSelectList, isSTable, joinQuery, timeWindowQuery) != TSDB_CODE_SUCCESS) {
return TSDB_CODE_TSC_INVALID_SQL;
}
......
src/common/inc/tdataformat.h
浏览文件 @ a591f8d0
......@@ -34,6 +34,14 @@ extern "C" {
memcpy(varDataVal(x), (str), __len); \
} while (0);
#define STR_TO_NET_VARSTR(x, str) \
do { \
VarDataLenT __len = (VarDataLenT)strlen(str); \
*(VarDataLenT *)(x) = htons(__len); \
memcpy(varDataVal(x), (str), __len); \
} while (0);
#define STR_WITH_MAXSIZE_TO_VARSTR(x, str, _maxs) \
do { \
char *_e = stpncpy(varDataVal(x), (str), (_maxs)-VARSTR_HEADER_SIZE); \
......
src/inc/ttype.h
浏览文件 @ a591f8d0
......@@ -26,6 +26,10 @@ typedef struct tstr {
#define varDataSetLen(v, _len) (((VarDataLenT *)(v))[0] = (VarDataLenT) (_len))
#define IS_VAR_DATA_TYPE(t) (((t) == TSDB_DATA_TYPE_BINARY) || ((t) == TSDB_DATA_TYPE_NCHAR))
#define varDataNetLen(v) (htons(((VarDataLenT *)(v))[0]))
#define varDataNetTLen(v) (sizeof(VarDataLenT) + varDataNetLen(v))
// this data type is internally used only in 'in' query to hold the values
#define TSDB_DATA_TYPE_ARRAY (1000)
......
src/mnode/src/mnodeFunc.c
浏览文件 @ a591f8d0
......@@ -326,12 +326,6 @@ static int32_t mnodeGetFuncMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pCo
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pShow->bytes[cols] = TSDB_FUNC_CODE_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "code");
pSchema[cols].bytes = htons(pShow->bytes[cols]);
cols++;
pMeta->numOfColumns = htons(cols);
strcpy(pMeta->tableFname, "show funcs");
pShow->numOfColumns = cols;
......@@ -404,10 +398,6 @@ static int32_t mnodeRetrieveFuncs(SShowObj *pShow, char *data, int32_t rows, voi
*(int32_t *)pWrite = pFunc->contLen;
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
STR_WITH_MAXSIZE_TO_VARSTR(pWrite, pFunc->cont, pShow->bytes[cols]);
cols++;
numOfRows++;
mnodeDecFuncRef(pFunc);
}
......@@ -446,11 +436,11 @@ static int32_t mnodeProcessRetrieveFuncImplMsg(SMnodeMsg *pMsg) {
SUdfFuncMsg *pFuncMsg = rpcMallocCont(t);
pFuncMsg->num = htonl(pInfo->num);
char* pOutput = pFuncMsg->content;
for(int32_t i = 0; i < pInfo->num; ++i) {
tstr* name = (tstr*) pInfo->name;
tstr* name = (tstr*) pInfo->name;
for(int32_t i = 0; i < pInfo->num; ++i) {
char buf[TSDB_FUNC_NAME_LEN] = {0};
tstrncpy(buf, name->data, TSDB_FUNC_NAME_LEN);
tstrncpy(buf, name->data, htons(name->len) + 1);
SFuncObj* pFuncObj = mnodeGetFunc(buf);
if (pFuncObj == NULL) {
......@@ -467,7 +457,9 @@ static int32_t mnodeProcessRetrieveFuncImplMsg(SMnodeMsg *pMsg) {
pFuncInfo->funcType = htonl(pFuncObj->funcType);
pFuncInfo->resType = pFuncObj->resType;
pFuncInfo->resBytes = htons(pFuncObj->resBytes);
pOutput += sizeof(SFunctionInfoMsg) + pFuncObj->contLen;
name =(void *)name + sizeof(*name) + htons(name->len);
}
pMsg->rpcRsp.rsp = pFuncMsg;
......
src/query/inc/qUdf.h
浏览文件 @ a591f8d0
......@@ -16,7 +16,7 @@
#ifndef TDENGINE_QUDF_H
#define TDENGINE_QUDF_H
enum { TSDB_UDF_FUNC_NORMAL = 0, TSDB_UDF_FUNC_INIT, TSDB_UDF_FUNC_FINALIZE, TSDB_UDF_FUNC_MAX_NUM };
enum { TSDB_UDF_FUNC_NORMAL = 0, TSDB_UDF_FUNC_INIT, TSDB_UDF_FUNC_FINALIZE, TSDB_UDF_FUNC_DESTROY, TSDB_UDF_FUNC_MAX_NUM };
......@@ -46,10 +46,10 @@ typedef struct SUdfInfo {
} SUdfInfo;
typedef void (*udfNormalFunc)(char* data, int8_t type, int32_t numOfRows, int64_t* ts, char* dataOutput, char* tsOutput,
int32_t* numOfOutput, char* buf);
typedef void (*udfInitFunc)(SUdfInit* data);
typedef void (*udfFinalizeFunc)(char* data, int8_t type, int32_t numOfRows, int64_t* ts, char* dataOutput, char* tsOutput,
int32_t* numOfOutput, char* buf);
int32_t* numOfOutput, SUdfInit* buf);
typedef int32_t (*udfInitFunc)(SUdfInit* data);
typedef void (*udfFinalizeFunc)(char* dataOutput, int32_t* numOfOutput, SUdfInit* buf);
typedef void (*udfDestroyFunc)(SUdfInit* buf);
#endif // TDENGINE_QUDF_H
src/query/src/qExecutor.c
浏览文件 @ a591f8d0
......@@ -772,12 +772,21 @@ static int32_t getNumOfRowsInTimeWindow(SQuery *pQuery, SDataBlockInfo *pDataBlo
return num;
}
static void doInvokeUdf(SUdfInfo* pUdfInfo, char* data, int8_t type, int32_t numOfRows, int64_t* ts, char* dataOutput, char* tsOutput,
int32_t* numOfOutput, char* buf) {
static void doInvokeUdf(SQueryRuntimeEnv *pRuntimeEnv, SQLFunctionCtx *pCtx, int32_t idx) {
int32_t output = 0;
SUdfInfo* pUdfInfo = pRuntimeEnv->pUdfInfo;
if (pUdfInfo && pUdfInfo->funcs[TSDB_UDF_FUNC_NORMAL]) {
qDebug("invoke udf function:%s,%p", pUdfInfo->name, pUdfInfo->funcs[TSDB_UDF_FUNC_NORMAL]);
(*(udfNormalFunc)pUdfInfo->funcs[TSDB_UDF_FUNC_NORMAL])(data, type, numOfRows, ts, dataOutput, tsOutput, numOfOutput, buf);
(*(udfNormalFunc)pUdfInfo->funcs[TSDB_UDF_FUNC_NORMAL])(pCtx->pInput + idx * pCtx->inputType, pCtx->inputType, pCtx->size, pCtx->ptsList, pCtx->pOutput,
pCtx->ptsOutputBuf, &output, &pUdfInfo->init);
// set the output value exist
pCtx->resultInfo->numOfRes += output;
if (output > 0) {
pCtx->resultInfo->hasResult = DATA_SET_FLAG;
}
return;
}
......@@ -817,17 +826,7 @@ static void doApplyFunctions(SQueryRuntimeEnv* pRuntimeEnv, SQLFunctionCtx* pCtx
if (functionNeedToExecute(pRuntimeEnv, &pCtx[k], functionId)) {
// aAggs[functionId].xFunction(&pCtx[k]);
if (functionId < 0) { // load the script and exec, pRuntimeEnv->pUdfInfo
int32_t output = 0;
char* buf = GET_ROWCELL_INTERBUF(pCtx[k].resultInfo);
doInvokeUdf(pRuntimeEnv->pUdfInfo, pCtx[k].pInput, pCtx[k].inputType, pCtx[k].size, pCtx[k].ptsList, pCtx[k].pOutput,
pCtx[k].ptsOutputBuf, &output, buf);
// set the output value exist
pCtx[k].resultInfo->numOfRes = output;
if (output > 0) {
pCtx[k].resultInfo->hasResult = DATA_SET_FLAG;
}
doInvokeUdf(pRuntimeEnv, &pCtx[k], 0);
} else {
aAggs[functionId].xFunction(&pCtx[k]);
}
......@@ -1059,18 +1058,7 @@ static void doAggregateImpl(SOperatorInfo* pOperator, TSKEY startTs, SQLFunction
pCtx[k].startTs = startTs;// this can be set during create the struct
// aAggs[functionId].xFunction(&pCtx[k]);
if (functionId < 0) {
int32_t output = 0;
char* buf = GET_ROWCELL_INTERBUF(pCtx[k].resultInfo);
doInvokeUdf(pRuntimeEnv->pUdfInfo, pCtx[k].pInput, pCtx[k].inputType, pCtx[k].size, pCtx[k].ptsList, pCtx[k].pOutput,
pCtx[k].ptsOutputBuf, &output, buf);
// set the output value exist
pCtx[k].resultInfo->numOfRes = output;
if (output > 0) {
pCtx[k].resultInfo->hasResult = DATA_SET_FLAG;
}
doInvokeUdf(pRuntimeEnv, &pCtx[k], 0);
} else {
aAggs[functionId].xFunction(&pCtx[k]);
}
......@@ -1085,18 +1073,8 @@ static void arithmeticApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SQLFunctionC
pCtx[k].startTs = pQuery->window.skey;
if (pCtx[k].functionId < 0) {
// load the script and exec, pRuntimeEnv->pUdfInfo
int32_t output = 0;
char* buf = GET_ROWCELL_INTERBUF(pCtx[k].resultInfo);
doInvokeUdf(pRuntimeEnv->pUdfInfo, pCtx[k].pInput, pCtx[k].inputType, pCtx[k].size, pCtx[k].ptsList, pCtx[k].pOutput,
pCtx[k].ptsOutputBuf, &output, buf);
// set the output value exist
pCtx[k].resultInfo->numOfRes = output;
if (output > 0) {
pCtx[k].resultInfo->hasResult = DATA_SET_FLAG;
}
// load the script and exec
doInvokeUdf(pRuntimeEnv, &pCtx[k], 0);
} else {
aAggs[pCtx[k].functionId].xFunction(&pCtx[k]);
}
......@@ -1414,7 +1392,9 @@ static void doHashGroupbyAgg(SOperatorInfo* pOperator, SGroupbyOperatorInfo *pIn
for (int32_t k = 0; k < pOperator->numOfOutput; ++k) {
pInfo->binfo.pCtx[k].size = 1;
int32_t functionId = pInfo->binfo.pCtx[k].functionId;
if (functionNeedToExecute(pRuntimeEnv, &pInfo->binfo.pCtx[k], functionId)) {
if (functionId < 0) {
doInvokeUdf(pRuntimeEnv, &pInfo->binfo.pCtx[k], j);
} else if (functionNeedToExecute(pRuntimeEnv, &pInfo->binfo.pCtx[k], functionId)) {
aAggs[functionId].xFunctionF(&pInfo->binfo.pCtx[k], j);
}
}
......@@ -3164,9 +3144,7 @@ void initCtxOutputBuffer(SQueryRuntimeEnv *pRuntimeEnv, SQLFunctionCtx* pCtx, in
}
if (pCtx[j].functionId < 0) { // todo udf initialization
if (pRuntimeEnv->pUdfInfo && pRuntimeEnv->pUdfInfo->funcs[TSDB_UDF_FUNC_INIT]) {
(*(udfInitFunc)pRuntimeEnv->pUdfInfo->funcs[TSDB_UDF_FUNC_INIT])(&pRuntimeEnv->pUdfInfo->init);
}
continue;
} else {
aAggs[pCtx[j].functionId].init(&pCtx[j], pCtx[j].resultInfo);
}
......@@ -3226,11 +3204,9 @@ void finalizeQueryResult(SOperatorInfo* pOperator, SQLFunctionCtx* pCtx, SResult
for (int32_t j = 0; j < numOfOutput; ++j) {
if (pCtx[j].functionId < 0) {
int32_t output = 0;
char* buf = GET_ROWCELL_INTERBUF(pCtx[j].resultInfo);
if (pRuntimeEnv->pUdfInfo && pRuntimeEnv->pUdfInfo->funcs[TSDB_UDF_FUNC_FINALIZE]) {
(*(udfFinalizeFunc)pRuntimeEnv->pUdfInfo->funcs[TSDB_UDF_FUNC_FINALIZE])(pCtx[j].pInput, pCtx[j].inputType, pCtx[j].size, pCtx[j].ptsList, pCtx[j].pOutput,
pCtx[j].ptsOutputBuf, &output, buf);
(*(udfFinalizeFunc)pRuntimeEnv->pUdfInfo->funcs[TSDB_UDF_FUNC_FINALIZE])(pCtx[j].pOutput, &output, &pRuntimeEnv->pUdfInfo->init);
}
// set the output value exist
......@@ -3255,11 +3231,9 @@ void finalizeQueryResult(SOperatorInfo* pOperator, SQLFunctionCtx* pCtx, SResult
for (int32_t j = 0; j < numOfOutput; ++j) {
if (pCtx[j].functionId < 0) {
int32_t output = 0;
char* buf = GET_ROWCELL_INTERBUF(pCtx[j].resultInfo);
if (pRuntimeEnv->pUdfInfo && pRuntimeEnv->pUdfInfo->funcs[TSDB_UDF_FUNC_FINALIZE]) {
(*(udfFinalizeFunc)pRuntimeEnv->pUdfInfo->funcs[TSDB_UDF_FUNC_FINALIZE])(pCtx[j].pInput, pCtx[j].inputType, pCtx[j].size, pCtx[j].ptsList, pCtx[j].pOutput,
pCtx[j].ptsOutputBuf, &output, buf);
(*(udfFinalizeFunc)pRuntimeEnv->pUdfInfo->funcs[TSDB_UDF_FUNC_FINALIZE])(pCtx[j].pOutput, &output, &pRuntimeEnv->pUdfInfo->init);
}
// set the output value exist
......@@ -3336,6 +3310,10 @@ void setResultRowOutputBufInitCtx(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *pRe
offset += pCtx[i].outputBytes;
int32_t functionId = pCtx[i].functionId;
if (functionId < 0) {
continue;
}
if (functionId == TSDB_FUNC_TOP || functionId == TSDB_FUNC_BOTTOM || functionId == TSDB_FUNC_DIFF) {
pCtx[i].ptsOutputBuf = pCtx[0].pOutput;
}
......@@ -5959,6 +5937,10 @@ void destroyUdfInfo(SUdfInfo* pUdfInfo) {
if (pUdfInfo == NULL) {
return;
}
if (pUdfInfo->funcs[TSDB_UDF_FUNC_DESTROY]) {
(*(udfDestroyFunc)pUdfInfo->funcs[TSDB_UDF_FUNC_DESTROY])(&pUdfInfo->init);
}
tfree(pUdfInfo->name);
......@@ -5986,6 +5968,9 @@ static char* getUdfFuncName(char* name, int type) {
case TSDB_UDF_FUNC_FINALIZE:
sprintf(funcname, "%s_finalize", name);
break;
case TSDB_UDF_FUNC_DESTROY:
sprintf(funcname, "%s_destroy", name);
break;
default:
assert(0);
break;
......@@ -5994,7 +5979,7 @@ static char* getUdfFuncName(char* name, int type) {
return funcname;
}
static int32_t flushUdfContentToDisk(SUdfInfo* pUdfInfo) {
static int32_t initUdfInfo(SUdfInfo* pUdfInfo) {
if (pUdfInfo == NULL) {
return TSDB_CODE_SUCCESS;
}
......@@ -6023,10 +6008,16 @@ static int32_t flushUdfContentToDisk(SUdfInfo* pUdfInfo) {
}
pUdfInfo->funcs[TSDB_UDF_FUNC_INIT] = taosLoadSym(pUdfInfo->handle, getUdfFuncName(pUdfInfo->name, TSDB_UDF_FUNC_INIT));
if (pUdfInfo->funcType == TSDB_UDF_TYPE_AGGREGATE) {
pUdfInfo->funcs[TSDB_UDF_FUNC_FINALIZE] = taosLoadSym(pUdfInfo->handle, getUdfFuncName(pUdfInfo->name, TSDB_UDF_FUNC_FINALIZE));
}
pUdfInfo->funcs[TSDB_UDF_FUNC_DESTROY] = taosLoadSym(pUdfInfo->handle, getUdfFuncName(pUdfInfo->name, TSDB_UDF_FUNC_DESTROY));
if (pUdfInfo->funcs[TSDB_UDF_FUNC_INIT]) {
return (*(udfInitFunc)pUdfInfo->funcs[TSDB_UDF_FUNC_INIT])(&pUdfInfo->init);
}
return TSDB_CODE_SUCCESS;
}
......@@ -6037,8 +6028,7 @@ int32_t createQueryFuncExprFromMsg(SQueryTableMsg* pQueryMsg, int32_t numOfOutpu
*pExprInfo = NULL;
int32_t code = TSDB_CODE_SUCCESS;
// save the udf script or so file
code = flushUdfContentToDisk(pUdfInfo);
code = initUdfInfo(pUdfInfo);
if (code) {
return code;
}
......@@ -6396,14 +6386,27 @@ SQInfo* createQInfoImpl(SQueryTableMsg* pQueryMsg, SSqlGroupbyExpr* pGroupbyExpr
}
}
// calculate the result row size
for (int16_t col = 0; col < numOfOutput; ++col) {
assert(pExprs[col].bytes > 0);
pQuery->resultRowSize += pExprs[col].bytes;
if (pSecExprs != NULL) {
// calculate the result row size
for (int16_t col = 0; col < pQuery->numOfExpr2; ++col) {
assert(pSecExprs[col].bytes > 0);
pQuery->resultRowSize += pSecExprs[col].bytes;
// keep the tag length
if (TSDB_COL_IS_TAG(pExprs[col].base.colInfo.flag)) {
pQuery->tagLen += pExprs[col].bytes;
// keep the tag length
if (TSDB_COL_IS_TAG(pSecExprs[col].base.colInfo.flag)) {
pQuery->tagLen += pSecExprs[col].bytes;
}
}
} else {
// calculate the result row size
for (int16_t col = 0; col < numOfOutput; ++col) {
assert(pExprs[col].bytes > 0);
pQuery->resultRowSize += pExprs[col].bytes;
// keep the tag length
if (TSDB_COL_IS_TAG(pExprs[col].base.colInfo.flag)) {
pQuery->tagLen += pExprs[col].bytes;
}
}
}
......
tests/script/general/parser/udf.sim
浏览文件 @ a591f8d0
......@@ -22,13 +22,13 @@ endi
if $data01 != /tmp/normal then
return -1
endi
if $data02 != INT then
if $data02 != 0 then
return -1
endi
if $data04 != 5 then
if $data03 != INT then
return -1
endi
if $data05 != hello then
if $data05 != 5 then
return -1
endi
......@@ -54,13 +54,13 @@ endi
if $data01 != /tmp/normal then
return -1
endi
if $data02 != BOOL then
if $data02 != 0 then
return -1
endi
if $data04 != 5 then
if $data03 != BOOL then
return -1
endi
if $data05 != hello then
if $data05 != 5 then
return -1
endi
......@@ -88,13 +88,13 @@ endi
if $data01 != /tmp/normal then
return -1
endi
if $data02 != TINYINT then
if $data02 != 0 then
return -1
endi
if $data04 != 5 then
if $data03 != TINYINT then
return -1
endi
if $data05 != hello then
if $data05 != 5 then
return -1
endi
......@@ -123,13 +123,13 @@ endi
if $data01 != /tmp/normal then
return -1
endi
if $data02 != SMALLINT then
if $data02 != 0 then
return -1
endi
if $data04 != 5 then
if $data03 != SMALLINT then
return -1
endi
if $data05 != hello then
if $data05 != 5 then
return -1
endi
......@@ -159,13 +159,13 @@ endi
if $data01 != /tmp/normal then
return -1
endi
if $data02 != INT then
if $data02 != 0 then
return -1
endi
if $data04 != 5 then
if $data03 != INT then
return -1
endi
if $data05 != hello then
if $data05 != 5 then
return -1
endi
......@@ -199,13 +199,13 @@ endi
if $data01 != /tmp/normal then
return -1
endi
if $data02 != BIGINT then
if $data02 != 0 then
return -1
endi
if $data04 != 5 then
if $data03 != BIGINT then
return -1
endi
if $data05 != hello then
if $data05 != 5 then
return -1
endi
......@@ -237,13 +237,13 @@ endi
if $data01 != /tmp/normal then
return -1
endi
if $data02 != FLOAT then
if $data02 != 0 then
return -1
endi
if $data04 != 5 then
if $data03 != FLOAT then
return -1
endi
if $data05 != hello then
if $data05 != 5 then
return -1
endi
......@@ -274,13 +274,13 @@ endi
if $data01 != /tmp/normal then
return -1
endi
if $data02 != DOUBLE then
if $data02 != 0 then
return -1
endi
if $data04 != 5 then
if $data03 != DOUBLE then
return -1
endi
if $data05 != hello then
if $data05 != 5 then
return -1
endi
......@@ -311,13 +311,13 @@ endi
if $data01 != /tmp/normal then
return -1
endi
if $data02 != BINARY(0) then
if $data02 != 0 then
return -1
endi
if $data04 != 5 then
if $data03 != BINARY(0) then
return -1
endi
if $data05 != hello then
if $data05 != 5 then
return -1
endi
......@@ -343,13 +343,13 @@ endi
if $data01 != /tmp/normal then
return -1
endi
if $data02 != BINARY(10) then
if $data02 != 0 then
return -1
endi
if $data04 != 5 then
if $data03 != BINARY(10) then
return -1
endi
if $data05 != hello then
if $data05 != 5 then
return -1
endi
......@@ -375,13 +375,13 @@ endi
if $data01 != /tmp/normal then
return -1
endi
if $data02 != TIMESTAMP then
if $data02 != 0 then
return -1
endi
if $data04 != 5 then
if $data03 != TIMESTAMP then
return -1
endi
if $data05 != hello then
if $data05 != 5 then
return -1
endi
......@@ -407,13 +407,13 @@ endi
if $data01 != /tmp/normal then
return -1
endi
if $data02 != NCHAR(0) then
if $data02 != 0 then
return -1
endi
if $data04 != 5 then
if $data03 != NCHAR(0) then
return -1
endi
if $data05 != hello then
if $data05 != 5 then
return -1
endi
......@@ -439,13 +439,13 @@ endi
if $data01 != /tmp/normal then
return -1
endi
if $data02 != NCHAR(10) then
if $data02 != 0 then
return -1
endi
if $data04 != 5 then
if $data03 != NCHAR(10) then
return -1
endi
if $data05 != hello then
if $data05 != 5 then
return -1
endi
......@@ -473,10 +473,10 @@ endi
if $data01 != /tmp/normal then
return -1
endi
if $data04 != 5 then
if $data02 != 0 then
return -1
endi
if $data05 != hello then
if $data05 != 5 then
return -1
endi
......@@ -503,10 +503,10 @@ endi
if $data01 != /tmp/normal then
return -1
endi
if $data04 != 5 then
if $data02 != 0 then
return -1
endi
if $data05 != hello then
if $data05 != 5 then
return -1
endi
......@@ -532,10 +532,10 @@ endi
if $data01 != /tmp/normal then
return -1
endi
if $data04 != 5 then
if $data02 != 0 then
return -1
endi
if $data05 != hello then
if $data05 != 5 then
return -1
endi
......@@ -562,10 +562,10 @@ endi
if $data01 != /tmp/normal then
return -1
endi
if $data04 != 5 then
if $data02 != 0 then
return -1
endi
if $data05 != hello then
if $data05 != 5 then
return -1
endi
......@@ -577,6 +577,34 @@ if $rows != 0 then
endi
sql create aggregate function n18 as '/tmp/normal' outputtype BIGINT UNSIGNED;
sql show functions;
if $rows != 1 then
return -1
endi
if $data00 != n18 then
return -1
endi
if $data01 != /tmp/normal then
return -1
endi
if $data02 != 1 then
return -1
endi
if $data05 != 5 then
return -1
endi
sql drop function n18;
sql show functions;
if $rows != 0 then
return -1
endi
sql create function t01 as '/tmp/normal' outputtype INT;
sql_error create function t01 as '/tmp/normal' outputtype SMALLINT;
......
tests/script/general/parser/udf_dll.sim 0 → 100644
浏览文件 @ a591f8d0
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/cfg.sh -n dnode1 -c walLevel -v 1
system sh/cfg.sh -n dnode1 -c maxtablesPerVnode -v 2
system sh/exec.sh -n dnode1 -s start
system sh/prepare_udf.sh
sleep 100
sql connect
print ======================== dnode1 start
sql create function add_one as '/tmp/add_one.so' outputtype int;
sql create aggregate function sum_double as '/tmp/sum_double.so' outputtype int;
sql show functions;
if $rows != 2 then
return -1
endi
sql create database db;
sql use db;
sql create table tb1 (ts timestamp, f1 int, f2 bool, f3 binary(10));
sql insert into tb1 values ('2021-03-23 17:17:19.660', 1, true, 'tb1-1');
sql insert into tb1 values ('2021-03-23 19:23:28.595', 2, false, 'tb1-2');
sql insert into tb1 values ('2021-03-23 19:33:39.070', 3, true, 'tb1-3');
sql insert into tb1 values ('2021-03-23 19:34:37.670', 4, false, 'tb1-4');
sql insert into tb1 values ('2021-03-24 19:08:06.609', 5, true, 'tb1-5');
sql insert into tb1 values ('2021-03-24 19:26:38.231', 6, false, 'tb1-6');
sql insert into tb1 values ('2021-03-25 10:03:17.688', 7, true, 'tb1-7');
sql select add_one(f1) from tb1;
if $rows != 7 then
return -1
endi
if $data00 != 2 then
return -1
endi
if $data10 != 3 then
return -1
endi
if $data20 != 4 then
return -1
endi
if $data30 != 5 then
return -1
endi
if $data40 != 6 then
return -1
endi
if $data50 != 7 then
return -1
endi
if $data60 != 8 then
return -1
endi
sql select sum_double(f1) from tb1;
if $rows != 1 then
return -1
endi
if $data00 != 56 then
return -1
endi
sql select ts,add_one(f1),f1 from tb1;
if $rows != 7 then
return -1
endi
if $data00 != @21-03-23 17:17:19.660@ then
return -1
endi
if $data01 != 2 then
return -1
endi
if $data02 != 1 then
return -1
endi
if $data10 != @21-03-23 19:23:28.595@ then
return -1
endi
if $data11 != 3 then
return -1
endi
if $data12 != 2 then
return -1
endi
if $data20 != @21-03-23 19:33:39.070@ then
return -1
endi
if $data21 != 4 then
return -1
endi
if $data22 != 3 then
return -1
endi
if $data30 != @21-03-23 19:34:37.670@ then
return -1
endi
if $data31 != 5 then
return -1
endi
if $data32 != 4 then
return -1
endi
if $data40 != @21-03-24 19:08:06.609@ then
return -1
endi
if $data41 != 6 then
return -1
endi
if $data42 != 5 then
return -1
endi
if $data50 != @21-03-24 19:26:38.231@ then
return -1
endi
if $data51 != 7 then
return -1
endi
if $data52 != 6 then
return -1
endi
if $data60 != @21-03-25 10:03:17.688@ then
return -1
endi
if $data61 != 8 then
return -1
endi
if $data62 != 7 then
return -1
endi
sql select add_one(f1),add_one(f1) from tb1;
if $rows != 7 then
return -1
endi
if $data00 != 2 then
return -1
endi
if $data01 != 2 then
return -1
endi
if $data10 != 3 then
return -1
endi
if $data11 != 3 then
return -1
endi
if $data20 != 4 then
return -1
endi
if $data21 != 4 then
return -1
endi
if $data30 != 5 then
return -1
endi
if $data31 != 5 then
return -1
endi
if $data40 != 6 then
return -1
endi
if $data41 != 6 then
return -1
endi
if $data50 != 7 then
return -1
endi
if $data51 != 7 then
return -1
endi
if $data60 != 8 then
return -1
endi
if $data61 != 8 then
return -1
endi
sql select add_one(f1)+1 from tb1;
if $rows != 7 then
return -1
endi
if $data00 != 3.000000000 then
return -1
endi
if $data10 != 4.000000000 then
return -1
endi
if $data20 != 5.000000000 then
return -1
endi
if $data30 != 6.000000000 then
return -1
endi
if $data40 != 7.000000000 then
return -1
endi
if $data50 != 8.000000000 then
return -1
endi
if $data60 != 9.000000000 then
return -1
endi
sql select sum_double(f1)+1 from tb1;
if $rows != 1 then
return -1
endi
if $data00 != 57.000000000 then
return -1
endi
sql select add_one(f1)+1,f1 from tb1;
if $rows != 7 then
return -1
endi
if $data00 != 3.000000000 then
return -1
endi
if $data01 != 1 then
return -1
endi
if $data10 != 4.000000000 then
return -1
endi
if $data11 != 2 then
return -1
endi
if $data20 != 5.000000000 then
return -1
endi
if $data21 != 3 then
return -1
endi
if $data30 != 6.000000000 then
return -1
endi
if $data31 != 4 then
return -1
endi
if $data40 != 7.000000000 then
return -1
endi
if $data41 != 5 then
return -1
endi
if $data50 != 8.000000000 then
return -1
endi
if $data51 != 6 then
return -1
endi
if $data60 != 9.000000000 then
return -1
endi
if $data61 != 7 then
return -1
endi
sql select sum_double(f1) from tb1 interval (10a);
if $rows != 7 then
return -1
endi
if $data00 != @21-03-23 17:17:19.660@ then
return -1
endi
if $data01 != 2 then
return -1
endi
if $data10 != @21-03-23 19:23:28.590@ then
return -1
endi
if $data11 != 4 then
return -1
endi
if $data20 != @21-03-23 19:33:39.070@ then
return -1
endi
if $data21 != 6 then
return -1
endi
if $data30 != @21-03-23 19:34:37.670@ then
return -1
endi
if $data31 != 8 then
return -1
endi
if $data40 != @21-03-24 19:08:06.600@ then
return -1
endi
if $data41 != 10 then
return -1
endi
if $data50 != @21-03-24 19:26:38.230@ then
return -1
endi
if $data51 != 12 then
return -1
endi
if $data60 != @21-03-25 10:03:17.680@ then
return -1
endi
if $data61 != 14 then
return -1
endi
sql select ts,add_one(f1) from tb1 where ts>="2021-03-23 17:00:00.000" and ts<="2021-03-24 20:00:00.000";
if $rows != 6 then
return -1
endi
if $data00 != @21-03-23 17:17:19.660@ then
return -1
endi
if $data01 != 2 then
return -1
endi
if $data10 != @21-03-23 19:23:28.595@ then
return -1
endi
if $data11 != 3 then
return -1
endi
if $data20 != @21-03-23 19:33:39.070@ then
return -1
endi
if $data21 != 4 then
return -1
endi
if $data30 != @21-03-23 19:34:37.670@ then
return -1
endi
if $data31 != 5 then
return -1
endi
if $data40 != @21-03-24 19:08:06.609@ then
return -1
endi
if $data41 != 6 then
return -1
endi
if $data50 != @21-03-24 19:26:38.231@ then
return -1
endi
if $data51 != 7 then
return -1
endi
sql select sum_double(f1) from tb1 where ts>="2021-03-23 17:00:00.000" and ts<="2021-03-24 20:00:00.000" interval (1h);
if $rows != 3 then
return -1
endi
if $data00 != @21-03-23 17:00:00.000@ then
return -1
endi
if $data01 != 2 then
return -1
endi
if $data10 != @21-03-23 19:00:00.000@ then
return -1
endi
if $data11 != 18 then
return -1
endi
if $data20 != @21-03-24 19:00:00.000@ then
return -1
endi
if $data21 != 22 then
return -1
endi
sql select sum_double(f1) from tb1 where ts>="2021-03-23 17:00:00.000" and ts<="2021-03-24 20:00:00.000" interval (1h) fill(value,999);
if $rows != 28 then
return -1
endi
sql select add_one(f1) from tb1 group by f1;
if $rows != 7 then
return -1
endi
if $data00 != 2 then
return -1
endi
if $data10 != 3 then
return -1
endi
if $data20 != 4 then
return -1
endi
if $data30 != 5 then
return -1
endi
if $data40 != 6 then
return -1
endi
if $data50 != 7 then
return -1
endi
if $data60 != 8 then
return -1
endi
sql select sum_double(f1) from tb1 group by f1;
if $rows != 7 then
return -1
endi
if $data00 != 2 then
return -1
endi
if $data10 != 4 then
return -1
endi
if $data20 != 6 then
return -1
endi
if $data30 != 8 then
return -1
endi
if $data40 != 10 then
return -1
endi
if $data50 != 12 then
return -1
endi
if $data60 != 14 then
return -1
endi
sql select sum_double(f1) from tb1 interval (1h) order by ts desc;
if $rows != 4 then
return -1
endi
if $data00 != @21-03-25 10:00:00.000@ then
return -1
endi
if $data01 != 14 then
return -1
endi
if $data10 != @21-03-24 19:00:00.000@ then
return -1
endi
if $data11 != 22 then
return -1
endi
if $data20 != @21-03-23 19:00:00.000@ then
return -1
endi
if $data21 != 18 then
return -1
endi
if $data30 != @21-03-23 17:00:00.000@ then
return -1
endi
if $data31 != 2 then
return -1
endi
sql select add_one(f1) from tb1 limit 2;
if $rows != 2 then
return -1
endi
if $data00 != 2 then
return -1
endi
if $data10 != 3 then
return -1
endi
sql select sum_double(f1) from tb1 interval (1d) limit 2;
if $rows != 2 then
return -1
endi
if $data00 != @21-03-23 00:00:00.000@ then
return -1
endi
if $data01 != 20 then
return -1
endi
if $data10 != @21-03-24 00:00:00.000@ then
return -1
endi
if $data11 != 22 then
return -1
endi
sql_error select ts,sum_double(f1),f1 from tb1;
sql_error select add_one(f1),count(f1) from tb1;
sql_error select sum_double(f1),count(f1) from tb1;
sql_error select add_one(f1),top(f1,3) from tb1;
sql_error select add_one(f1) from tb1 interval(10a);
system sh/exec.sh -n dnode1 -s stop -x SIGINT
tests/script/sh/add_one.c 0 → 100644
浏览文件 @ a591f8d0
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
typedef struct SUdfInit{
int maybe_null; /* 1 if function can return NULL */
int decimals; /* for real functions */
long long length; /* For string functions */
char *ptr; /* free pointer for function data */
int const_item; /* 0 if result is independent of arguments */
} SUdfInit;
void add_one(char* data, char type, int numOfRows, long long* ts, char* dataOutput, char* tsOutput,
int* numOfOutput, SUdfInit* buf) {
int i;
int r = 0;
printf("add_one input data:%p, type:%d, rows:%d, ts:%p,%lld, dataoutput:%p, tsOutput:%p, numOfOutput:%p, buf:%p\n", data, type, numOfRows, ts, *ts, dataOutput, tsOutput, numOfOutput, buf);
if (type == 4) {
for(i=0;i<numOfRows;++i) {
printf("input %d - %d", i, *((int *)data + i));
*((int *)dataOutput+i)=*((int *)data + i) + 1;
printf(", output %d\n", *((int *)dataOutput+i));
if (tsOutput) {
*(long long*)tsOutput=1000000;
}
}
*numOfOutput=numOfRows;
printf("add_one out, numOfOutput:%d\n", *numOfOutput);
}
}
tests/script/sh/prepare_udf.sh
浏览文件 @ a591f8d0
......@@ -4,3 +4,7 @@ echo -n "hello" > /tmp/normal
echo -n "" > /tmp/empty
dd if=/dev/zero bs=3584 of=/tmp/big count=1
rm -rf /tmp/sum_double.so /tmp/add_one.so
gcc -g -O0 -fPIC -shared sh/sum_double.c -o /tmp/sum_double.so
gcc -g -O0 -fPIC -shared sh/add_one.c -o /tmp/add_one.so
tests/script/sh/sum_double.c 0 → 100644
浏览文件 @ a591f8d0
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
typedef struct SUdfInit{
int maybe_null; /* 1 if function can return NULL */
int decimals; /* for real functions */
long long length; /* For string functions */
char *ptr; /* free pointer for function data */
int const_item; /* 0 if result is independent of arguments */
} SUdfInit;
void sum_double(char* data, char type, int numOfRows, long long* ts, char* dataOutput, char* tsOutput,
int* numOfOutput, SUdfInit* buf) {
int i;
int r = 0;
printf("sum_double input data:%p, type:%d, rows:%d, ts:%p,%lld, dataoutput:%p, tsOutput:%p, numOfOutput:%p, buf:%p\n", data, type, numOfRows, ts, *ts, dataOutput, tsOutput, numOfOutput, buf);
if (type == 4) {
r=*(int *)dataOutput;
for(i=0;i<numOfRows;++i) {
r+=*((int *)data + i);
if (tsOutput) {
*(long long*)tsOutput=1000000;
}
}
*(int *)dataOutput=r;
*numOfOutput=1;
printf("sum_double out, dataoutput:%d, numOfOutput:%d\n", *(int *)dataOutput, *numOfOutput);
}
}
void sum_double_finalize(char* dataOutput, int* numOfOutput, SUdfInit* buf) {
int i;
int r = 0;
printf("sum_double_finalize dataoutput:%p, numOfOutput:%p, buf:%p\n", dataOutput, numOfOutput, buf);
*numOfOutput=1;
*(int*)(buf->ptr)=*(int*)dataOutput*2;
*(int*)dataOutput=*(int*)(buf->ptr);
printf("sum_double finalize, dataoutput:%d, numOfOutput:%d\n", *(int *)dataOutput, *numOfOutput);
}
int sum_double_init(SUdfInit* buf) {
buf->maybe_null=1;
buf->ptr = malloc(sizeof(int));
printf("sum_double init\n");
return 0;
}
void sum_double_destroy(SUdfInit* buf) {
free(buf->ptr);
printf("sum_double destroy\n");
}
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或