diff --git a/include/common/tmsg.h b/include/common/tmsg.h index d7f9e16d87b058709ddcc89d7a8487d58d76ee54..5d732c7c6f22fd49f6a261599f230471fe98d458 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -1054,6 +1054,7 @@ typedef struct { int64_t signature; char* pComment; char* pCode; + int8_t orReplace; } SCreateFuncReq; int32_t tSerializeSCreateFuncReq(void* buf, int32_t bufLen, SCreateFuncReq* pReq); @@ -1095,6 +1096,7 @@ typedef struct { typedef struct { int32_t numOfFuncs; SArray* pFuncInfos; + SArray* pFuncVersions; } SRetrieveFuncRsp; int32_t tSerializeSRetrieveFuncRsp(void* buf, int32_t bufLen, SRetrieveFuncRsp* pRsp); diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index ccc3ceae7b600a39fda4bd95544f8be8ef317ca4..b4945da699bed49ccced499bbc0b86d881e6754f 100644 --- a/source/common/src/tmsg.c +++ b/source/common/src/tmsg.c @@ -1702,6 +1702,8 @@ 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); int32_t tlen = encoder.pos; @@ -1744,6 +1746,8 @@ int32_t tDeserializeSCreateFuncReq(void *buf, int32_t bufLen, SCreateFuncReq *pR if (tDecodeCStrTo(&decoder, pReq->pComment) < 0) return -1; } + if (tDecoodeI8(&decoder, &pReq->orReplace) < 0) return -1; + tEndDecode(&decoder); tDecoderClear(&decoder); @@ -1855,6 +1859,12 @@ int32_t tSerializeSRetrieveFuncRsp(void *buf, int32_t bufLen, SRetrieveFuncRsp * } } + if (pRsp->numOfFuncs != (int32_t)taosArrayGetSize(pRsp->pFuncVersions)) return -1; + for (int32_t i = 0; i < pRsp->numOfFuncs; ++i) { + int32_t version = *(int32_t*)taosArrayGet(pRsp->pFuncVersions, i); + if (tEncodeI32(&encoder, version) < 0) return -1; + } + tEndEncode(&encoder); int32_t tlen = encoder.pos; @@ -1902,6 +1912,15 @@ int32_t tDeserializeSRetrieveFuncRsp(void *buf, int32_t bufLen, SRetrieveFuncRsp taosArrayPush(pRsp->pFuncInfos, &fInfo); } + + pRsp->pFuncVersions = taosArrayInit(pRsp->numOfFuncs, sizeof(int32_t)); + if (pRsp->pFuncVersions == NULL) return -1; + + for (int32_t i = 0; i < pRsp->numOfFuncs; ++i) { + int32_t version = 0; + if (tDecodeI32(&decoder, &version) < 0) return -1; + taosArrayPush(pRsp->pFuncVersions, &version); + } tEndDecode(&decoder); tDecoderClear(&decoder); diff --git a/source/libs/function/src/udfd.c b/source/libs/function/src/udfd.c index a8b993290e6367bf49a833f36a6455f3bae67abe..095d9afd429c744c317241c9d136fa796a0b1235 100644 --- a/source/libs/function/src/udfd.c +++ b/source/libs/function/src/udfd.c @@ -246,6 +246,7 @@ 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; int8_t funcType; int8_t scriptType; @@ -833,13 +834,13 @@ void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) { 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; udf->outputType = pFuncInfo->outputType; udf->outputLen = pFuncInfo->outputLen; udf->bufSize = pFuncInfo->bufSize; + udf->version = *(int32_t*)taosArrayGet(retrieveRsp.pFuncVersions,0); if (!osTempSpaceAvailable()) { terrno = TSDB_CODE_NO_AVAIL_DISK; @@ -850,9 +851,9 @@ void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) { char path[PATH_MAX] = {0}; #ifdef WINDOWS - snprintf(path, sizeof(path), "%s%s", tsTempDir, pFuncInfo->name); + snprintf(path, sizeof(path), "%s%s%d", tsTempDir, pFuncInfo->name, udf->version); #else - snprintf(path, sizeof(path), "%s/%s", tsTempDir, pFuncInfo->name); + snprintf(path, sizeof(path), "%s/%s%d", tsTempDir, pFuncInfo->name, udf->version); #endif TdFilePtr file = taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC); if (file == NULL) {