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

Merge pull request #20781 from taosdata/szhou/replace-function 

feature: udf replace function
This commit is contained in:
Xiaoyu Wang 2023-04-11 17:25:21 +08:00 committed by GitHub
parent 81da6cd4a4 6f35badc78
commit 5ce6220106
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 382 additions and 139 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
include/common/tmsg.h
View File

@ -1093,9 +1093,15 @@ typedef struct {
char* pCode;
} SFuncInfo;
typedef struct {
int32_t funcVersion;
int64_t funcCreatedTime;
} SFuncExtraInfo;
typedef struct {
int32_t numOfFuncs;
SArray* pFuncInfos;
SArray* pFuncExtraInfos;
} SRetrieveFuncRsp;
int32_t tSerializeSRetrieveFuncRsp(void* buf, int32_t bufLen, SRetrieveFuncRsp* pRsp);

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

@ -276,6 +276,8 @@ typedef enum EUdfFuncType { UDF_FUNC_TYPE_SCALAR = 1, UDF_FUNC_TYPE_AGG = 2 } EU
typedef struct SScriptUdfInfo {
const char *name;
int32_t version;
int64_t createdTime;
EUdfFuncType funcType;
int8_t scriptType;

1
source/common/src/systable.c
View File

@ -116,6 +116,7 @@ static const SSysDbTableSchema userFuncSchema[] = {
{.name = "bufsize", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false},
{.name = "func_language", .bytes = TSDB_TYPE_STR_MAX_LEN - 1 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false},
{.name = "func_body", .bytes = TSDB_MAX_BINARY_LEN, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false},
{.name = "func_version", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false},
};
static const SSysDbTableSchema userIdxSchema[] = {

26
source/common/src/tmsg.c
View File

@ -1702,6 +1702,7 @@ int32_t tSerializeSCreateFuncReq(void *buf, int32_t bufLen, SCreateFuncReq *pReq
if (tEncodeCStr(&encoder, pReq->pComment) < 0) return -1;
}
if (tEncodeI8(&encoder, pReq->orReplace) < 0) return -1;
tEndEncode(&encoder);
@ -1746,6 +1747,7 @@ int32_t tDeserializeSCreateFuncReq(void *buf, int32_t bufLen, SCreateFuncReq *pR
if (tDecodeCStrTo(&decoder, pReq->pComment) < 0) return -1;
}
if (!tDecodeIsEnd(&decoder)) {
if (tDecodeI8(&decoder, &pReq->orReplace) < 0) return -1;
} else {
@ -1863,6 +1865,13 @@ int32_t tSerializeSRetrieveFuncRsp(void *buf, int32_t bufLen, SRetrieveFuncRsp *
}
}
if (pRsp->numOfFuncs != (int32_t)taosArrayGetSize(pRsp->pFuncExtraInfos)) return -1;
for (int32_t i = 0; i < pRsp->numOfFuncs; ++i) {
SFuncExtraInfo *extraInfo = taosArrayGet(pRsp->pFuncExtraInfos, i);
if (tEncodeI32(&encoder, extraInfo->funcVersion) < 0) return -1;
if (tEncodeI64(&encoder, extraInfo->funcCreatedTime) < 0) return -1;
}
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
@ -1910,6 +1919,22 @@ int32_t tDeserializeSRetrieveFuncRsp(void *buf, int32_t bufLen, SRetrieveFuncRsp
taosArrayPush(pRsp->pFuncInfos, &fInfo);
}
pRsp->pFuncExtraInfos = taosArrayInit(pRsp->numOfFuncs, sizeof(SFuncExtraInfo));
if (pRsp->pFuncExtraInfos == NULL) return -1;
if (tDecodeIsEnd(&decoder)) {
for (int32_t i = 0; i < pRsp->numOfFuncs; ++i) {
SFuncExtraInfo extraInfo = { 0 };
taosArrayPush(pRsp->pFuncExtraInfos, &extraInfo);
}
} else {
for (int32_t i = 0; i < pRsp->numOfFuncs; ++i) {
SFuncExtraInfo extraInfo = { 0 };
if (tDecodeI32(&decoder, &extraInfo.funcVersion) < 0) return -1;
if (tDecodeI64(&decoder, &extraInfo.funcCreatedTime) < 0) return -1;
taosArrayPush(pRsp->pFuncExtraInfos, &extraInfo);
}
}
tEndDecode(&decoder);
tDecoderClear(&decoder);
@ -1932,6 +1957,7 @@ void tFreeSRetrieveFuncRsp(SRetrieveFuncRsp *pRsp) {
tFreeSFuncInfo(pInfo);
}
taosArrayDestroy(pRsp->pFuncInfos);
taosArrayDestroy(pRsp->pFuncExtraInfos);
}
int32_t tSerializeSTableCfgReq(void *buf, int32_t bufLen, STableCfgReq *pReq) {

2
source/dnode/mnode/impl/inc/mndDef.h
View File

@ -447,6 +447,8 @@ typedef struct {
int32_t codeSize;
char* pComment;
char* pCode;
int32_t funcVersion;
SRWLatch lock;
} SFuncObj;
typedef struct {

94
source/dnode/mnode/impl/src/mndFunc.c
View File

@ -21,7 +21,7 @@
#include "mndTrans.h"
#include "mndUser.h"
#define SDB_FUNC_VER 1
#define SDB_FUNC_VER 2
#define SDB_FUNC_RESERVE_SIZE 64
static SSdbRaw *mndFuncActionEncode(SFuncObj *pFunc);
@ -83,6 +83,7 @@ static SSdbRaw *mndFuncActionEncode(SFuncObj *pFunc) {
SDB_SET_BINARY(pRaw, dataPos, pFunc->pComment, pFunc->commentSize, _OVER)
}
SDB_SET_BINARY(pRaw, dataPos, pFunc->pCode, pFunc->codeSize, _OVER)
SDB_SET_INT32(pRaw, dataPos, pFunc->funcVersion, _OVER)
SDB_SET_RESERVE(pRaw, dataPos, SDB_FUNC_RESERVE_SIZE, _OVER)
SDB_SET_DATALEN(pRaw, dataPos, _OVER);
@ -107,7 +108,7 @@ static SSdbRow *mndFuncActionDecode(SSdbRaw *pRaw) {
int8_t sver = 0;
if (sdbGetRawSoftVer(pRaw, &sver) != 0) goto _OVER;
if (sver != SDB_FUNC_VER) {
if (sver != 1 && sver != 2) {
terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
goto _OVER;
}
@ -144,8 +145,15 @@ static SSdbRow *mndFuncActionDecode(SSdbRaw *pRaw) {
goto _OVER;
}
SDB_GET_BINARY(pRaw, dataPos, pFunc->pCode, pFunc->codeSize, _OVER)
if (sver >= 2) {
SDB_GET_INT32(pRaw, dataPos, &pFunc->funcVersion, _OVER)
}
SDB_GET_RESERVE(pRaw, dataPos, SDB_FUNC_RESERVE_SIZE, _OVER)
taosInitRWLatch(&pFunc->lock);
terrno = 0;
_OVER:
@ -173,6 +181,44 @@ static int32_t mndFuncActionDelete(SSdb *pSdb, SFuncObj *pFunc) {
static int32_t mndFuncActionUpdate(SSdb *pSdb, SFuncObj *pOld, SFuncObj *pNew) {
mTrace("func:%s, perform update action, old row:%p new row:%p", pOld->name, pOld, pNew);
taosWLockLatch(&pOld->lock);
pOld->align = pNew->align;
pOld->bufSize = pNew->bufSize;
pOld->codeSize = pNew->codeSize;
pOld->commentSize = pNew->commentSize;
pOld->createdTime = pNew->createdTime;
pOld->funcType = pNew->funcType;
pOld->funcVersion = pNew->funcVersion;
pOld->outputLen = pNew->outputLen;
pOld->outputType = pNew->outputType;
if (pOld->pComment != NULL) {
taosMemoryFree(pOld->pComment);
pOld->pComment = NULL;
}
if (pNew->commentSize > 0 && pNew->pComment != NULL) {
pOld->commentSize = pNew->commentSize;
pOld->pComment = taosMemoryMalloc(pOld->commentSize);
memcpy(pOld->pComment, pNew->pComment, pOld->commentSize);
}
if (pOld->pCode != NULL) {
taosMemoryFree(pOld->pCode);
pOld->pCode = NULL;
}
if (pNew->codeSize > 0 && pNew->pCode != NULL) {
pOld->codeSize = pNew->codeSize;
pOld->pCode = taosMemoryMalloc(pOld->codeSize);
memcpy(pOld->pCode, pNew->pCode, pOld->codeSize);
}
pOld->scriptType = pNew->scriptType;
pOld->signature = pNew->signature;
taosWUnLockLatch(&pOld->lock);
return 0;
}
@ -225,9 +271,25 @@ static int32_t mndCreateFunc(SMnode *pMnode, SRpcMsg *pReq, SCreateFuncReq *pCre
pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pReq, "create-func");
if (pTrans == NULL) goto _OVER;
mInfo("trans:%d, used to create func:%s", pTrans->id, pCreate->name);
SFuncObj *oldFunc = mndAcquireFunc(pMnode, pCreate->name);
if (pCreate->orReplace == 1 && oldFunc != NULL) {
func.funcVersion = oldFunc->funcVersion + 1;
func.createdTime = oldFunc->createdTime;
SSdbRaw *pRedoRaw = mndFuncActionEncode(oldFunc);
if (pRedoRaw == NULL || mndTransAppendRedolog(pTrans, pRedoRaw) != 0) goto _OVER;
if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY) != 0) goto _OVER;
SSdbRaw *pUndoRaw = mndFuncActionEncode(oldFunc);
if (pUndoRaw == NULL || mndTransAppendUndolog(pTrans, pUndoRaw) != 0) goto _OVER;
if (sdbSetRawStatus(pUndoRaw, SDB_STATUS_READY) != 0) goto _OVER;
SSdbRaw *pCommitRaw = mndFuncActionEncode(&func);
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) goto _OVER;
if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY) != 0) goto _OVER;
} else {
SSdbRaw *pRedoRaw = mndFuncActionEncode(&func);
if (pRedoRaw == NULL || mndTransAppendRedolog(pTrans, pRedoRaw) != 0) goto _OVER;
if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_CREATING) != 0) goto _OVER;
@ -239,12 +301,17 @@ static int32_t mndCreateFunc(SMnode *pMnode, SRpcMsg *pReq, SCreateFuncReq *pCre
SSdbRaw *pCommitRaw = mndFuncActionEncode(&func);
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) goto _OVER;
if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY) != 0) goto _OVER;
}
if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
code = 0;
_OVER:
if (oldFunc != NULL) {
mndReleaseFunc(pMnode, oldFunc);
}
taosMemoryFree(func.pCode);
taosMemoryFree(func.pComment);
mndTransDrop(pTrans);
@ -304,6 +371,9 @@ static int32_t mndProcessCreateFuncReq(SRpcMsg *pReq) {
mInfo("func:%s, already exist, ignore exist is set", createReq.name);
code = 0;
goto _OVER;
} else if (createReq.orReplace) {
mInfo("func:%s, replace function is set", createReq.name);
code = 0;
} else {
terrno = TSDB_CODE_MND_FUNC_ALREADY_EXIST;
goto _OVER;
@ -413,6 +483,12 @@ static int32_t mndProcessRetrieveFuncReq(SRpcMsg *pReq) {
goto RETRIEVE_FUNC_OVER;
}
retrieveRsp.pFuncExtraInfos = taosArrayInit(retrieveReq.numOfFuncs, sizeof(SFuncExtraInfo));
if (retrieveRsp.pFuncExtraInfos == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto RETRIEVE_FUNC_OVER;
}
for (int32_t i = 0; i < retrieveReq.numOfFuncs; ++i) {
char *funcName = taosArrayGet(retrieveReq.pFuncNames, i);
@ -451,6 +527,11 @@ static int32_t mndProcessRetrieveFuncReq(SRpcMsg *pReq) {
}
}
taosArrayPush(retrieveRsp.pFuncInfos, &funcInfo);
SFuncExtraInfo extraInfo = {0};
extraInfo.funcVersion = pFunc->funcVersion;
extraInfo.funcCreatedTime = pFunc->createdTime;
taosArrayPush(retrieveRsp.pFuncExtraInfos, &extraInfo);
mndReleaseFunc(pMnode, pFunc);
}
@ -559,13 +640,18 @@ static int32_t mndRetrieveFuncs(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
colDataSetVal(pColInfo, numOfRows, (const char *)varLang, false);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
int32_t varCodeLen = (pFunc->codeSize + VARSTR_HEADER_SIZE) > TSDB_MAX_BINARY_LEN ? TSDB_MAX_BINARY_LEN : pFunc->codeSize + VARSTR_HEADER_SIZE;
int32_t varCodeLen = (pFunc->codeSize + VARSTR_HEADER_SIZE) > TSDB_MAX_BINARY_LEN
? TSDB_MAX_BINARY_LEN
: pFunc->codeSize + VARSTR_HEADER_SIZE;
char *b4 = taosMemoryMalloc(varCodeLen);
memcpy(varDataVal(b4), pFunc->pCode, varCodeLen - VARSTR_HEADER_SIZE);
varDataSetLen(b4, varCodeLen - VARSTR_HEADER_SIZE);
colDataSetVal(pColInfo, numOfRows, (const char *)b4, false);
taosMemoryFree(b4);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)&pFunc->funcVersion, false);
numOfRows++;
sdbRelease(pSdb, pFunc);
}

5
source/libs/catalog/test/catalogTests.cpp
View File

@ -672,11 +672,16 @@ void ctgTestRspUdfInfo(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pR
SRetrieveFuncRsp funcRsp = {0};
funcRsp.numOfFuncs = 1;
funcRsp.pFuncInfos = taosArrayInit(1, sizeof(SFuncInfo));
funcRsp.pFuncExtraInfos = taosArrayInit(1, sizeof(SFuncExtraInfo));
SFuncInfo funcInfo = {0};
strcpy(funcInfo.name, "func1");
funcInfo.funcType = ctgTestFuncType;
(void)taosArrayPush(funcRsp.pFuncInfos, &funcInfo);
SFuncExtraInfo extraInfo = {0};
extraInfo.funcVersion = 0;
extraInfo.funcCreatedTime = taosGetTimestampMs();
(void)taosArrayPush(funcRsp.pFuncExtraInfos, &extraInfo);
int32_t rspLen = tSerializeSRetrieveFuncRsp(NULL, 0, &funcRsp);
void *pReq = rpcMallocCont(rspLen);

5
source/libs/function/src/tudf.c
View File

@ -1400,9 +1400,10 @@ void udfcUvHandleError(SClientUvConn *conn) {
QUEUE_REMOVE(&task->procTaskQueue);
uv_sem_post(&task->taskSem);
}
if (!uv_is_closing((uv_handle_t *)conn->pipe)) {
uv_close((uv_handle_t *)conn->pipe, onUdfcPipeClose);
}
}
void onUdfcPipeRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf) {
fnDebug("udfc client %p, client read from pipe. nread: %zd", client, nread);
@ -1553,7 +1554,9 @@ int32_t udfcStartUvTask(SClientUvTaskNode *uvTask) {
} else {
SClientUvConn *conn = pipe->data;
QUEUE_INSERT_TAIL(&conn->taskQueue, &uvTask->connTaskQueue);
if (!uv_is_closing((uv_handle_t *)uvTask->pipe)) {
uv_close((uv_handle_t *)uvTask->pipe, onUdfcPipeClose);
}
code = 0;
}
break;

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

@ -53,15 +53,7 @@ int32_t udfdCPluginOpen(SScriptUdfEnvItem *items, int numItems) { return 0; }
int32_t udfdCPluginClose() { return 0; }
int32_t udfdCPluginUdfInit(SScriptUdfInfo *udf, void **pUdfCtx) {
int32_t err = 0;
SUdfCPluginCtx *udfCtx = taosMemoryCalloc(1, sizeof(SUdfCPluginCtx));
err = uv_dlopen(udf->path, &udfCtx->lib);
if (err != 0) {
fnError("can not load library %s. error: %s", udf->path, uv_strerror(err));
return TSDB_CODE_UDF_LOAD_UDF_FAILURE;
}
const char *udfName = udf->name;
const char *udfdCPluginUdfInitLoadInitDestoryFuncs(SUdfCPluginCtx *udfCtx, const char *udfName) {
char initFuncName[TSDB_FUNC_NAME_LEN + 5] = {0};
char *initSuffix = "_init";
strcpy(initFuncName, udfName);
@ -73,34 +65,55 @@ int32_t udfdCPluginUdfInit(SScriptUdfInfo *udf, void **pUdfCtx) {
strcpy(destroyFuncName, udfName);
strncat(destroyFuncName, destroySuffix, strlen(destroySuffix));
uv_dlsym(&udfCtx->lib, destroyFuncName, (void **)(&udfCtx->destroyFunc));
return udfName;
}
if (udf->funcType == UDF_FUNC_TYPE_SCALAR) {
char processFuncName[TSDB_FUNC_NAME_LEN] = {0};
strcpy(processFuncName, udfName);
uv_dlsym(&udfCtx->lib, processFuncName, (void **)(&udfCtx->scalarProcFunc));
} else if (udf->funcType == UDF_FUNC_TYPE_AGG) {
void udfdCPluginUdfInitLoadAggFuncs(SUdfCPluginCtx *udfCtx, const char *udfName) {
char processFuncName[TSDB_FUNC_NAME_LEN] = {0};
strcpy(processFuncName, udfName);
uv_dlsym(&udfCtx->lib, processFuncName, (void **)(&udfCtx->aggProcFunc));
char startFuncName[TSDB_FUNC_NAME_LEN + 6] = {0};
char *startSuffix = "_start";
strncpy(startFuncName, processFuncName, sizeof(startFuncName));
strncat(startFuncName, startSuffix, strlen(startSuffix));
uv_dlsym(&udfCtx->lib, startFuncName, (void **)(&udfCtx->aggStartFunc));
char finishFuncName[TSDB_FUNC_NAME_LEN + 7] = {0};
char *finishSuffix = "_finish";
strncpy(finishFuncName, processFuncName, sizeof(finishFuncName));
strncat(finishFuncName, finishSuffix, strlen(finishSuffix));
uv_dlsym(&udfCtx->lib, finishFuncName, (void **)(&udfCtx->aggFinishFunc));
char mergeFuncName[TSDB_FUNC_NAME_LEN + 6] = {0};
char *mergeSuffix = "_merge";
strncpy(mergeFuncName, processFuncName, sizeof(mergeFuncName));
strncat(mergeFuncName, mergeSuffix, strlen(mergeSuffix));
uv_dlsym(&udfCtx->lib, mergeFuncName, (void **)(&udfCtx->aggMergeFunc));
}
int32_t udfdCPluginUdfInit(SScriptUdfInfo *udf, void **pUdfCtx) {
int32_t err = 0;
SUdfCPluginCtx *udfCtx = taosMemoryCalloc(1, sizeof(SUdfCPluginCtx));
err = uv_dlopen(udf->path, &udfCtx->lib);
if (err != 0) {
fnError("can not load library %s. error: %s", udf->path, uv_strerror(err));
return TSDB_CODE_UDF_LOAD_UDF_FAILURE;
}
const char *udfName = udf->name;
udfdCPluginUdfInitLoadInitDestoryFuncs(udfCtx, udfName);
if (udf->funcType == UDF_FUNC_TYPE_SCALAR) {
char processFuncName[TSDB_FUNC_NAME_LEN] = {0};
strcpy(processFuncName, udfName);
uv_dlsym(&udfCtx->lib, processFuncName, (void **)(&udfCtx->scalarProcFunc));
} else if (udf->funcType == UDF_FUNC_TYPE_AGG) {
udfdCPluginUdfInitLoadAggFuncs(udfCtx, udfName);
}
int32_t code = 0;
if (udfCtx->initFunc) {
// TODO: handle init call return error
code = (udfCtx->initFunc)();
if (code != 0) {
uv_dlclose(&udfCtx->lib);
@ -216,6 +229,7 @@ typedef struct SUdfdContext {
SArray *residentFuncs;
char udfDataDir[PATH_MAX];
bool printVersion;
} SUdfdContext;
@ -246,6 +260,8 @@ typedef enum { UDF_STATE_INIT = 0, UDF_STATE_LOADING, UDF_STATE_READY } EUdfStat
typedef struct SUdf {
char name[TSDB_FUNC_NAME_LEN + 1];
int32_t version;
int64_t createdTime;
int8_t funcType;
int8_t scriptType;
@ -263,9 +279,11 @@ typedef struct SUdf {
SUdfScriptPlugin *scriptPlugin;
void *scriptUdfCtx;
int64_t lastFetchTime; // last fetch time in milliseconds
bool expired;
} SUdf;
// TODO: add private udf structure.
typedef struct SUdfcFuncHandle {
SUdf *udf;
} SUdfcFuncHandle;
@ -318,6 +336,9 @@ static void udfdCloseWalkCb(uv_handle_t *handle, void *arg);
static int32_t udfdRun();
static void udfdConnectMnodeThreadFunc(void *args);
SUdf *udfdNewUdf(const char *udfName);
void udfdGetFuncBodyPath(const SUdf *udf, char *path);
void udfdInitializeCPlugin(SUdfScriptPlugin *plugin) {
plugin->scriptType = TSDB_FUNC_SCRIPT_BIN_LIB;
plugin->openFunc = udfdCPluginOpen;
@ -370,12 +391,13 @@ int32_t udfdInitializePythonPlugin(SUdfScriptPlugin *plugin) {
}
if (plugin->openFunc) {
int16_t lenPythonPath = strlen(tsUdfdLdLibPath) + strlen(tsTempDir) + 1 + 1; // tsTempDir:tsUdfdLdLibPath
int16_t lenPythonPath =
strlen(tsUdfdLdLibPath) + strlen(global.udfDataDir) + 1 + 1; // global.udfDataDir:tsUdfdLdLibPath
char *pythonPath = taosMemoryMalloc(lenPythonPath);
#ifdef WINDOWS
snprintf(pythonPath, lenPythonPath, "%s;%s", tsTempDir, tsUdfdLdLibPath);
snprintf(pythonPath, lenPythonPath, "%s;%s", global.udfDataDir, tsUdfdLdLibPath);
#else
snprintf(pythonPath, lenPythonPath, "%s:%s", tsTempDir, tsUdfdLdLibPath);
snprintf(pythonPath, lenPythonPath, "%s:%s", global.udfDataDir, tsUdfdLdLibPath);
#endif
SScriptUdfEnvItem items[] = {{"PYTHONPATH", pythonPath}, {"LOGDIR", tsLogDir}};
err = plugin->openFunc(items, 2);
@ -500,28 +522,14 @@ void convertUdf2UdfInfo(SUdf *udf, SScriptUdfInfo *udfInfo) {
udfInfo->funcType = UDF_FUNC_TYPE_SCALAR;
}
udfInfo->name = udf->name;
udfInfo->version = udf->version;
udfInfo->createdTime = udf->createdTime;
udfInfo->outputLen = udf->outputLen;
udfInfo->outputType = udf->outputType;
udfInfo->path = udf->path;
udfInfo->scriptType = udf->scriptType;
}
int32_t udfdRenameUdfFile(SUdf *udf) {
char newPath[PATH_MAX];
if (udf->scriptType == TSDB_FUNC_SCRIPT_BIN_LIB) {
snprintf(newPath, PATH_MAX, "%s/lib%s.so", tsTempDir, udf->name);
} else if (udf->scriptType == TSDB_FUNC_SCRIPT_PYTHON) {
snprintf(newPath, PATH_MAX, "%s/%s.py", tsTempDir, udf->name);
} else {
return TSDB_CODE_UDF_SCRIPT_NOT_SUPPORTED;
}
int32_t code = taosRenameFile(udf->path, newPath);
if (code == 0) {
sprintf(udf->path, "%s", newPath);
}
return 0;
}
int32_t udfdInitUdf(char *udfName, SUdf *udf) {
int32_t err = 0;
err = udfdFillUdfInfoFromMNode(global.clientRpc, udfName, udf);
@ -546,8 +554,6 @@ int32_t udfdInitUdf(char *udfName, SUdf *udf) {
uv_mutex_unlock(&global.scriptPluginsMutex);
udf->scriptPlugin = global.scriptPlugins[udf->scriptType];
udfdRenameUdfFile(udf);
SScriptUdfInfo info = {0};
convertUdf2UdfInfo(udf, &info);
err = udf->scriptPlugin->udfInitFunc(&info, &udf->scriptUdfCtx);
@ -560,36 +566,56 @@ int32_t udfdInitUdf(char *udfName, SUdf *udf) {
return 0;
}
SUdf *udfdGetOrCreateUdf(const char *udfName) {
SUdf *udf = NULL;
uv_mutex_lock(&global.udfsMutex);
SUdf **udfInHash = taosHashGet(global.udfsHash, udfName, strlen(udfName));
if (udfInHash) {
++(*udfInHash)->refCount;
udf = *udfInHash;
uv_mutex_unlock(&global.udfsMutex);
} else {
SUdf *udfdNewUdf(const char *udfName) {
SUdf *udfNew = taosMemoryCalloc(1, sizeof(SUdf));
udfNew->refCount = 1;
udfNew->lastFetchTime = taosGetTimestampMs();
strncpy(udfNew->name, udfName, TSDB_FUNC_NAME_LEN);
udfNew->state = UDF_STATE_INIT;
uv_mutex_init(&udfNew->lock);
uv_cond_init(&udfNew->condReady);
udf = udfNew;
udf->resident = false;
udfNew->resident = false;
udfNew->expired = false;
for (int32_t i = 0; i < taosArrayGetSize(global.residentFuncs); ++i) {
char *funcName = taosArrayGet(global.residentFuncs, i);
if (strcmp(udfName, funcName) == 0) {
udf->resident = true;
udfNew->resident = true;
break;
}
}
return udfNew;
}
SUdf *udfdGetOrCreateUdf(const char *udfName) {
uv_mutex_lock(&global.udfsMutex);
SUdf **pUdfHash = taosHashGet(global.udfsHash, udfName, strlen(udfName));
int64_t currTime = taosGetTimestampSec();
bool expired = false;
if (pUdfHash) {
expired = currTime - (*pUdfHash)->lastFetchTime > 10 * 1000; // 10s
if (!expired) {
++(*pUdfHash)->refCount;
SUdf *udf = *pUdfHash;
uv_mutex_unlock(&global.udfsMutex);
fnInfo("udfd reuse existing udf. udf %s udf version %d, udf created time %" PRIx64, udf->name, udf->version,
udf->createdTime);
return udf;
} else {
(*pUdfHash)->expired = true;
taosHashRemove(global.udfsHash, udfName, strlen(udfName));
fnInfo("udfd expired, check for new version. existing udf %s udf version %d, udf created time %" PRIx64,
(*pUdfHash)->name, (*pUdfHash)->version, (*pUdfHash)->createdTime);
}
}
SUdf *udf = udfdNewUdf(udfName);
SUdf **pUdf = &udf;
taosHashPut(global.udfsHash, udfName, strlen(udfName), pUdf, POINTER_BYTES);
uv_mutex_unlock(&global.udfsMutex);
}
return udf;
}
@ -760,7 +786,7 @@ void udfdProcessTeardownRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
uv_mutex_lock(&global.udfsMutex);
udf->refCount--;
if (udf->refCount == 0 && !udf->resident) {
if (udf->refCount == 0 && (!udf->resident || udf->expired)) {
unloadUdf = true;
taosHashRemove(global.udfsHash, udf->name, strlen(udf->name));
}
@ -791,6 +817,61 @@ void udfdProcessTeardownRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
return;
}
void udfdGetFuncBodyPath(const SUdf *udf, char *path) {
if (udf->scriptType == TSDB_FUNC_SCRIPT_BIN_LIB) {
#ifdef WINDOWS
snprintf(path, PATH_MAX, "%s%s_%d_%" PRIx64 ".dll", global.udfDataDir, udf->name, udf->version, udf->createdTime);
#else
snprintf(path, PATH_MAX, "%s/lib%s_%d_%" PRIx64 ".so", global.udfDataDir, udf->name, udf->version,
udf->createdTime);
#endif
} else if (udf->scriptType == TSDB_FUNC_SCRIPT_PYTHON) {
#ifdef WINDOWS
snprintf(path, PATH_MAX, "%s%s_%d_%" PRIx64 ".py", global.udfDataDir, udf->name, udf->version, udf->createdTime);
#else
snprintf(path, PATH_MAX, "%s/%s_%d_%" PRIx64 ".py", global.udfDataDir, udf->name, udf->version, udf->createdTime);
#endif
} else {
#ifdef WINDOWS
snprintf(path, PATH_MAX, "%s%s_%d_%" PRIx64, global.udfDataDir, udf->name, udf->version, udf->createdTime);
#else
snprintf(path, PATH_MAX, "%s/lib%s_%d_%" PRIx64, global.udfDataDir, udf->name, udf->version, udf->createdTime);
#endif
}
}
int32_t udfdSaveFuncBodyToFile(SFuncInfo *pFuncInfo, SUdf *udf) {
if (!osDataSpaceAvailable()) {
terrno = TSDB_CODE_NO_AVAIL_DISK;
fnError("udfd create shared library failed since %s", terrstr(terrno));
return terrno;
}
char path[PATH_MAX] = {0};
udfdGetFuncBodyPath(udf, path);
bool fileExist = !(taosStatFile(path, NULL, NULL) < 0);
if (fileExist) {
strncpy(udf->path, path, PATH_MAX);
fnInfo("udfd func body file. reuse existing file %s", path);
return TSDB_CODE_SUCCESS;
}
TdFilePtr file = taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC);
if (file == NULL) {
fnError("udfd write udf shared library: %s failed, error: %d %s", path, errno, strerror(errno));
return TSDB_CODE_FILE_CORRUPTED;
}
int64_t count = taosWriteFile(file, pFuncInfo->pCode, pFuncInfo->codeSize);
if (count != pFuncInfo->codeSize) {
fnError("udfd write udf shared library failed");
return TSDB_CODE_FILE_CORRUPTED;
}
taosCloseFile(&file);
strncpy(udf->path, path, PATH_MAX);
return TSDB_CODE_SUCCESS;
}
void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
SUdfdRpcSendRecvInfo *msgInfo = (SUdfdRpcSendRecvInfo *)pMsg->info.ahandle;
@ -829,11 +910,8 @@ void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
} else if (msgInfo->rpcType == UDFD_RPC_RETRIVE_FUNC) {
SRetrieveFuncRsp retrieveRsp = {0};
tDeserializeSRetrieveFuncRsp(pMsg->pCont, pMsg->contLen, &retrieveRsp);
if (retrieveRsp.pFuncInfos == NULL) {
goto _return;
}
SFuncInfo *pFuncInfo = (SFuncInfo *)taosArrayGet(retrieveRsp.pFuncInfos, 0);
// SUdf *udf = msgInfo->param;
SUdf *udf = msgInfo->param;
udf->funcType = pFuncInfo->funcType;
udf->scriptType = pFuncInfo->scriptType;
@ -841,37 +919,16 @@ void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
udf->outputLen = pFuncInfo->outputLen;
udf->bufSize = pFuncInfo->bufSize;
if (!osTempSpaceAvailable()) {
terrno = TSDB_CODE_NO_AVAIL_DISK;
msgInfo->code = terrno;
fnError("udfd create shared library failed since %s", terrstr(terrno));
goto _return;
SFuncExtraInfo *pFuncExtraInfo = (SFuncExtraInfo *)taosArrayGet(retrieveRsp.pFuncExtraInfos, 0);
udf->version = pFuncExtraInfo->funcVersion;
udf->createdTime = pFuncExtraInfo->funcCreatedTime;
msgInfo->code = udfdSaveFuncBodyToFile(pFuncInfo, udf);
if (msgInfo->code != 0) {
udf->lastFetchTime = 0;
}
char path[PATH_MAX] = {0};
#ifdef WINDOWS
snprintf(path, sizeof(path), "%s%s", tsTempDir, pFuncInfo->name);
#else
snprintf(path, sizeof(path), "%s/%s", tsTempDir, pFuncInfo->name);
#endif
TdFilePtr file = taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC);
if (file == NULL) {
fnError("udfd write udf shared library: %s failed, error: %d %s", path, errno, strerror(errno));
msgInfo->code = TSDB_CODE_FILE_CORRUPTED;
goto _return;
}
int64_t count = taosWriteFile(file, pFuncInfo->pCode, pFuncInfo->codeSize);
if (count != pFuncInfo->codeSize) {
fnError("udfd write udf shared library failed");
msgInfo->code = TSDB_CODE_FILE_CORRUPTED;
goto _return;
}
taosCloseFile(&file);
strncpy(udf->path, path, PATH_MAX);
tFreeSFuncInfo(pFuncInfo);
taosArrayDestroy(retrieveRsp.pFuncInfos);
msgInfo->code = 0;
taosArrayDestroy(retrieveRsp.pFuncExtraInfos);
}
_return:
@ -1380,6 +1437,24 @@ int32_t udfdCleanup() {
return 0;
}
int32_t udfdCreateUdfSourceDir() {
snprintf(global.udfDataDir, PATH_MAX, "%s/.udf", tsDataDir);
int32_t code = taosMkDir(global.udfDataDir);
if (code != TSDB_CODE_SUCCESS) {
snprintf(global.udfDataDir, PATH_MAX, "%s/.udf", tsTempDir);
code = taosMkDir(global.udfDataDir);
}
fnInfo("udfd create udf source directory %s. result: %s", global.udfDataDir, tstrerror(code));
return code;
}
int32_t udfdDestroyUdfSourceDir() {
fnInfo("destory udf source directory %s", global.udfDataDir);
taosRemoveDir(global.udfDataDir);
return 0;
}
int main(int argc, char *argv[]) {
if (!taosCheckSystemIsLittleEnd()) {
printf("failed to start since on non-little-end machines\n");
@ -1408,10 +1483,15 @@ int main(int argc, char *argv[]) {
initEpSetFromCfg(tsFirst, tsSecond, &global.mgmtEp);
if (udfdOpenClientRpc() != 0) {
fnError("open rpc connection to mnode failure");
fnError("open rpc connection to mnode failed");
return -3;
}
if (udfdCreateUdfSourceDir() != 0) {
fnError("create udf source directory failed");
return -4;
}
if (udfdUvInit() != 0) {
fnError("uv init failure");
return -5;
@ -1425,6 +1505,7 @@ int main(int argc, char *argv[]) {
udfdRun();
removeListeningPipe();
udfdDestroyUdfSourceDir();
udfdCloseClientRpc();
udfdDeinitResidentFuncs();

1
source/libs/qcom/src/querymsg.c
View File

@ -587,6 +587,7 @@ int32_t queryProcessRetrieveFuncRsp(void *output, char *msg, int32_t msgSize) {
memcpy(output, funcInfo, sizeof(*funcInfo));
taosArrayDestroy(out.pFuncInfos);
taosArrayDestroy(out.pFuncExtraInfos);
return TSDB_CODE_SUCCESS;
}

30
tests/script/tsim/query/udfpy.sim
View File

@ -280,7 +280,37 @@ if $data20 != 8.000000000 then
return -1
endi
sql create or replace function bit_and as '/tmp/udf/libbitand.so' outputtype int
sql select func_version from information_schema.ins_functions where name='bit_and'
if $data00 != 1 then
return -1
endi
sql select bit_and(f1, f2) from t2;
print $rows , $data00 , $data10 , $data20 , $data30 , $data40 , $data50
if $rows != 6 then
return -1
endi
if $data00 != 0 then
return -1
endi
if $data10 != 1 then
return -1
endi
if $data20 != NULL then
return -1
endi
if $data30 != NULL then
return -1
endi
if $data40 != 0 then
return -1
endi
if $data50 != 1 then
return -1
endi
#sql drop function bit_and;
#sql show functions;
#if $rows != 1 then

2
tests/system-test/2-query/odbc.py
View File

@ -22,7 +22,7 @@ class TDTestCase:
tdSql.execute("insert into db.ctb using db.stb tags(1) (ts, c1) values (now, 1)")
tdSql.query("select count(*) from information_schema.ins_columns")
tdSql.checkData(0, 0, 274)
tdSql.checkData(0, 0, 275)
tdSql.query("select * from information_schema.ins_columns where table_name = 'ntb'")
tdSql.checkRows(14)