Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
f8dd3a13
T
TDengine
项目概览
taosdata
/
TDengine
1 年多 前同步成功
通知
1185
Star
22016
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
提交
f8dd3a13
编写于
5月 07, 2022
作者:
S
slzhou
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
feature(udf):refactor code and fix bugs
上级
f54cdb02
变更
5
隐藏空白更改
内联
并排
Showing
5 changed file
with
145 addition
and
70 deletion
+145
-70
include/libs/function/tudf.h
include/libs/function/tudf.h
+51
-7
source/libs/function/src/tudf.c
source/libs/function/src/tudf.c
+64
-50
source/libs/function/test/udf1.c
source/libs/function/test/udf1.c
+11
-4
source/libs/function/test/udf2.c
source/libs/function/test/udf2.c
+18
-8
tests/script/tsim/query/udf.sim
tests/script/tsim/query/udf.sim
+1
-1
未找到文件。
include/libs/function/tudf.h
浏览文件 @
f8dd3a13
...
...
@@ -44,7 +44,8 @@ enum {
UDFC_CODE_PIPE_READ_ERR
=
-
2
,
UDFC_CODE_CONNECT_PIPE_ERR
=
-
3
,
UDFC_CODE_LOAD_UDF_FAILURE
=
-
4
,
UDFC_CODE_INVALID_STATE
=
-
5
UDFC_CODE_INVALID_STATE
=
-
5
,
UDFC_CODE_NO_PIPE
=
-
6
,
};
typedef
void
*
UdfcFuncHandle
;
...
...
@@ -140,6 +141,44 @@ typedef int32_t (*TUdfDestroyFunc)();
#define UDF_MEMORY_EXP_GROWTH 1.5
#define udfColDataIsNull_var(pColumn, row) ((pColumn->colData.varLenCol.varOffsets)[row] == -1)
#define udfColDataIsNull_f(pColumn, row) ((BMCharPos(pColumn->colData.fixLenCol.nullBitmap, row) & (1u << (7u - BitPos(row)))) == (1u << (7u - BitPos(row))))
#define udfColDataSetNull_f(pColumn, row) \
do { \
BMCharPos(pColumn->colData.fixLenCol.nullBitmap, row) |= (1u << (7u - BitPos(row))); \
} while (0)
#define udfColDataSetNotNull_f(pColumn, r_) \
do { \
BMCharPos(pColumn->colData.fixLenCol.nullBitmap, r_) &= ~(1u << (7u - BitPos(r_))); \
} while (0)
#define udfColDataSetNull_var(pColumn, row) ((pColumn->colData.varLenCol.varOffsets)[row] = -1)
static
FORCE_INLINE
char
*
udfColDataGetData
(
const
SUdfColumn
*
pColumn
,
int32_t
row
)
{
if
(
IS_VAR_DATA_TYPE
(
pColumn
->
colMeta
.
type
))
{
return
pColumn
->
colData
.
varLenCol
.
payload
+
pColumn
->
colData
.
varLenCol
.
varOffsets
[
row
];
}
else
{
return
pColumn
->
colData
.
fixLenCol
.
data
+
pColumn
->
colMeta
.
bytes
*
row
;
}
}
static
FORCE_INLINE
bool
udfColDataIsNull
(
const
SUdfColumn
*
pColumn
,
int32_t
row
)
{
if
(
IS_VAR_DATA_TYPE
(
pColumn
->
colMeta
.
type
))
{
if
(
pColumn
->
colMeta
.
type
==
TSDB_DATA_TYPE_JSON
)
{
if
(
udfColDataIsNull_var
(
pColumn
,
row
))
{
return
true
;
}
char
*
data
=
udfColDataGetData
(
pColumn
,
row
);
return
(
*
data
==
TSDB_DATA_TYPE_NULL
);
}
else
{
return
udfColDataIsNull_var
(
pColumn
,
row
);
}
}
else
{
return
udfColDataIsNull_f
(
pColumn
,
row
);
}
}
static
FORCE_INLINE
int32_t
udfColEnsureCapacity
(
SUdfColumn
*
pColumn
,
int32_t
newCapacity
)
{
SUdfColumnMeta
*
meta
=
&
pColumn
->
colMeta
;
SUdfColumnData
*
data
=
&
pColumn
->
colData
;
...
...
@@ -186,17 +225,22 @@ static FORCE_INLINE int32_t udfColEnsureCapacity(SUdfColumn* pColumn, int32_t ne
return
TSDB_CODE_SUCCESS
;
}
static
FORCE_INLINE
int32_t
udfColSetRow
(
SUdfColumn
*
pColumn
,
uint32_t
currentRow
,
const
char
*
pData
,
bool
isNull
)
{
static
FORCE_INLINE
void
udfColDataSetNull
(
SUdfColumn
*
pColumn
,
int32_t
row
)
{
udfColEnsureCapacity
(
pColumn
,
row
+
1
);
if
(
IS_VAR_DATA_TYPE
(
pColumn
->
colMeta
.
type
))
{
udfColDataSetNull_var
(
pColumn
,
row
);
}
else
{
udfColDataSetNull_f
(
pColumn
,
row
);
}
}
static
FORCE_INLINE
int32_t
udfColDataSet
(
SUdfColumn
*
pColumn
,
uint32_t
currentRow
,
const
char
*
pData
,
bool
isNull
)
{
SUdfColumnMeta
*
meta
=
&
pColumn
->
colMeta
;
SUdfColumnData
*
data
=
&
pColumn
->
colData
;
udfColEnsureCapacity
(
pColumn
,
currentRow
+
1
);
bool
isVarCol
=
IS_VAR_DATA_TYPE
(
meta
->
type
);
if
(
isNull
)
{
if
(
isVarCol
)
{
data
->
varLenCol
.
varOffsets
[
currentRow
]
=
-
1
;
}
else
{
colDataSetNull_f
(
data
->
fixLenCol
.
nullBitmap
,
currentRow
);
}
udfColDataSetNull
(
pColumn
,
currentRow
);
}
else
{
if
(
!
isVarCol
)
{
colDataSetNotNull_f
(
data
->
fixLenCol
.
nullBitmap
,
currentRow
);
...
...
source/libs/function/src/tudf.c
浏览文件 @
f8dd3a13
...
...
@@ -146,15 +146,15 @@ typedef struct SUdfdProxy {
SUdfdProxy
gUdfdProxy
=
{
0
};
typedef
struct
SUdfUvSession
{
typedef
struct
S
Client
UdfUvSession
{
SUdfdProxy
*
udfc
;
int64_t
severHandle
;
uv_pipe_t
*
udf
Svc
Pipe
;
uv_pipe_t
*
udf
Uv
Pipe
;
int8_t
outputType
;
int32_t
outputLen
;
int32_t
bufSize
;
}
SUdfUvSession
;
}
S
Client
UdfUvSession
;
typedef
struct
SClientUvTaskNode
{
SUdfdProxy
*
udfc
;
...
...
@@ -177,7 +177,7 @@ typedef struct SClientUvTaskNode {
typedef
struct
SClientUdfTask
{
int8_t
type
;
SUdfUvSession
*
session
;
S
Client
UdfUvSession
*
session
;
int32_t
errCode
;
...
...
@@ -209,6 +209,7 @@ typedef struct SClientUvConn {
uv_pipe_t
*
pipe
;
QUEUE
taskQueue
;
SClientConnBuf
readBuf
;
SClientUdfUvSession
*
session
;
}
SClientUvConn
;
enum
{
...
...
@@ -617,18 +618,17 @@ void onUdfcPipeClose(uv_handle_t *handle) {
QUEUE
*
h
=
QUEUE_HEAD
(
&
conn
->
taskQueue
);
SClientUvTaskNode
*
task
=
QUEUE_DATA
(
h
,
SClientUvTaskNode
,
connTaskQueue
);
task
->
errCode
=
0
;
uv_sem_post
(
&
task
->
taskSem
);
QUEUE_REMOVE
(
&
task
->
procTaskQueue
);
uv_sem_post
(
&
task
->
taskSem
);
}
conn
->
session
->
udfUvPipe
=
NULL
;
taosMemoryFree
(
conn
->
readBuf
.
buf
);
taosMemoryFree
(
conn
);
taosMemoryFree
((
uv_pipe_t
*
)
handle
);
}
int32_t
udfcGetU
vTaskResponseResult
(
SClientUdfTask
*
task
,
SClientUvTaskNode
*
uvTask
)
{
fnDebug
(
"udfc get uv task result. task: %p
"
,
t
ask
);
int32_t
udfcGetU
dfTaskResultFromUvTask
(
SClientUdfTask
*
task
,
SClientUvTaskNode
*
uvTask
)
{
fnDebug
(
"udfc get uv task result. task: %p
, uvTask: %p"
,
task
,
uvT
ask
);
if
(
uvTask
->
type
==
UV_TASK_REQ_RSP
)
{
if
(
uvTask
->
rspBuf
.
base
!=
NULL
)
{
SUdfResponse
rsp
;
...
...
@@ -748,8 +748,8 @@ void udfcUvHandleRsp(SClientUvConn *conn) {
if
(
taskFound
)
{
taskFound
->
rspBuf
=
uv_buf_init
(
connBuf
->
buf
,
connBuf
->
len
);
QUEUE_REMOVE
(
&
taskFound
->
connTaskQueue
);
uv_sem_post
(
&
taskFound
->
taskSem
);
QUEUE_REMOVE
(
&
taskFound
->
procTaskQueue
);
uv_sem_post
(
&
taskFound
->
taskSem
);
}
else
{
fnError
(
"no task is waiting for the response."
);
}
...
...
@@ -764,14 +764,12 @@ void udfcUvHandleError(SClientUvConn *conn) {
QUEUE
*
h
=
QUEUE_HEAD
(
&
conn
->
taskQueue
);
SClientUvTaskNode
*
task
=
QUEUE_DATA
(
h
,
SClientUvTaskNode
,
connTaskQueue
);
task
->
errCode
=
UDFC_CODE_PIPE_READ_ERR
;
uv_sem_post
(
&
task
->
taskSem
);
QUEUE_REMOVE
(
&
task
->
connTaskQueue
);
QUEUE_REMOVE
(
&
task
->
procTaskQueue
);
uv_sem_post
(
&
task
->
taskSem
);
}
uv_close
((
uv_handle_t
*
)
conn
->
pipe
,
NULL
);
taosMemoryFree
(
conn
->
pipe
);
taosMemoryFree
(
conn
->
readBuf
.
buf
);
taosMemoryFree
(
conn
);
uv_close
((
uv_handle_t
*
)
conn
->
pipe
,
onUdfcPipeClose
);
}
void
onUdfcRead
(
uv_stream_t
*
client
,
ssize_t
nread
,
const
uv_buf_t
*
buf
)
{
...
...
@@ -788,9 +786,9 @@ void onUdfcRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf) {
}
if
(
nread
<
0
)
{
fnError
(
"udfc client pipe %p read error: %
s"
,
client
,
uv_strerror
(
nread
));
fnError
(
"udfc client pipe %p read error: %
zd, %s."
,
client
,
nread
,
uv_strerror
(
nread
));
if
(
nread
==
UV_EOF
)
{
fnError
(
"udfc client pipe %p closed"
,
client
);
fnError
(
"
\t
udfc client pipe %p closed"
,
client
);
}
udfcUvHandleError
(
conn
);
}
...
...
@@ -823,14 +821,14 @@ void onUdfClientConnect(uv_connect_t *connect, int status) {
QUEUE_REMOVE
(
&
uvTask
->
procTaskQueue
);
}
int32_t
createUdfc
UvTask
(
SClientUdfTask
*
task
,
int8_t
uvTaskType
,
SClientUvTaskNode
**
pUvTask
)
{
int32_t
udfcCreate
UvTask
(
SClientUdfTask
*
task
,
int8_t
uvTaskType
,
SClientUvTaskNode
**
pUvTask
)
{
SClientUvTaskNode
*
uvTask
=
taosMemoryCalloc
(
1
,
sizeof
(
SClientUvTaskNode
));
uvTask
->
type
=
uvTaskType
;
uvTask
->
udfc
=
task
->
session
->
udfc
;
if
(
uvTaskType
==
UV_TASK_CONNECT
)
{
}
else
if
(
uvTaskType
==
UV_TASK_REQ_RSP
)
{
uvTask
->
pipe
=
task
->
session
->
udf
Svc
Pipe
;
uvTask
->
pipe
=
task
->
session
->
udf
Uv
Pipe
;
SUdfRequest
request
;
request
.
type
=
task
->
type
;
request
.
seqNum
=
atomic_fetch_add_64
(
&
gUdfTaskSeqNum
,
1
);
...
...
@@ -855,7 +853,7 @@ int32_t createUdfcUvTask(SClientUdfTask *task, int8_t uvTaskType, SClientUvTaskN
uvTask
->
reqBuf
=
uv_buf_init
(
bufBegin
,
bufLen
);
uvTask
->
seqNum
=
request
.
seqNum
;
}
else
if
(
uvTaskType
==
UV_TASK_DISCONNECT
)
{
uvTask
->
pipe
=
task
->
session
->
udf
Svc
Pipe
;
uvTask
->
pipe
=
task
->
session
->
udf
Uv
Pipe
;
}
uv_sem_init
(
&
uvTask
->
taskSem
,
0
);
...
...
@@ -863,7 +861,7 @@ int32_t createUdfcUvTask(SClientUdfTask *task, int8_t uvTaskType, SClientUvTaskN
return
0
;
}
int32_t
queueUvUdf
Task
(
SClientUvTaskNode
*
uvTask
)
{
int32_t
udfcQueueUv
Task
(
SClientUvTaskNode
*
uvTask
)
{
fnTrace
(
"queue uv task to event loop, task: %d, %p"
,
uvTask
->
type
,
uvTask
);
SUdfdProxy
*
udfc
=
uvTask
->
udfc
;
uv_mutex_lock
(
&
udfc
->
gUdfTaskQueueMutex
);
...
...
@@ -872,12 +870,13 @@ int32_t queueUvUdfTask(SClientUvTaskNode *uvTask) {
uv_async_send
(
&
udfc
->
gUdfLoopTaskAync
);
uv_sem_wait
(
&
uvTask
->
taskSem
);
fnInfo
(
"udfc uv task finished. task: %d, %p"
,
uvTask
->
type
,
uvTask
);
uv_sem_destroy
(
&
uvTask
->
taskSem
);
return
0
;
}
int32_t
startUvUdf
Task
(
SClientUvTaskNode
*
uvTask
)
{
int32_t
udfcStartUv
Task
(
SClientUvTaskNode
*
uvTask
)
{
fnTrace
(
"event loop start uv task. task: %d, %p"
,
uvTask
->
type
,
uvTask
);
switch
(
uvTask
->
type
)
{
case
UV_TASK_CONNECT
:
{
...
...
@@ -885,7 +884,7 @@ int32_t startUvUdfTask(SClientUvTaskNode *uvTask) {
uv_pipe_init
(
&
uvTask
->
udfc
->
gUdfdLoop
,
pipe
,
0
);
uvTask
->
pipe
=
pipe
;
SClientUvConn
*
conn
=
taosMemory
Malloc
(
sizeof
(
SClientUvConn
));
SClientUvConn
*
conn
=
taosMemory
Calloc
(
1
,
sizeof
(
SClientUvConn
));
conn
->
pipe
=
pipe
;
conn
->
readBuf
.
len
=
0
;
conn
->
readBuf
.
cap
=
0
;
...
...
@@ -933,13 +932,14 @@ void udfClientAsyncCb(uv_async_t *async) {
QUEUE
*
h
=
QUEUE_HEAD
(
&
wq
);
QUEUE_REMOVE
(
h
);
SClientUvTaskNode
*
task
=
QUEUE_DATA
(
h
,
SClientUvTaskNode
,
recvTaskQueue
);
startUvUdf
Task
(
task
);
udfcStartUv
Task
(
task
);
QUEUE_INSERT_TAIL
(
&
udfc
->
gUvProcTaskQueue
,
&
task
->
procTaskQueue
);
}
}
void
cleanUpUvTasks
(
SUdfdProxy
*
udfc
)
{
fnDebug
(
"clean up uv tasks"
)
QUEUE
wq
;
uv_mutex_lock
(
&
udfc
->
gUdfTaskQueueMutex
);
...
...
@@ -956,7 +956,6 @@ void cleanUpUvTasks(SUdfdProxy *udfc) {
uv_sem_post
(
&
task
->
taskSem
);
}
// TODO: deal with tasks that are waiting result.
while
(
!
QUEUE_EMPTY
(
&
udfc
->
gUvProcTaskQueue
))
{
QUEUE
*
h
=
QUEUE_HEAD
(
&
udfc
->
gUvProcTaskQueue
);
QUEUE_REMOVE
(
h
);
...
...
@@ -1027,14 +1026,16 @@ int32_t udfcClose() {
return
0
;
}
int32_t
udfcRunUvTask
(
SClientUdfTask
*
task
,
int8_t
uvTaskType
)
{
int32_t
udfcRunU
dfU
vTask
(
SClientUdfTask
*
task
,
int8_t
uvTaskType
)
{
SClientUvTaskNode
*
uvTask
=
NULL
;
createUdfc
UvTask
(
task
,
uvTaskType
,
&
uvTask
);
queueUvUdf
Task
(
uvTask
);
udfcGetU
vTaskResponseResult
(
task
,
uvTask
);
udfcCreate
UvTask
(
task
,
uvTaskType
,
&
uvTask
);
udfcQueueUv
Task
(
uvTask
);
udfcGetU
dfTaskResultFromUvTask
(
task
,
uvTask
);
if
(
uvTaskType
==
UV_TASK_CONNECT
)
{
task
->
session
->
udfSvcPipe
=
uvTask
->
pipe
;
task
->
session
->
udfUvPipe
=
uvTask
->
pipe
;
SClientUvConn
*
conn
=
uvTask
->
pipe
->
data
;
conn
->
session
=
task
->
session
;
}
taosMemoryFree
(
uvTask
);
uvTask
=
NULL
;
...
...
@@ -1046,22 +1047,22 @@ int32_t setupUdf(char udfName[], UdfcFuncHandle *funcHandle) {
if
(
gUdfdProxy
.
gUdfcState
!=
UDFC_STATE_READY
)
{
return
UDFC_CODE_INVALID_STATE
;
}
SClientUdfTask
*
task
=
taosMemory
Malloc
(
sizeof
(
SClientUdfTask
));
SClientUdfTask
*
task
=
taosMemory
Calloc
(
1
,
sizeof
(
SClientUdfTask
));
task
->
errCode
=
0
;
task
->
session
=
taosMemory
Malloc
(
sizeof
(
S
UdfUvSession
));
task
->
session
=
taosMemory
Calloc
(
1
,
sizeof
(
SClient
UdfUvSession
));
task
->
session
->
udfc
=
&
gUdfdProxy
;
task
->
type
=
UDF_TASK_SETUP
;
SUdfSetupRequest
*
req
=
&
task
->
_setup
.
req
;
memcpy
(
req
->
udfName
,
udfName
,
TSDB_FUNC_NAME_LEN
);
int32_t
errCode
=
udfcRunUvTask
(
task
,
UV_TASK_CONNECT
);
int32_t
errCode
=
udfcRunU
dfU
vTask
(
task
,
UV_TASK_CONNECT
);
if
(
errCode
!=
0
)
{
fnError
(
"failed to connect to pipe. udfName: %s, pipe: %s"
,
udfName
,
(
&
gUdfdProxy
)
->
udfdPipeName
);
return
UDFC_CODE_CONNECT_PIPE_ERR
;
}
udfcRunUvTask
(
task
,
UV_TASK_REQ_RSP
);
udfcRunU
dfU
vTask
(
task
,
UV_TASK_REQ_RSP
);
SUdfSetupResponse
*
rsp
=
&
task
->
_setup
.
rsp
;
task
->
session
->
severHandle
=
rsp
->
udfHandle
;
...
...
@@ -1082,10 +1083,14 @@ int32_t setupUdf(char udfName[], UdfcFuncHandle *funcHandle) {
int32_t
callUdf
(
UdfcFuncHandle
handle
,
int8_t
callType
,
SSDataBlock
*
input
,
SUdfInterBuf
*
state
,
SUdfInterBuf
*
state2
,
SSDataBlock
*
output
,
SUdfInterBuf
*
newState
)
{
fnTrace
(
"udfc call udf. callType: %d, funcHandle: %p"
,
callType
,
handle
);
SClientUdfTask
*
task
=
taosMemoryMalloc
(
sizeof
(
SClientUdfTask
));
SClientUdfUvSession
*
session
=
(
SClientUdfUvSession
*
)
handle
;
if
(
session
->
udfUvPipe
==
NULL
)
{
fnError
(
"No pipe to udfd"
);
return
UDFC_CODE_NO_PIPE
;
}
SClientUdfTask
*
task
=
taosMemoryCalloc
(
1
,
sizeof
(
SClientUdfTask
));
task
->
errCode
=
0
;
task
->
session
=
(
SUdfUvSession
*
)
handle
;
task
->
session
=
(
S
Client
UdfUvSession
*
)
handle
;
task
->
type
=
UDF_TASK_CALL
;
SUdfCallRequest
*
req
=
&
task
->
_call
.
req
;
...
...
@@ -1117,7 +1122,7 @@ int32_t callUdf(UdfcFuncHandle handle, int8_t callType, SSDataBlock *input, SUdf
}
}
udfcRunUvTask
(
task
,
UV_TASK_REQ_RSP
);
udfcRunU
dfU
vTask
(
task
,
UV_TASK_REQ_RSP
);
if
(
task
->
errCode
!=
0
)
{
fnError
(
"call udf failure. err: %d"
,
task
->
errCode
);
...
...
@@ -1145,9 +1150,10 @@ int32_t callUdf(UdfcFuncHandle handle, int8_t callType, SSDataBlock *input, SUdf
break
;
}
}
}
};
int
err
=
task
->
errCode
;
taosMemoryFree
(
task
);
return
task
->
errCode
;
return
err
;
}
int32_t
callUdfAggInit
(
UdfcFuncHandle
handle
,
SUdfInterBuf
*
interBuf
)
{
...
...
@@ -1188,28 +1194,36 @@ int32_t callUdfScalarFunc(UdfcFuncHandle handle, SScalarParam *input, int32_t nu
convertScalarParamToDataBlock
(
input
,
numOfCols
,
&
inputBlock
);
SSDataBlock
resultBlock
=
{
0
};
int32_t
err
=
callUdf
(
handle
,
callType
,
&
inputBlock
,
NULL
,
NULL
,
&
resultBlock
,
NULL
);
convertDataBlockToScalarParm
(
&
resultBlock
,
output
);
if
(
err
==
0
)
{
convertDataBlockToScalarParm
(
&
resultBlock
,
output
);
}
return
err
;
}
int32_t
teardownUdf
(
UdfcFuncHandle
handle
)
{
fnInfo
(
"tear down udf. udf func handle: %p"
,
handle
);
SClientUdfTask
*
task
=
taosMemoryMalloc
(
sizeof
(
SClientUdfTask
));
SClientUdfUvSession
*
session
=
(
SClientUdfUvSession
*
)
handle
;
if
(
session
->
udfUvPipe
==
NULL
)
{
fnError
(
"pipe to udfd does not exist"
);
return
UDFC_CODE_NO_PIPE
;
}
SClientUdfTask
*
task
=
taosMemoryCalloc
(
1
,
sizeof
(
SClientUdfTask
));
task
->
errCode
=
0
;
task
->
session
=
(
SUdfUvSession
*
)
handle
;
task
->
session
=
session
;
task
->
type
=
UDF_TASK_TEARDOWN
;
SUdfTeardownRequest
*
req
=
&
task
->
_teardown
.
req
;
req
->
udfHandle
=
task
->
session
->
severHandle
;
udfcRunUvTask
(
task
,
UV_TASK_REQ_RSP
);
udfcRunU
dfU
vTask
(
task
,
UV_TASK_REQ_RSP
);
SUdfTeardownResponse
*
rsp
=
&
task
->
_teardown
.
rsp
;
int32_t
err
=
task
->
errCode
;
udfcRunUvTask
(
task
,
UV_TASK_DISCONNECT
);
udfcRunU
dfU
vTask
(
task
,
UV_TASK_DISCONNECT
);
taosMemoryFree
(
task
->
session
);
taosMemoryFree
(
task
);
...
...
@@ -1219,7 +1233,7 @@ int32_t teardownUdf(UdfcFuncHandle handle) {
//memory layout |---SUdfAggRes----|-----final result-----|---inter result----|
typedef
struct
SUdfAggRes
{
SUdfUvSession
*
session
;
S
Client
UdfUvSession
*
session
;
int8_t
finalResNum
;
int8_t
interResNum
;
char
*
finalResBuf
;
...
...
@@ -1242,7 +1256,7 @@ bool udfAggInit(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResult
if
(
setupUdf
((
char
*
)
pCtx
->
udfName
,
&
handle
)
!=
0
)
{
return
false
;
}
S
UdfUvSession
*
session
=
(
S
UdfUvSession
*
)
handle
;
S
ClientUdfUvSession
*
session
=
(
SClient
UdfUvSession
*
)
handle
;
SUdfAggRes
*
udfRes
=
(
SUdfAggRes
*
)
GET_ROWCELL_INTERBUF
(
pResultCellInfo
);
int32_t
envSize
=
sizeof
(
SUdfAggRes
)
+
session
->
outputLen
+
session
->
bufSize
;
memset
(
udfRes
,
0
,
envSize
);
...
...
@@ -1250,7 +1264,7 @@ bool udfAggInit(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResult
udfRes
->
finalResBuf
=
(
char
*
)
udfRes
+
sizeof
(
SUdfAggRes
);
udfRes
->
interResBuf
=
(
char
*
)
udfRes
+
sizeof
(
SUdfAggRes
)
+
session
->
outputLen
;
udfRes
->
session
=
(
SUdfUvSession
*
)
handle
;
udfRes
->
session
=
(
S
Client
UdfUvSession
*
)
handle
;
SUdfInterBuf
buf
=
{
0
};
if
(
callUdfAggInit
(
handle
,
&
buf
)
!=
0
)
{
return
false
;
...
...
@@ -1265,7 +1279,7 @@ int32_t udfAggProcess(struct SqlFunctionCtx *pCtx) {
int32_t
numOfCols
=
pInput
->
numOfInputCols
;
SUdfAggRes
*
udfRes
=
(
SUdfAggRes
*
)
GET_ROWCELL_INTERBUF
(
GET_RES_INFO
(
pCtx
));
SUdfUvSession
*
session
=
udfRes
->
session
;
S
Client
UdfUvSession
*
session
=
udfRes
->
session
;
udfRes
->
finalResBuf
=
(
char
*
)
udfRes
+
sizeof
(
SUdfAggRes
);
udfRes
->
interResBuf
=
(
char
*
)
udfRes
+
sizeof
(
SUdfAggRes
)
+
session
->
outputLen
;
...
...
@@ -1315,7 +1329,7 @@ int32_t udfAggProcess(struct SqlFunctionCtx *pCtx) {
int32_t
udfAggFinalize
(
struct
SqlFunctionCtx
*
pCtx
,
SSDataBlock
*
pBlock
)
{
SUdfAggRes
*
udfRes
=
(
SUdfAggRes
*
)
GET_ROWCELL_INTERBUF
(
GET_RES_INFO
(
pCtx
));
SUdfUvSession
*
session
=
udfRes
->
session
;
S
Client
UdfUvSession
*
session
=
udfRes
->
session
;
udfRes
->
finalResBuf
=
(
char
*
)
udfRes
+
sizeof
(
SUdfAggRes
);
udfRes
->
interResBuf
=
(
char
*
)
udfRes
+
sizeof
(
SUdfAggRes
)
+
session
->
outputLen
;
...
...
source/libs/function/test/udf1.c
浏览文件 @
f8dd3a13
...
...
@@ -26,11 +26,18 @@ int32_t udf1(SUdfDataBlock* block, SUdfColumn *resultCol) {
SUdfColumnData
*
resultData
=
&
resultCol
->
colData
;
resultData
->
numOfRows
=
block
->
numOfRows
;
SUdfColumnData
*
srcData
=
&
block
->
udfCols
[
0
]
->
colData
;
for
(
int32_t
i
=
0
;
i
<
resultData
->
numOfRows
;
++
i
)
{
int32_t
luckyNum
=
88
;
udfColSetRow
(
resultCol
,
i
,
(
char
*
)
&
luckyNum
,
false
);
int
j
=
0
;
for
(;
j
<
block
->
numOfCols
;
++
j
)
{
if
(
udfColDataIsNull
(
block
->
udfCols
[
j
],
i
))
{
udfColDataSetNull
(
resultCol
,
i
);
break
;
}
}
if
(
j
==
block
->
numOfCols
)
{
int32_t
luckyNum
=
88
;
udfColDataSet
(
resultCol
,
i
,
(
char
*
)
&
luckyNum
,
false
);
}
}
return
0
;
...
...
source/libs/function/test/udf2.c
浏览文件 @
f8dd3a13
...
...
@@ -26,24 +26,34 @@ int32_t udf2_start(SUdfInterBuf *buf) {
int32_t
udf2
(
SUdfDataBlock
*
block
,
SUdfInterBuf
*
interBuf
,
SUdfInterBuf
*
newInterBuf
)
{
int64_t
sumSquares
=
*
(
int64_t
*
)
interBuf
->
buf
;
int8_t
numOutput
=
0
;
for
(
int32_t
i
=
0
;
i
<
block
->
numOfCols
;
++
i
)
{
for
(
int32_t
j
=
0
;
j
<
block
->
numOfRows
;
++
j
)
{
SUdfColumn
*
col
=
block
->
udfCols
[
i
];
//TODO: check the bitmap for null value
int32_t
*
rows
=
(
int32_t
*
)
col
->
colData
.
fixLenCol
.
data
;
sumSquares
+=
rows
[
j
]
*
rows
[
j
];
if
(
udfColDataIsNull
(
col
,
j
))
{
continue
;
}
char
*
cell
=
udfColDataGetData
(
col
,
j
);
int32_t
num
=
*
(
int32_t
*
)
cell
;
sumSquares
+=
num
*
num
;
numOutput
=
1
;
}
}
*
(
int64_t
*
)(
newInterBuf
->
buf
)
=
sumSquares
;
newInterBuf
->
bufLen
=
sizeof
(
int64_t
);
//TODO: if all null value, numOfResult = 0;
newInterBuf
->
numOfResult
=
1
;
if
(
numOutput
==
1
)
{
*
(
int64_t
*
)(
newInterBuf
->
buf
)
=
sumSquares
;
newInterBuf
->
bufLen
=
sizeof
(
int64_t
);
}
newInterBuf
->
numOfResult
=
numOutput
;
return
0
;
}
int32_t
udf2_finish
(
SUdfInterBuf
*
buf
,
SUdfInterBuf
*
resultData
)
{
//TODO: check numOfResults;
if
(
buf
->
numOfResult
==
0
)
{
resultData
->
numOfResult
=
0
;
return
0
;
}
int64_t
sumSquares
=
*
(
int64_t
*
)(
buf
->
buf
);
*
(
double
*
)(
resultData
->
buf
)
=
sqrt
(
sumSquares
);
resultData
->
bufLen
=
sizeof
(
double
);
...
...
tests/script/tsim/query/udf.sim
浏览文件 @
f8dd3a13
...
...
@@ -66,4 +66,4 @@ endi
#sql drop function udf1;
#sql drop function udf2;
system sh/exec.sh -n dnode1 -s stop -x SIG
KILL
system sh/exec.sh -n dnode1 -s stop -x SIG
TERM
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录