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

fix: change script udf info structure

This commit is contained in:
slzhou 2023-02-21 17:09:26 +08:00
parent 89db7bf600
commit 8e5dae3970
2 changed files with 19 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
include/libs/function/taosudf.h
View File

@ -262,17 +262,22 @@ typedef int32_t (*TUdfAggFinishFunc)(SUdfInterBuf *buf, SUdfInterBuf *resultData
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
typedef struct SUdfInfo { typedef enum EUdfFuncType {
UDF_FUNC_TYPE_SCALAR = 1,
UDF_FUNC_TYPE_AGG = 2
} EUdfFuncType;
typedef struct SScriptUdfInfo {
char *name; char *name;
int8_t funcType; EUdfFuncType funcType;
int8_t scriptType; int8_t scriptType;
int8_t outputType; int8_t outputType;
int32_t outputLen; int32_t outputLen;
int32_t bufSize; int32_t bufSize;
char *path; char *path;
} SUdfInfo; } SScriptUdfInfo;
typedef int32_t (*TScriptUdfScalarProcFunc)(SUdfDataBlock *block, SUdfColumn *resultCol, void *udfCtx); typedef int32_t (*TScriptUdfScalarProcFunc)(SUdfDataBlock *block, SUdfColumn *resultCol, void *udfCtx);
@ -282,7 +287,7 @@ typedef int32_t (*TScriptUdfAggProcessFunc)(SUdfDataBlock *block, SUdfInterBuf *
typedef int32_t (*TScriptUdfAggMergeFunc)(SUdfInterBuf *inputBuf1, SUdfInterBuf *inputBuf2, SUdfInterBuf *outputBuf, typedef int32_t (*TScriptUdfAggMergeFunc)(SUdfInterBuf *inputBuf1, SUdfInterBuf *inputBuf2, SUdfInterBuf *outputBuf,
void *udfCtx); void *udfCtx);
typedef int32_t (*TScriptUdfAggFinishFunc)(SUdfInterBuf *buf, SUdfInterBuf *resultData, void *udfCtx); typedef int32_t (*TScriptUdfAggFinishFunc)(SUdfInterBuf *buf, SUdfInterBuf *resultData, void *udfCtx);
typedef int32_t (*TScriptUdfInitFunc)(SUdfInfo *info, void **pUdfCtx); typedef int32_t (*TScriptUdfInitFunc)(SScriptUdfInfo *info, void **pUdfCtx);
typedef int32_t (*TScriptUdfDestoryFunc)(void *udfCtx); typedef int32_t (*TScriptUdfDestoryFunc)(void *udfCtx);
// the following function is for open/close script plugin. // the following function is for open/close script plugin.

16
source/libs/function/src/udfd.c
View File

@ -52,7 +52,7 @@ int32_t udfdCPluginOpen(void *scriptCtx) { return 0; }
int32_t udfdCPluginClose(void *scriptCtx) { return 0; } int32_t udfdCPluginClose(void *scriptCtx) { return 0; }
int32_t udfdCPluginUdfInit(SUdfInfo *udf, void **pUdfCtx) { int32_t udfdCPluginUdfInit(SScriptUdfInfo *udf, void **pUdfCtx) {
int32_t err = 0; int32_t err = 0;
SUdfCPluginCtx *udfCtx = taosMemoryCalloc(1, sizeof(SUdfCPluginCtx)); SUdfCPluginCtx *udfCtx = taosMemoryCalloc(1, sizeof(SUdfCPluginCtx));
err = uv_dlopen(udf->path, &udfCtx->lib); err = uv_dlopen(udf->path, &udfCtx->lib);
@ -73,11 +73,11 @@ int32_t udfdCPluginUdfInit(SUdfInfo *udf, void **pUdfCtx) {
strncat(destroyFuncName, destroySuffix, strlen(destroySuffix)); strncat(destroyFuncName, destroySuffix, strlen(destroySuffix));
uv_dlsym(&udfCtx->lib, destroyFuncName, (void **)(&udfCtx->destroyFunc)); uv_dlsym(&udfCtx->lib, destroyFuncName, (void **)(&udfCtx->destroyFunc));
if (udf->funcType == TSDB_FUNC_TYPE_SCALAR) { if (udf->funcType == UDF_FUNC_TYPE_SCALAR) {
char processFuncName[TSDB_FUNC_NAME_LEN] = {0}; char processFuncName[TSDB_FUNC_NAME_LEN] = {0};
strcpy(processFuncName, udfName); strcpy(processFuncName, udfName);
uv_dlsym(&udfCtx->lib, processFuncName, (void **)(&udfCtx->scalarProcFunc)); uv_dlsym(&udfCtx->lib, processFuncName, (void **)(&udfCtx->scalarProcFunc));
} else if (udf->funcType == TSDB_FUNC_TYPE_AGGREGATE) { } else if (udf->funcType == UDF_FUNC_TYPE_AGG) {
char processFuncName[TSDB_FUNC_NAME_LEN] = {0}; char processFuncName[TSDB_FUNC_NAME_LEN] = {0};
strcpy(processFuncName, udfName); strcpy(processFuncName, udfName);
uv_dlsym(&udfCtx->lib, processFuncName, (void **)(&udfCtx->aggProcFunc)); uv_dlsym(&udfCtx->lib, processFuncName, (void **)(&udfCtx->aggProcFunc));
@ -411,9 +411,13 @@ void udfdProcessRequest(uv_work_t *req) {
} }
} }
void convertUdf2UdfInfo(SUdf *udf, SUdfInfo *udfInfo) { void convertUdf2UdfInfo(SUdf *udf, SScriptUdfInfo *udfInfo) {
udfInfo->bufSize = udf->bufSize; udfInfo->bufSize = udf->bufSize;
udfInfo->funcType = udf->funcType; if (udf->funcType == TSDB_FUNC_TYPE_AGGREGATE) {
udfInfo->funcType = UDF_FUNC_TYPE_AGG;
} else if (udf->funcType == TSDB_FUNC_TYPE_SCALAR) {
udfInfo->funcType = UDF_FUNC_TYPE_SCALAR;
}
udfInfo->name = udf->name; udfInfo->name = udf->name;
udfInfo->outputLen = udf->outputLen; udfInfo->outputLen = udf->outputLen;
udfInfo->outputType = udf->outputType; udfInfo->outputType = udf->outputType;
@ -438,7 +442,7 @@ int32_t udfdInitUdf(char *udfName, SUdf *udf) {
} }
uv_mutex_unlock(&global.scriptPluginsMutex); uv_mutex_unlock(&global.scriptPluginsMutex);
udf->scriptPlugin = scriptPlugin; udf->scriptPlugin = scriptPlugin;
SUdfInfo info = {0}; SScriptUdfInfo info = {0};
convertUdf2UdfInfo(udf, &info); convertUdf2UdfInfo(udf, &info);
udf->scriptPlugin->udfInitFunc(&info, &udf->scriptUdfCtx); udf->scriptPlugin->udfInitFunc(&info, &udf->scriptUdfCtx);
return 0; return 0;