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

feature/qnode

This commit is contained in:
dapan1121 2022-04-25 09:32:46 +08:00
parent 8bbc6b09ca
commit 54dd9e3ca4
2 changed files with 33 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
source/dnode/qnode/src/qnode.c
View File

@ -17,6 +17,7 @@
#include "qndInt.h" #include "qndInt.h"
#include "query.h" #include "query.h"
#include "qworker.h" #include "qworker.h"
//#include "tudf.h"
SQnode *qndOpen(const SQnodeOpt *pOption) { SQnode *qndOpen(const SQnodeOpt *pOption) {
SQnode *pQnode = taosMemoryCalloc(1, sizeof(SQnode)); SQnode *pQnode = taosMemoryCalloc(1, sizeof(SQnode));
@ -25,6 +26,8 @@ SQnode *qndOpen(const SQnodeOpt *pOption) {
return NULL; return NULL;
} }
//udfcOpen();
if (qWorkerInit(NODE_TYPE_QNODE, pQnode->qndId, NULL, (void **)&pQnode->pQuery, &pOption->msgCb)) { if (qWorkerInit(NODE_TYPE_QNODE, pQnode->qndId, NULL, (void **)&pQnode->pQuery, &pOption->msgCb)) {
taosMemoryFreeClear(pQnode); taosMemoryFreeClear(pQnode);
return NULL; return NULL;
@ -37,6 +40,8 @@ SQnode *qndOpen(const SQnodeOpt *pOption) {
void qndClose(SQnode *pQnode) { void qndClose(SQnode *pQnode) {
qWorkerDestroy((void **)&pQnode->pQuery); qWorkerDestroy((void **)&pQnode->pQuery);
//udfcClose();
taosMemoryFree(pQnode); taosMemoryFree(pQnode);
} }

26
source/libs/scalar/src/scalar.c
View File

@ -329,18 +329,29 @@ _return:
} }
int32_t sclExecFunction(SFunctionNode *node, SScalarCtx *ctx, SScalarParam *output) { int32_t sclExecFunction(SFunctionNode *node, SScalarCtx *ctx, SScalarParam *output) {
SScalarFuncExecFuncs ffpSet = {0};
int32_t code = fmGetScalarFuncExecFuncs(node->funcId, &ffpSet);
if (code) {
sclError("fmGetFuncExecFuncs failed, funcId:%d, code:%s", node->funcId, tstrerror(code));
SCL_ERR_RET(code);
}
SScalarParam *params = NULL; SScalarParam *params = NULL;
int32_t rowNum = 0; int32_t rowNum = 0;
int32_t paramNum = 0; int32_t paramNum = 0;
int32_t code = 0;
SCL_ERR_RET(sclInitParamList(&params, node->pParameterList, ctx, &paramNum, &rowNum)); SCL_ERR_RET(sclInitParamList(&params, node->pParameterList, ctx, &paramNum, &rowNum));
if (fmIsUserDefinedFunc(node->funcId)) {
#if 0
UdfcFuncHandle udfHandle = NULL;
SCL_ERR_JRET(setupUdf(node->functionName, &udfHandle));
code = callUdfScalarFunc(udfHandle, params, paramNum, output);
teardownUdf(udfHandle);
SCL_ERR_JRET(code);
#endif
} else {
SScalarFuncExecFuncs ffpSet = {0};
code = fmGetScalarFuncExecFuncs(node->funcId, &ffpSet);
if (code) {
sclError("fmGetFuncExecFuncs failed, funcId:%d, code:%s", node->funcId, tstrerror(code));
SCL_ERR_JRET(code);
}
output->columnData = createColumnInfoData(&node->node.resType, rowNum); output->columnData = createColumnInfoData(&node->node.resType, rowNum);
if (output->columnData == NULL) { if (output->columnData == NULL) {
sclError("calloc %d failed", (int32_t)(rowNum * output->columnData->info.bytes)); sclError("calloc %d failed", (int32_t)(rowNum * output->columnData->info.bytes));
@ -352,6 +363,7 @@ int32_t sclExecFunction(SFunctionNode *node, SScalarCtx *ctx, SScalarParam *outp
sclError("scalar function exec failed, funcId:%d, code:%s", node->funcId, tstrerror(code)); sclError("scalar function exec failed, funcId:%d, code:%s", node->funcId, tstrerror(code));
SCL_ERR_JRET(code); SCL_ERR_JRET(code);
} }
}
_return: _return: