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

fix: the buf returend by the udf has a smaller size than the function bufsize

This commit is contained in:
slzhou 2023-03-06 09:20:55 +08:00
parent 44e3b11580
commit 6c4400c3ba
2 changed files with 16 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -899,9 +899,11 @@ int32_t convertDataBlockToScalarParm(SSDataBlock *input, SScalarParam *output) {
typedef struct SUdfAggRes { typedef struct SUdfAggRes {
int8_t finalResNum; int8_t finalResNum;
int8_t interResNum; int8_t interResNum;
int32_t interResBufLen;
char *finalResBuf; char *finalResBuf;
char *interResBuf; char *interResBuf;
} SUdfAggRes; } SUdfAggRes;
void onUdfcPipeClose(uv_handle_t *handle); void onUdfcPipeClose(uv_handle_t *handle);
int32_t udfcGetUdfTaskResultFromUvTask(SClientUdfTask *task, SClientUvTaskNode *uvTask); int32_t udfcGetUdfTaskResultFromUvTask(SClientUdfTask *task, SClientUvTaskNode *uvTask);
void udfcAllocateBuffer(uv_handle_t *handle, size_t suggestedSize, uv_buf_t *buf); void udfcAllocateBuffer(uv_handle_t *handle, size_t suggestedSize, uv_buf_t *buf);
@ -1096,9 +1098,10 @@ bool udfAggInit(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo *pResult
releaseUdfFuncHandle(pCtx->udfName); releaseUdfFuncHandle(pCtx->udfName);
return false; return false;
} }
udfRes->interResNum = buf.numOfResult;
if (buf.bufLen <= session->bufSize) { if (buf.bufLen <= session->bufSize) {
memcpy(udfRes->interResBuf, buf.buf, buf.bufLen); memcpy(udfRes->interResBuf, buf.buf, buf.bufLen);
udfRes->interResBufLen = buf.bufLen;
udfRes->interResNum = buf.numOfResult;
} 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); releaseUdfFuncHandle(pCtx->udfName);
@ -1136,7 +1139,7 @@ int32_t udfAggProcess(struct SqlFunctionCtx *pCtx) {
SSDataBlock *inputBlock = blockDataExtractBlock(pTempBlock, start, numOfRows); SSDataBlock *inputBlock = blockDataExtractBlock(pTempBlock, start, numOfRows);
SUdfInterBuf state = {.buf = udfRes->interResBuf, .bufLen = session->bufSize, .numOfResult = udfRes->interResNum}; SUdfInterBuf state = {.buf = udfRes->interResBuf, .bufLen = udfRes->interResBufLen, .numOfResult = udfRes->interResNum};
SUdfInterBuf newState = {0}; SUdfInterBuf newState = {0};
udfCode = doCallUdfAggProcess(session, inputBlock, &state, &newState); udfCode = doCallUdfAggProcess(session, inputBlock, &state, &newState);
@ -1144,17 +1147,17 @@ int32_t udfAggProcess(struct SqlFunctionCtx *pCtx) {
fnError("udfAggProcess error. code: %d", udfCode); fnError("udfAggProcess error. code: %d", udfCode);
newState.numOfResult = 0; newState.numOfResult = 0;
} else { } else {
udfRes->interResNum = newState.numOfResult;
if (newState.bufLen <= session->bufSize) { if (newState.bufLen <= session->bufSize) {
memcpy(udfRes->interResBuf, newState.buf, newState.bufLen); memcpy(udfRes->interResBuf, newState.buf, newState.bufLen);
udfRes->interResBufLen = newState.bufLen;
udfRes->interResNum = newState.numOfResult;
} else { } else {
fnError("udfc inter buf size %d is greater than function bufSize %d", newState.bufLen, session->bufSize); fnError("udfc inter buf size %d is greater than function bufSize %d", newState.bufLen, session->bufSize);
udfCode = TSDB_CODE_UDF_INVALID_BUFSIZE; udfCode = TSDB_CODE_UDF_INVALID_BUFSIZE;
} }
} }
if (newState.numOfResult == 1 || state.numOfResult == 1) {
GET_RES_INFO(pCtx)->numOfRes = 1; GET_RES_INFO(pCtx)->numOfRes = udfRes->interResNum;
}
blockDataDestroy(inputBlock); blockDataDestroy(inputBlock);
@ -1180,7 +1183,7 @@ int32_t udfAggFinalize(struct SqlFunctionCtx *pCtx, SSDataBlock *pBlock) {
udfRes->interResBuf = (char *)udfRes + sizeof(SUdfAggRes) + session->outputLen; udfRes->interResBuf = (char *)udfRes + sizeof(SUdfAggRes) + session->outputLen;
SUdfInterBuf resultBuf = {0}; SUdfInterBuf resultBuf = {0};
SUdfInterBuf state = {.buf = udfRes->interResBuf, .bufLen = session->bufSize, .numOfResult = udfRes->interResNum}; SUdfInterBuf state = {.buf = udfRes->interResBuf, .bufLen = udfRes->interResBufLen, .numOfResult = udfRes->interResNum};
int32_t udfCallCode = 0; int32_t udfCallCode = 0;
udfCallCode = doCallUdfAggFinalize(session, &state, &resultBuf); udfCallCode = doCallUdfAggFinalize(session, &state, &resultBuf);
if (udfCallCode != 0) { if (udfCallCode != 0) {

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

@ -173,6 +173,7 @@ int32_t udfdCPluginUdfAggFinish(SUdfInterBuf *buf, SUdfInterBuf *resultData, voi
// for others, dlopen/dlsym to find function pointers // for others, dlopen/dlsym to find function pointers
typedef struct SUdfScriptPlugin { typedef struct SUdfScriptPlugin {
int8_t scriptType; int8_t scriptType;
const char* scriptSuffix;
char libPath[PATH_MAX]; char libPath[PATH_MAX];
bool libLoaded; bool libLoaded;
@ -313,6 +314,7 @@ static void udfdConnectMnodeThreadFunc(void *args);
void udfdInitializeCPlugin(SUdfScriptPlugin *plugin) { void udfdInitializeCPlugin(SUdfScriptPlugin *plugin) {
plugin->scriptType = TSDB_FUNC_SCRIPT_BIN_LIB; plugin->scriptType = TSDB_FUNC_SCRIPT_BIN_LIB;
plugin->scriptSuffix = '.so';
plugin->openFunc = udfdCPluginOpen; plugin->openFunc = udfdCPluginOpen;
plugin->closeFunc = udfdCPluginClose; plugin->closeFunc = udfdCPluginClose;
plugin->udfInitFunc = udfdCPluginUdfInit; plugin->udfInitFunc = udfdCPluginUdfInit;
@ -346,6 +348,7 @@ int32_t udfdLoadSharedLib(char *libPath, uv_lib_t *pLib, const char *funcName[],
void udfdInitializePythonPlugin(SUdfScriptPlugin *plugin) { void udfdInitializePythonPlugin(SUdfScriptPlugin *plugin) {
plugin->scriptType = TSDB_FUNC_SCRIPT_PYTHON; plugin->scriptType = TSDB_FUNC_SCRIPT_PYTHON;
plugin->scriptSuffix = "py";
//todo: windows support //todo: windows support
sprintf(plugin->libPath, "%s", "libtaospyudf.so"); sprintf(plugin->libPath, "%s", "libtaospyudf.so");
plugin->libLoaded = false; plugin->libLoaded = false;
@ -776,7 +779,7 @@ void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
udf->outputType = pFuncInfo->outputType; udf->outputType = pFuncInfo->outputType;
udf->outputLen = pFuncInfo->outputLen; udf->outputLen = pFuncInfo->outputLen;
udf->bufSize = pFuncInfo->bufSize; udf->bufSize = pFuncInfo->bufSize;
char* suffix = global.scriptPlugins[udf->scriptType]->scriptSuffix;
if (!osTempSpaceAvailable()) { if (!osTempSpaceAvailable()) {
terrno = TSDB_CODE_NO_AVAIL_DISK; terrno = TSDB_CODE_NO_AVAIL_DISK;
msgInfo->code = terrno; msgInfo->code = terrno;
@ -786,9 +789,9 @@ void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
char path[PATH_MAX] = {0}; char path[PATH_MAX] = {0};
#ifdef WINDOWS #ifdef WINDOWS
snprintf(path, sizeof(path), "%s%s", tsTempDir, pFuncInfo->name); snprintf(path, sizeof(path), "%s%s.%s", tsTempDir, pFuncInfo->name, suffix);
#else #else
snprintf(path, sizeof(path), "%s/%s", tsTempDir, pFuncInfo->name); snprintf(path, sizeof(path), "%s/%s.%s", tsTempDir, pFuncInfo->name, suffix);
#endif #endif
TdFilePtr file = taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC); TdFilePtr file = taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC);
if (file == NULL) { if (file == NULL) {