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

fix: release udf func handle when udf agg failure

This commit is contained in:
slzhou 2022-05-18 10:29:32 +08:00
parent 6e1d926f36
commit d5c30f5ea5
1 changed files with 6 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1386,7 +1386,7 @@ int32_t cleanUpUdfs() {
if (handle != NULL && ((SUdfcUvSession*)handle)->udfUvPipe != NULL) {
taosArrayPush(udfStubs, stub);
} else {
fnInfo("invalid handle for %s, refCount: %d, last ref time: %"PRId64". remove it from cache",
fnInfo("udf invalid handle for %s, refCount: %d, last ref time: %"PRId64". remove it from cache",
stub->udfName, stub->refCount, stub->lastRefTime);
}
}
@ -1602,6 +1602,7 @@ bool udfAggInit(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResult
SUdfInterBuf buf = {0};
if ((udfCode = doCallUdfAggInit(handle, &buf)) != 0) {
fnError("udfAggInit error. step doCallUdfAggInit. udf code: %d", udfCode);
releaseUdfFuncHandle(pCtx->udfName);
return false;
}
udfRes->interResNum = buf.numOfResult;
@ -1609,6 +1610,7 @@ bool udfAggInit(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResult
memcpy(udfRes->interResBuf, buf.buf, buf.bufLen);
} else {
fnError("udfc inter buf size %d is greater than function bufSize %d", buf.bufLen, session->bufSize);
releaseUdfFuncHandle(pCtx->udfName);
return false;
}
freeUdfInterBuf(&buf);
@ -1674,6 +1676,9 @@ int32_t udfAggProcess(struct SqlFunctionCtx *pCtx) {
blockDataDestroy(inputBlock);
taosArrayDestroy(tempBlock.pDataBlock);
if (udfCode != 0) {
releaseUdfFuncHandle(pCtx->udfName);
}
freeUdfInterBuf(&newState);
return udfCode;
}