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

Merge pull request #12627 from taosdata/feature/udf 

fix: release udf func handle when udf agg failure
This commit is contained in:
shenglian-zhou 2022-05-18 10:44:47 +08:00 committed by GitHub
parent d42961521c d5c30f5ea5
commit ea3e48827f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
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) { if (handle != NULL && ((SUdfcUvSession*)handle)->udfUvPipe != NULL) {
taosArrayPush(udfStubs, stub); taosArrayPush(udfStubs, stub);
} else { } 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); stub->udfName, stub->refCount, stub->lastRefTime);
} }
} }
@ -1602,6 +1602,7 @@ bool udfAggInit(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResult
SUdfInterBuf buf = {0}; SUdfInterBuf buf = {0};
if ((udfCode = doCallUdfAggInit(handle, &buf)) != 0) { if ((udfCode = doCallUdfAggInit(handle, &buf)) != 0) {
fnError("udfAggInit error. step doCallUdfAggInit. udf code: %d", udfCode); fnError("udfAggInit error. step doCallUdfAggInit. udf code: %d", udfCode);
releaseUdfFuncHandle(pCtx->udfName);
return false; return false;
} }
udfRes->interResNum = buf.numOfResult; udfRes->interResNum = buf.numOfResult;
@ -1609,6 +1610,7 @@ bool udfAggInit(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResult
memcpy(udfRes->interResBuf, buf.buf, buf.bufLen); memcpy(udfRes->interResBuf, buf.buf, buf.bufLen);
} else { } else {
fnError("udfc inter buf size %d is greater than function bufSize %d", buf.bufLen, session->bufSize); fnError("udfc inter buf size %d is greater than function bufSize %d", buf.bufLen, session->bufSize);
releaseUdfFuncHandle(pCtx->udfName);
return false; return false;
} }
freeUdfInterBuf(&buf); freeUdfInterBuf(&buf);
@ -1674,6 +1676,9 @@ int32_t udfAggProcess(struct SqlFunctionCtx *pCtx) {
blockDataDestroy(inputBlock); blockDataDestroy(inputBlock);
taosArrayDestroy(tempBlock.pDataBlock); taosArrayDestroy(tempBlock.pDataBlock);
if (udfCode != 0) {
releaseUdfFuncHandle(pCtx->udfName);
}
freeUdfInterBuf(&newState); freeUdfInterBuf(&newState);
return udfCode; return udfCode;
} }