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

udf merge: temporarily block code

This commit is contained in:
factosea 2024-12-19 15:43:24 +08:00
parent 50dbfdd4a3
commit 1cb0e2fec2
3 changed files with 63 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
include/libs/function/tudf.h
View File

@ -109,8 +109,9 @@ int32_t doCallUdfAggProcess(UdfcFuncHandle handle, SSDataBlock *block, SUdfInter
int32_t doCallUdfAggFinalize(UdfcFuncHandle handle, SUdfInterBuf *interBuf, SUdfInterBuf *resultData); int32_t doCallUdfAggFinalize(UdfcFuncHandle handle, SUdfInterBuf *interBuf, SUdfInterBuf *resultData);
// input: interbuf1, interbuf2 // input: interbuf1, interbuf2
// output: resultBuf // output: resultBuf
int32_t doCallUdfAggMerge(UdfcFuncHandle handle, SUdfInterBuf *interBuf1, SUdfInterBuf *interBuf2, // udf todo: aggmerge
SUdfInterBuf *resultBuf); // int32_t doCallUdfAggMerge(UdfcFuncHandle handle, SUdfInterBuf *interBuf1, SUdfInterBuf *interBuf2,
// SUdfInterBuf *resultBuf);
// input: block // input: block
// output: resultData // output: resultData
int32_t doCallUdfScalarFunc(UdfcFuncHandle handle, SScalarParam *input, int32_t numOfCols, SScalarParam *output); int32_t doCallUdfScalarFunc(UdfcFuncHandle handle, SScalarParam *input, int32_t numOfCols, SScalarParam *output);

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

@ -668,8 +668,8 @@ int32_t encodeUdfCallRequest(void **buf, const SUdfCallRequest *call) {
len += tEncodeDataBlock(buf, &call->block); len += tEncodeDataBlock(buf, &call->block);
len += encodeUdfInterBuf(buf, &call->interBuf); len += encodeUdfInterBuf(buf, &call->interBuf);
} else if (call->callType == TSDB_UDF_CALL_AGG_MERGE) { } else if (call->callType == TSDB_UDF_CALL_AGG_MERGE) {
len += encodeUdfInterBuf(buf, &call->interBuf); // len += encodeUdfInterBuf(buf, &call->interBuf);
len += encodeUdfInterBuf(buf, &call->interBuf2); // len += encodeUdfInterBuf(buf, &call->interBuf2);
} else if (call->callType == TSDB_UDF_CALL_AGG_FIN) { } else if (call->callType == TSDB_UDF_CALL_AGG_FIN) {
len += encodeUdfInterBuf(buf, &call->interBuf); len += encodeUdfInterBuf(buf, &call->interBuf);
} }
@ -690,10 +690,10 @@ void *decodeUdfCallRequest(const void *buf, SUdfCallRequest *call) {
buf = tDecodeDataBlock(buf, &call->block); buf = tDecodeDataBlock(buf, &call->block);
buf = decodeUdfInterBuf(buf, &call->interBuf); buf = decodeUdfInterBuf(buf, &call->interBuf);
break; break;
case TSDB_UDF_CALL_AGG_MERGE: // case TSDB_UDF_CALL_AGG_MERGE:
buf = decodeUdfInterBuf(buf, &call->interBuf); // buf = decodeUdfInterBuf(buf, &call->interBuf);
buf = decodeUdfInterBuf(buf, &call->interBuf2); // buf = decodeUdfInterBuf(buf, &call->interBuf2);
break; // break;
case TSDB_UDF_CALL_AGG_FIN: case TSDB_UDF_CALL_AGG_FIN:
buf = decodeUdfInterBuf(buf, &call->interBuf); buf = decodeUdfInterBuf(buf, &call->interBuf);
break; break;
@ -779,9 +779,9 @@ int32_t encodeUdfCallResponse(void **buf, const SUdfCallResponse *callRsp) {
case TSDB_UDF_CALL_AGG_PROC: case TSDB_UDF_CALL_AGG_PROC:
len += encodeUdfInterBuf(buf, &callRsp->resultBuf); len += encodeUdfInterBuf(buf, &callRsp->resultBuf);
break; break;
case TSDB_UDF_CALL_AGG_MERGE: // case TSDB_UDF_CALL_AGG_MERGE:
len += encodeUdfInterBuf(buf, &callRsp->resultBuf); // len += encodeUdfInterBuf(buf, &callRsp->resultBuf);
break; // break;
case TSDB_UDF_CALL_AGG_FIN: case TSDB_UDF_CALL_AGG_FIN:
len += encodeUdfInterBuf(buf, &callRsp->resultBuf); len += encodeUdfInterBuf(buf, &callRsp->resultBuf);
break; break;
@ -801,9 +801,9 @@ void *decodeUdfCallResponse(const void *buf, SUdfCallResponse *callRsp) {
case TSDB_UDF_CALL_AGG_PROC: case TSDB_UDF_CALL_AGG_PROC:
buf = decodeUdfInterBuf(buf, &callRsp->resultBuf); buf = decodeUdfInterBuf(buf, &callRsp->resultBuf);
break; break;
case TSDB_UDF_CALL_AGG_MERGE: // case TSDB_UDF_CALL_AGG_MERGE:
buf = decodeUdfInterBuf(buf, &callRsp->resultBuf); // buf = decodeUdfInterBuf(buf, &callRsp->resultBuf);
break; // break;
case TSDB_UDF_CALL_AGG_FIN: case TSDB_UDF_CALL_AGG_FIN:
buf = decodeUdfInterBuf(buf, &callRsp->resultBuf); buf = decodeUdfInterBuf(buf, &callRsp->resultBuf);
break; break;
@ -1129,8 +1129,9 @@ int32_t callUdf(UdfcFuncHandle handle, int8_t callType, SSDataBlock *input, SUdf
SSDataBlock *output, SUdfInterBuf *newState); SSDataBlock *output, SUdfInterBuf *newState);
int32_t doCallUdfAggInit(UdfcFuncHandle handle, SUdfInterBuf *interBuf); int32_t doCallUdfAggInit(UdfcFuncHandle handle, SUdfInterBuf *interBuf);
int32_t doCallUdfAggProcess(UdfcFuncHandle handle, SSDataBlock *block, SUdfInterBuf *state, SUdfInterBuf *newState); int32_t doCallUdfAggProcess(UdfcFuncHandle handle, SSDataBlock *block, SUdfInterBuf *state, SUdfInterBuf *newState);
int32_t doCallUdfAggMerge(UdfcFuncHandle handle, SUdfInterBuf *interBuf1, SUdfInterBuf *interBuf2, // udf todo: aggmerge
SUdfInterBuf *resultBuf); // int32_t doCallUdfAggMerge(UdfcFuncHandle handle, SUdfInterBuf *interBuf1, SUdfInterBuf *interBuf2,
// SUdfInterBuf *resultBuf);
int32_t doCallUdfAggFinalize(UdfcFuncHandle handle, SUdfInterBuf *interBuf, SUdfInterBuf *resultData); int32_t doCallUdfAggFinalize(UdfcFuncHandle handle, SUdfInterBuf *interBuf, SUdfInterBuf *resultData);
int32_t doCallUdfScalarFunc(UdfcFuncHandle handle, SScalarParam *input, int32_t numOfCols, SScalarParam *output); int32_t doCallUdfScalarFunc(UdfcFuncHandle handle, SScalarParam *input, int32_t numOfCols, SScalarParam *output);
int32_t callUdfScalarFunc(char *udfName, SScalarParam *input, int32_t numOfCols, SScalarParam *output); int32_t callUdfScalarFunc(char *udfName, SScalarParam *input, int32_t numOfCols, SScalarParam *output);
@ -2176,11 +2177,11 @@ int32_t callUdf(UdfcFuncHandle handle, int8_t callType, SSDataBlock *input, SUdf
req->interBuf = *state; req->interBuf = *state;
break; break;
} }
case TSDB_UDF_CALL_AGG_MERGE: { // case TSDB_UDF_CALL_AGG_MERGE: {
req->interBuf = *state; // req->interBuf = *state;
req->interBuf2 = *state2; // req->interBuf2 = *state2;
break; // break;
} // }
case TSDB_UDF_CALL_AGG_FIN: { case TSDB_UDF_CALL_AGG_FIN: {
req->interBuf = *state; req->interBuf = *state;
break; break;
@ -2205,10 +2206,10 @@ int32_t callUdf(UdfcFuncHandle handle, int8_t callType, SSDataBlock *input, SUdf
*newState = rsp->resultBuf; *newState = rsp->resultBuf;
break; break;
} }
case TSDB_UDF_CALL_AGG_MERGE: { // case TSDB_UDF_CALL_AGG_MERGE: {
*newState = rsp->resultBuf; // *newState = rsp->resultBuf;
break; // break;
} // }
case TSDB_UDF_CALL_AGG_FIN: { case TSDB_UDF_CALL_AGG_FIN: {
*newState = rsp->resultBuf; *newState = rsp->resultBuf;
break; break;
@ -2241,12 +2242,13 @@ int32_t doCallUdfAggProcess(UdfcFuncHandle handle, SSDataBlock *block, SUdfInter
// input: interbuf1, interbuf2 // input: interbuf1, interbuf2
// output: resultBuf // output: resultBuf
int32_t doCallUdfAggMerge(UdfcFuncHandle handle, SUdfInterBuf *interBuf1, SUdfInterBuf *interBuf2, // udf todo: aggmerge
SUdfInterBuf *resultBuf) { // int32_t doCallUdfAggMerge(UdfcFuncHandle handle, SUdfInterBuf *interBuf1, SUdfInterBuf *interBuf2,
int8_t callType = TSDB_UDF_CALL_AGG_MERGE; // SUdfInterBuf *resultBuf) {
int32_t err = callUdf(handle, callType, NULL, interBuf1, interBuf2, NULL, resultBuf); // int8_t callType = TSDB_UDF_CALL_AGG_MERGE;
return err; // int32_t err = callUdf(handle, callType, NULL, interBuf1, interBuf2, NULL, resultBuf);
} // return err;
// }
// input: interBuf // input: interBuf
// output: resultData // output: resultData

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

@ -194,17 +194,17 @@ int32_t udfdCPluginUdfAggProc(SUdfDataBlock *block, SUdfInterBuf *interBuf, SUdf
} }
} }
int32_t udfdCPluginUdfAggMerge(SUdfInterBuf *inputBuf1, SUdfInterBuf *inputBuf2, SUdfInterBuf *outputBuf, // int32_t udfdCPluginUdfAggMerge(SUdfInterBuf *inputBuf1, SUdfInterBuf *inputBuf2, SUdfInterBuf *outputBuf,
void *udfCtx) { // void *udfCtx) {
TAOS_UDF_CHECK_PTR_RCODE(inputBuf1, inputBuf2, outputBuf, udfCtx); // TAOS_UDF_CHECK_PTR_RCODE(inputBuf1, inputBuf2, outputBuf, udfCtx);
SUdfCPluginCtx *ctx = udfCtx; // SUdfCPluginCtx *ctx = udfCtx;
if (ctx->aggMergeFunc) { // if (ctx->aggMergeFunc) {
return ctx->aggMergeFunc(inputBuf1, inputBuf2, outputBuf); // return ctx->aggMergeFunc(inputBuf1, inputBuf2, outputBuf);
} else { // } else {
fnError("udfd c plugin aggregation merge not implemented"); // fnError("udfd c plugin aggregation merge not implemented");
return TSDB_CODE_UDF_FUNC_EXEC_FAILURE; // return TSDB_CODE_UDF_FUNC_EXEC_FAILURE;
} // }
} // }
int32_t udfdCPluginUdfAggFinish(SUdfInterBuf *buf, SUdfInterBuf *resultData, void *udfCtx) { int32_t udfdCPluginUdfAggFinish(SUdfInterBuf *buf, SUdfInterBuf *resultData, void *udfCtx) {
TAOS_UDF_CHECK_PTR_RCODE(buf, resultData, udfCtx); TAOS_UDF_CHECK_PTR_RCODE(buf, resultData, udfCtx);
@ -378,7 +378,7 @@ int32_t udfdInitializeCPlugin(SUdfScriptPlugin *plugin) {
plugin->udfScalarProcFunc = udfdCPluginUdfScalarProc; plugin->udfScalarProcFunc = udfdCPluginUdfScalarProc;
plugin->udfAggStartFunc = udfdCPluginUdfAggStart; plugin->udfAggStartFunc = udfdCPluginUdfAggStart;
plugin->udfAggProcFunc = udfdCPluginUdfAggProc; plugin->udfAggProcFunc = udfdCPluginUdfAggProc;
plugin->udfAggMergeFunc = udfdCPluginUdfAggMerge; // plugin->udfAggMergeFunc = udfdCPluginUdfAggMerge;
plugin->udfAggFinishFunc = udfdCPluginUdfAggFinish; plugin->udfAggFinishFunc = udfdCPluginUdfAggFinish;
SScriptUdfEnvItem items[1] = {{"LD_LIBRARY_PATH", tsUdfdLdLibPath}}; SScriptUdfEnvItem items[1] = {{"LD_LIBRARY_PATH", tsUdfdLdLibPath}};
@ -889,19 +889,19 @@ void udfdProcessCallRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
break; break;
} }
case TSDB_UDF_CALL_AGG_MERGE: { // case TSDB_UDF_CALL_AGG_MERGE: {
SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize), .bufLen = udf->bufSize, .numOfResult = 0}; // SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize), .bufLen = udf->bufSize, .numOfResult = 0};
if (outBuf.buf != NULL) { // if (outBuf.buf != NULL) {
code = udf->scriptPlugin->udfAggMergeFunc(&call->interBuf, &call->interBuf2, &outBuf, udf->scriptUdfCtx); // code = udf->scriptPlugin->udfAggMergeFunc(&call->interBuf, &call->interBuf2, &outBuf, udf->scriptUdfCtx);
freeUdfInterBuf(&call->interBuf); // freeUdfInterBuf(&call->interBuf);
freeUdfInterBuf(&call->interBuf2); // freeUdfInterBuf(&call->interBuf2);
subRsp->resultBuf = outBuf; // subRsp->resultBuf = outBuf;
} else { // } else {
code = terrno; // code = terrno;
} // }
//
break; // break;
} // }
case TSDB_UDF_CALL_AGG_FIN: { case TSDB_UDF_CALL_AGG_FIN: {
SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize), .bufLen = udf->bufSize, .numOfResult = 0}; SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize), .bufLen = udf->bufSize, .numOfResult = 0};
if (outBuf.buf != NULL) { if (outBuf.buf != NULL) {
@ -959,10 +959,10 @@ _exit:
freeUdfInterBuf(&subRsp->resultBuf); freeUdfInterBuf(&subRsp->resultBuf);
break; break;
} }
case TSDB_UDF_CALL_AGG_MERGE: { // case TSDB_UDF_CALL_AGG_MERGE: {
freeUdfInterBuf(&subRsp->resultBuf); // freeUdfInterBuf(&subRsp->resultBuf);
break; // break;
} // }
case TSDB_UDF_CALL_AGG_FIN: { case TSDB_UDF_CALL_AGG_FIN: {
freeUdfInterBuf(&subRsp->resultBuf); freeUdfInterBuf(&subRsp->resultBuf);
break; break;