Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
51509110
TDengine
项目概览
taosdata
/
TDengine
1 年多 前同步成功
通知
1185
Star
22016
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
提交
51509110
编写于
1月 10, 2022
作者:
H
Hongze Cheng
浏览文件
操作
浏览文件
下载
差异文件
Merge branch 'feature/vnode' of
https://github.com/taosdata/TDengine
into feature/vnode
上级
49be2ab4
73675bc5
变更
17
隐藏空白更改
内联
并排
Showing
17 changed file
with
155 addition
and
150 deletion
+155
-150
source/client/src/clientImpl.c
source/client/src/clientImpl.c
+2
-0
source/dnode/mnode/impl/inc/mndVgroup.h
source/dnode/mnode/impl/inc/mndVgroup.h
+2
-2
source/dnode/mnode/impl/src/mndAcct.c
source/dnode/mnode/impl/src/mndAcct.c
+1
-1
source/dnode/mnode/impl/src/mndBnode.c
source/dnode/mnode/impl/src/mndBnode.c
+1
-1
source/dnode/mnode/impl/src/mndCluster.c
source/dnode/mnode/impl/src/mndCluster.c
+1
-1
source/dnode/mnode/impl/src/mndDb.c
source/dnode/mnode/impl/src/mndDb.c
+85
-126
source/dnode/mnode/impl/src/mndDnode.c
source/dnode/mnode/impl/src/mndDnode.c
+1
-1
source/dnode/mnode/impl/src/mndFunc.c
source/dnode/mnode/impl/src/mndFunc.c
+1
-1
source/dnode/mnode/impl/src/mndMnode.c
source/dnode/mnode/impl/src/mndMnode.c
+1
-1
source/dnode/mnode/impl/src/mndQnode.c
source/dnode/mnode/impl/src/mndQnode.c
+1
-1
source/dnode/mnode/impl/src/mndSnode.c
source/dnode/mnode/impl/src/mndSnode.c
+1
-1
source/dnode/mnode/impl/src/mndStb.c
source/dnode/mnode/impl/src/mndStb.c
+1
-1
source/dnode/mnode/impl/src/mndUser.c
source/dnode/mnode/impl/src/mndUser.c
+1
-1
source/dnode/mnode/impl/src/mndVgroup.c
source/dnode/mnode/impl/src/mndVgroup.c
+3
-3
source/libs/parser/test/mockCatalogService.cpp
source/libs/parser/test/mockCatalogService.cpp
+3
-0
source/libs/planner/src/physicalPlan.c
source/libs/planner/src/physicalPlan.c
+1
-1
source/libs/planner/src/physicalPlanJson.c
source/libs/planner/src/physicalPlanJson.c
+49
-8
未找到文件。
source/client/src/clientImpl.c
浏览文件 @
51509110
...
...
@@ -305,6 +305,8 @@ TAOS_RES *taos_query_l(TAOS *taos, const char *sql, int sqlLen) {
}
else
{
CHECK_CODE_GOTO
(
getPlan
(
pRequest
,
pQuery
,
&
pDag
),
_return
);
CHECK_CODE_GOTO
(
scheduleQuery
(
pRequest
,
pDag
,
&
pJob
),
_return
);
pRequest
->
code
=
terrno
;
return
pRequest
;
}
_return:
...
...
source/dnode/mnode/impl/inc/mndVgroup.h
浏览文件 @
51509110
...
...
@@ -31,8 +31,8 @@ int32_t mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups);
SEpSet
mndGetVgroupEpset
(
SMnode
*
pMnode
,
SVgObj
*
pVgroup
);
int32_t
mndGetVnodesNum
(
SMnode
*
pMnode
,
int32_t
dnodeId
);
SCreateVnodeReq
*
mndBuildCreateVnode
Msg
(
SMnode
*
pMnode
,
SDnodeObj
*
pDnode
,
SDbObj
*
pDb
,
SVgObj
*
pVgroup
);
SDropVnodeReq
*
mndBuildDropVnode
Msg
(
SMnode
*
pMnode
,
SDnodeObj
*
pDnode
,
SDbObj
*
pDb
,
SVgObj
*
pVgroup
);
SCreateVnodeReq
*
mndBuildCreateVnode
Req
(
SMnode
*
pMnode
,
SDnodeObj
*
pDnode
,
SDbObj
*
pDb
,
SVgObj
*
pVgroup
);
SDropVnodeReq
*
mndBuildDropVnode
Req
(
SMnode
*
pMnode
,
SDnodeObj
*
pDnode
,
SDbObj
*
pDb
,
SVgObj
*
pVgroup
);
#ifdef __cplusplus
}
...
...
source/dnode/mnode/impl/src/mndAcct.c
浏览文件 @
51509110
...
...
@@ -177,7 +177,7 @@ static int32_t mndAcctActionDelete(SSdb *pSdb, SAcctObj *pAcct) {
}
static
int32_t
mndAcctActionUpdate
(
SSdb
*
pSdb
,
SAcctObj
*
pOld
,
SAcctObj
*
pNew
)
{
mTrace
(
"acct:%s, perform update action, old
_row:%p new_
row:%p"
,
pOld
->
acct
,
pOld
,
pNew
);
mTrace
(
"acct:%s, perform update action, old
row:%p new
row:%p"
,
pOld
->
acct
,
pOld
,
pNew
);
pOld
->
updateTime
=
pNew
->
updateTime
;
pOld
->
status
=
pNew
->
status
;
...
...
source/dnode/mnode/impl/src/mndBnode.c
浏览文件 @
51509110
...
...
@@ -155,7 +155,7 @@ static int32_t mndBnodeActionDelete(SSdb *pSdb, SBnodeObj *pObj) {
}
static
int32_t
mndBnodeActionUpdate
(
SSdb
*
pSdb
,
SBnodeObj
*
pOld
,
SBnodeObj
*
pNew
)
{
mTrace
(
"bnode:%d, perform update action, old
_row:%p new_
row:%p"
,
pOld
->
id
,
pOld
,
pNew
);
mTrace
(
"bnode:%d, perform update action, old
row:%p new
row:%p"
,
pOld
->
id
,
pOld
,
pNew
);
pOld
->
updateTime
=
pNew
->
updateTime
;
return
0
;
}
...
...
source/dnode/mnode/impl/src/mndCluster.c
浏览文件 @
51509110
...
...
@@ -135,7 +135,7 @@ static int32_t mndClusterActionDelete(SSdb *pSdb, SClusterObj *pCluster) {
}
static
int32_t
mndClusterActionUpdate
(
SSdb
*
pSdb
,
SClusterObj
*
pOld
,
SClusterObj
*
pNew
)
{
mTrace
(
"cluster:%"
PRId64
", perform update action, old
_row:%p new_
row:%p"
,
pOld
->
id
,
pOld
,
pNew
);
mTrace
(
"cluster:%"
PRId64
", perform update action, old
row:%p new
row:%p"
,
pOld
->
id
,
pOld
,
pNew
);
return
0
;
}
...
...
source/dnode/mnode/impl/src/mndDb.c
浏览文件 @
51509110
...
...
@@ -28,7 +28,7 @@ static SSdbRaw *mndDbActionEncode(SDbObj *pDb);
static
SSdbRow
*
mndDbActionDecode
(
SSdbRaw
*
pRaw
);
static
int32_t
mndDbActionInsert
(
SSdb
*
pSdb
,
SDbObj
*
pDb
);
static
int32_t
mndDbActionDelete
(
SSdb
*
pSdb
,
SDbObj
*
pDb
);
static
int32_t
mndDbActionUpdate
(
SSdb
*
pSdb
,
SDbObj
*
pOld
Db
,
SDbObj
*
pNewDb
);
static
int32_t
mndDbActionUpdate
(
SSdb
*
pSdb
,
SDbObj
*
pOld
,
SDbObj
*
pNew
);
static
int32_t
mndProcessCreateDbReq
(
SMnodeMsg
*
pReq
);
static
int32_t
mndProcessAlterDbReq
(
SMnodeMsg
*
pReq
);
static
int32_t
mndProcessDropDbReq
(
SMnodeMsg
*
pReq
);
...
...
@@ -182,12 +182,12 @@ static int32_t mndDbActionDelete(SSdb *pSdb, SDbObj *pDb) {
return
0
;
}
static
int32_t
mndDbActionUpdate
(
SSdb
*
pSdb
,
SDbObj
*
pOld
Db
,
SDbObj
*
pNewDb
)
{
mTrace
(
"db:%s, perform update action, old
_row:%p new_row:%p"
,
pOldDb
->
name
,
pOldDb
,
pNewDb
);
pOld
Db
->
updateTime
=
pNewDb
->
updateTime
;
pOld
Db
->
cfgVersion
=
pNewDb
->
cfgVersion
;
pOld
Db
->
vgVersion
=
pNewDb
->
vgVersion
;
memcpy
(
&
pOld
Db
->
cfg
,
&
pNewDb
->
cfg
,
sizeof
(
SDbCfg
));
static
int32_t
mndDbActionUpdate
(
SSdb
*
pSdb
,
SDbObj
*
pOld
,
SDbObj
*
pNew
)
{
mTrace
(
"db:%s, perform update action, old
row:%p new row:%p"
,
pOld
->
name
,
pOld
,
pNew
);
pOld
->
updateTime
=
pNew
->
updateTime
;
pOld
->
cfgVersion
=
pNew
->
cfgVersion
;
pOld
->
vgVersion
=
pNew
->
vgVersion
;
memcpy
(
&
pOld
->
cfg
,
&
pNew
->
cfg
,
sizeof
(
SDbCfg
));
return
0
;
}
...
...
@@ -331,14 +331,15 @@ static int32_t mndSetCreateDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj
action
.
epSet
=
mndGetDnodeEpset
(
pDnode
);
mndReleaseDnode
(
pMnode
,
pDnode
);
SCreateVnodeReq
*
p
Msg
=
mndBuildCreateVnodeMsg
(
pMnode
,
pDnode
,
pDb
,
pVgroup
);
if
(
p
Msg
==
NULL
)
return
-
1
;
SCreateVnodeReq
*
p
Req
=
mndBuildCreateVnodeReq
(
pMnode
,
pDnode
,
pDb
,
pVgroup
);
if
(
p
Req
==
NULL
)
return
-
1
;
action
.
pCont
=
p
Msg
;
action
.
pCont
=
p
Req
;
action
.
contLen
=
sizeof
(
SCreateVnodeReq
);
action
.
msgType
=
TDMT_DND_CREATE_VNODE
;
action
.
acceptableCode
=
TSDB_CODE_DND_VNODE_ALREADY_DEPLOYED
;
if
(
mndTransAppendRedoAction
(
pTrans
,
&
action
)
!=
0
)
{
free
(
p
Msg
);
free
(
p
Req
);
return
-
1
;
}
}
...
...
@@ -360,14 +361,15 @@ static int32_t mndSetCreateDbUndoActions(SMnode *pMnode, STrans *pTrans, SDbObj
action
.
epSet
=
mndGetDnodeEpset
(
pDnode
);
mndReleaseDnode
(
pMnode
,
pDnode
);
SDropVnodeReq
*
p
Msg
=
mndBuildDropVnodeMsg
(
pMnode
,
pDnode
,
pDb
,
pVgroup
);
if
(
p
Msg
==
NULL
)
return
-
1
;
SDropVnodeReq
*
p
Req
=
mndBuildDropVnodeReq
(
pMnode
,
pDnode
,
pDb
,
pVgroup
);
if
(
p
Req
==
NULL
)
return
-
1
;
action
.
pCont
=
p
Msg
;
action
.
pCont
=
p
Req
;
action
.
contLen
=
sizeof
(
SDropVnodeReq
);
action
.
msgType
=
TDMT_DND_DROP_VNODE
;
action
.
acceptableCode
=
TSDB_CODE_DND_VNODE_NOT_DEPLOYED
;
if
(
mndTransAppendUndoAction
(
pTrans
,
&
action
)
!=
0
)
{
free
(
p
Msg
);
free
(
p
Req
);
return
-
1
;
}
}
...
...
@@ -376,7 +378,7 @@ static int32_t mndSetCreateDbUndoActions(SMnode *pMnode, STrans *pTrans, SDbObj
return
0
;
}
static
int32_t
mndCreateDb
(
SMnode
*
pMnode
,
SMnodeMsg
*
p
Msg
,
SCreateDbMsg
*
pCreate
,
SUserObj
*
pUser
)
{
static
int32_t
mndCreateDb
(
SMnode
*
pMnode
,
SMnodeMsg
*
p
Req
,
SCreateDbMsg
*
pCreate
,
SUserObj
*
pUser
)
{
SDbObj
dbObj
=
{
0
};
memcpy
(
dbObj
.
name
,
pCreate
->
db
,
TSDB_DB_FNAME_LEN
);
memcpy
(
dbObj
.
acct
,
pUser
->
acct
,
TSDB_USER_LEN
);
...
...
@@ -425,43 +427,17 @@ static int32_t mndCreateDb(SMnode *pMnode, SMnodeMsg *pMsg, SCreateDbMsg *pCreat
}
int32_t
code
=
-
1
;
STrans
*
pTrans
=
mndTransCreate
(
pMnode
,
TRN_POLICY_ROLLBACK
,
&
pMsg
->
rpcMsg
);
if
(
pTrans
==
NULL
)
{
mError
(
"db:%s, failed to create since %s"
,
pCreate
->
db
,
terrstr
());
goto
CREATE_DB_OVER
;
}
STrans
*
pTrans
=
mndTransCreate
(
pMnode
,
TRN_POLICY_ROLLBACK
,
&
pReq
->
rpcMsg
);
if
(
pTrans
==
NULL
)
goto
CREATE_DB_OVER
;
mDebug
(
"trans:%d, used to create db:%s"
,
pTrans
->
id
,
pCreate
->
db
);
if
(
mndSetCreateDbRedoLogs
(
pMnode
,
pTrans
,
&
dbObj
,
pVgroups
)
!=
0
)
{
mError
(
"trans:%d, failed to set redo log since %s"
,
pTrans
->
id
,
terrstr
());
goto
CREATE_DB_OVER
;
}
if
(
mndSetCreateDbUndoLogs
(
pMnode
,
pTrans
,
&
dbObj
,
pVgroups
)
!=
0
)
{
mError
(
"trans:%d, failed to set undo log since %s"
,
pTrans
->
id
,
terrstr
());
goto
CREATE_DB_OVER
;
}
if
(
mndSetCreateDbCommitLogs
(
pMnode
,
pTrans
,
&
dbObj
,
pVgroups
)
!=
0
)
{
mError
(
"trans:%d, failed to set commit log since %s"
,
pTrans
->
id
,
terrstr
());
goto
CREATE_DB_OVER
;
}
if
(
mndSetCreateDbRedoActions
(
pMnode
,
pTrans
,
&
dbObj
,
pVgroups
)
!=
0
)
{
mError
(
"trans:%d, failed to set redo actions since %s"
,
pTrans
->
id
,
terrstr
());
goto
CREATE_DB_OVER
;
}
if
(
mndSetCreateDbUndoActions
(
pMnode
,
pTrans
,
&
dbObj
,
pVgroups
)
!=
0
)
{
mError
(
"trans:%d, failed to set redo actions since %s"
,
pTrans
->
id
,
terrstr
());
goto
CREATE_DB_OVER
;
}
if
(
mndTransPrepare
(
pMnode
,
pTrans
)
!=
0
)
{
mError
(
"trans:%d, failed to prepare since %s"
,
pTrans
->
id
,
terrstr
());
goto
CREATE_DB_OVER
;
}
if
(
mndSetCreateDbRedoLogs
(
pMnode
,
pTrans
,
&
dbObj
,
pVgroups
)
!=
0
)
goto
CREATE_DB_OVER
;
if
(
mndSetCreateDbUndoLogs
(
pMnode
,
pTrans
,
&
dbObj
,
pVgroups
)
!=
0
)
goto
CREATE_DB_OVER
;
if
(
mndSetCreateDbCommitLogs
(
pMnode
,
pTrans
,
&
dbObj
,
pVgroups
)
!=
0
)
goto
CREATE_DB_OVER
;
if
(
mndSetCreateDbRedoActions
(
pMnode
,
pTrans
,
&
dbObj
,
pVgroups
)
!=
0
)
goto
CREATE_DB_OVER
;
if
(
mndSetCreateDbUndoActions
(
pMnode
,
pTrans
,
&
dbObj
,
pVgroups
)
!=
0
)
goto
CREATE_DB_OVER
;
if
(
mndTransPrepare
(
pMnode
,
pTrans
)
!=
0
)
goto
CREATE_DB_OVER
;
code
=
0
;
...
...
@@ -471,9 +447,9 @@ CREATE_DB_OVER:
return
code
;
}
static
int32_t
mndProcessCreateDbReq
(
SMnodeMsg
*
p
Msg
)
{
SMnode
*
pMnode
=
p
Msg
->
pMnode
;
SCreateDbMsg
*
pCreate
=
p
Msg
->
rpcMsg
.
pCont
;
static
int32_t
mndProcessCreateDbReq
(
SMnodeMsg
*
p
Req
)
{
SMnode
*
pMnode
=
p
Req
->
pMnode
;
SCreateDbMsg
*
pCreate
=
p
Req
->
rpcMsg
.
pCont
;
pCreate
->
numOfVgroups
=
htonl
(
pCreate
->
numOfVgroups
);
pCreate
->
cacheBlockSize
=
htonl
(
pCreate
->
cacheBlockSize
);
...
...
@@ -502,13 +478,13 @@ static int32_t mndProcessCreateDbReq(SMnodeMsg *pMsg) {
}
}
SUserObj
*
pOperUser
=
mndAcquireUser
(
pMnode
,
p
Msg
->
user
);
SUserObj
*
pOperUser
=
mndAcquireUser
(
pMnode
,
p
Req
->
user
);
if
(
pOperUser
==
NULL
)
{
mError
(
"db:%s, failed to create since %s"
,
pCreate
->
db
,
terrstr
());
return
-
1
;
}
int32_t
code
=
mndCreateDb
(
pMnode
,
p
Msg
,
pCreate
,
pOperUser
);
int32_t
code
=
mndCreateDb
(
pMnode
,
p
Req
,
pCreate
,
pOperUser
);
mndReleaseUser
(
pMnode
,
pOperUser
);
if
(
code
!=
0
)
{
...
...
@@ -565,8 +541,8 @@ static int32_t mndSetDbCfgFromAlterDbMsg(SDbObj *pDb, SAlterDbMsg *pAlter) {
return
terrno
;
}
static
int32_t
mndSetUpdateDbRedoLogs
(
SMnode
*
pMnode
,
STrans
*
pTrans
,
SDbObj
*
pOld
Db
,
SDbObj
*
pNewDb
)
{
SSdbRaw
*
pRedoRaw
=
mndDbActionEncode
(
pOld
Db
);
static
int32_t
mndSetUpdateDbRedoLogs
(
SMnode
*
pMnode
,
STrans
*
pTrans
,
SDbObj
*
pOld
,
SDbObj
*
pNew
)
{
SSdbRaw
*
pRedoRaw
=
mndDbActionEncode
(
pOld
);
if
(
pRedoRaw
==
NULL
)
return
-
1
;
if
(
mndTransAppendRedolog
(
pTrans
,
pRedoRaw
)
!=
0
)
return
-
1
;
if
(
sdbSetRawStatus
(
pRedoRaw
,
SDB_STATUS_UPDATING
)
!=
0
)
return
-
1
;
...
...
@@ -574,8 +550,8 @@ static int32_t mndSetUpdateDbRedoLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pO
return
0
;
}
static
int32_t
mndSetUpdateDbCommitLogs
(
SMnode
*
pMnode
,
STrans
*
pTrans
,
SDbObj
*
pOld
Db
,
SDbObj
*
pNewDb
)
{
SSdbRaw
*
pCommitRaw
=
mndDbActionEncode
(
pNew
Db
);
static
int32_t
mndSetUpdateDbCommitLogs
(
SMnode
*
pMnode
,
STrans
*
pTrans
,
SDbObj
*
pOld
,
SDbObj
*
pNew
)
{
SSdbRaw
*
pCommitRaw
=
mndDbActionEncode
(
pNew
);
if
(
pCommitRaw
==
NULL
)
return
-
1
;
if
(
mndTransAppendCommitlog
(
pTrans
,
pCommitRaw
)
!=
0
)
return
-
1
;
if
(
sdbSetRawStatus
(
pCommitRaw
,
SDB_STATUS_READY
)
!=
0
)
return
-
1
;
...
...
@@ -593,14 +569,14 @@ static int32_t mndBuildUpdateVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj
action
.
epSet
=
mndGetDnodeEpset
(
pDnode
);
mndReleaseDnode
(
pMnode
,
pDnode
);
SAlterVnodeReq
*
p
Msg
=
(
SAlterVnodeReq
*
)
mndBuildCreateVnodeMsg
(
pMnode
,
pDnode
,
pDb
,
pVgroup
);
if
(
p
Msg
==
NULL
)
return
-
1
;
SAlterVnodeReq
*
p
Req
=
(
SAlterVnodeReq
*
)
mndBuildCreateVnodeReq
(
pMnode
,
pDnode
,
pDb
,
pVgroup
);
if
(
p
Req
==
NULL
)
return
-
1
;
action
.
pCont
=
p
Msg
;
action
.
pCont
=
p
Req
;
action
.
contLen
=
sizeof
(
SAlterVnodeReq
);
action
.
msgType
=
TDMT_DND_ALTER_VNODE
;
if
(
mndTransAppendRedoAction
(
pTrans
,
&
action
)
!=
0
)
{
free
(
p
Msg
);
free
(
p
Req
);
return
-
1
;
}
}
...
...
@@ -608,7 +584,7 @@ static int32_t mndBuildUpdateVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj
return
0
;
}
static
int32_t
mndSetUpdateDbRedoActions
(
SMnode
*
pMnode
,
STrans
*
pTrans
,
SDbObj
*
pOld
Db
,
SDbObj
*
pNewDb
)
{
static
int32_t
mndSetUpdateDbRedoActions
(
SMnode
*
pMnode
,
STrans
*
pTrans
,
SDbObj
*
pOld
,
SDbObj
*
pNew
)
{
SSdb
*
pSdb
=
pMnode
->
pSdb
;
void
*
pIter
=
NULL
;
...
...
@@ -617,8 +593,8 @@ static int32_t mndSetUpdateDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj
pIter
=
sdbFetch
(
pSdb
,
SDB_VGROUP
,
pIter
,
(
void
**
)
&
pVgroup
);
if
(
pIter
==
NULL
)
break
;
if
(
pVgroup
->
dbUid
==
pNew
Db
->
uid
)
{
if
(
mndBuildUpdateVgroupAction
(
pMnode
,
pTrans
,
pNew
Db
,
pVgroup
)
!=
0
)
{
if
(
pVgroup
->
dbUid
==
pNew
->
uid
)
{
if
(
mndBuildUpdateVgroupAction
(
pMnode
,
pTrans
,
pNew
,
pVgroup
)
!=
0
)
{
sdbCancelFetch
(
pSdb
,
pIter
);
sdbRelease
(
pSdb
,
pVgroup
);
return
-
1
;
...
...
@@ -631,27 +607,27 @@ static int32_t mndSetUpdateDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj
return
0
;
}
static
int32_t
mndUpdateDb
(
SMnode
*
pMnode
,
SMnodeMsg
*
p
Msg
,
SDbObj
*
pOldDb
,
SDbObj
*
pNewDb
)
{
static
int32_t
mndUpdateDb
(
SMnode
*
pMnode
,
SMnodeMsg
*
p
Req
,
SDbObj
*
pOld
,
SDbObj
*
pNew
)
{
int32_t
code
=
-
1
;
STrans
*
pTrans
=
mndTransCreate
(
pMnode
,
TRN_POLICY_RETRY
,
&
p
Msg
->
rpcMsg
);
STrans
*
pTrans
=
mndTransCreate
(
pMnode
,
TRN_POLICY_RETRY
,
&
p
Req
->
rpcMsg
);
if
(
pTrans
==
NULL
)
{
mError
(
"db:%s, failed to update since %s"
,
pOld
Db
->
name
,
terrstr
());
mError
(
"db:%s, failed to update since %s"
,
pOld
->
name
,
terrstr
());
return
terrno
;
}
mDebug
(
"trans:%d, used to update db:%s"
,
pTrans
->
id
,
pOld
Db
->
name
);
mDebug
(
"trans:%d, used to update db:%s"
,
pTrans
->
id
,
pOld
->
name
);
if
(
mndSetUpdateDbRedoLogs
(
pMnode
,
pTrans
,
pOld
Db
,
pNewDb
)
!=
0
)
{
if
(
mndSetUpdateDbRedoLogs
(
pMnode
,
pTrans
,
pOld
,
pNew
)
!=
0
)
{
mError
(
"trans:%d, failed to set redo log since %s"
,
pTrans
->
id
,
terrstr
());
goto
UPDATE_DB_OVER
;
}
if
(
mndSetUpdateDbCommitLogs
(
pMnode
,
pTrans
,
pOld
Db
,
pNewDb
)
!=
0
)
{
if
(
mndSetUpdateDbCommitLogs
(
pMnode
,
pTrans
,
pOld
,
pNew
)
!=
0
)
{
mError
(
"trans:%d, failed to set commit log since %s"
,
pTrans
->
id
,
terrstr
());
goto
UPDATE_DB_OVER
;
}
if
(
mndSetUpdateDbRedoActions
(
pMnode
,
pTrans
,
pOld
Db
,
pNewDb
)
!=
0
)
{
if
(
mndSetUpdateDbRedoActions
(
pMnode
,
pTrans
,
pOld
,
pNew
)
!=
0
)
{
mError
(
"trans:%d, failed to set redo actions since %s"
,
pTrans
->
id
,
terrstr
());
goto
UPDATE_DB_OVER
;
}
...
...
@@ -668,9 +644,9 @@ UPDATE_DB_OVER:
return
code
;
}
static
int32_t
mndProcessAlterDbReq
(
SMnodeMsg
*
p
Msg
)
{
SMnode
*
pMnode
=
p
Msg
->
pMnode
;
SAlterDbMsg
*
pAlter
=
p
Msg
->
rpcMsg
.
pCont
;
static
int32_t
mndProcessAlterDbReq
(
SMnodeMsg
*
p
Req
)
{
SMnode
*
pMnode
=
p
Req
->
pMnode
;
SAlterDbMsg
*
pAlter
=
p
Req
->
rpcMsg
.
pCont
;
pAlter
->
totalBlocks
=
htonl
(
pAlter
->
totalBlocks
);
pAlter
->
daysToKeep0
=
htonl
(
pAlter
->
daysToKeep0
);
pAlter
->
daysToKeep1
=
htonl
(
pAlter
->
daysToKeep1
);
...
...
@@ -697,7 +673,7 @@ static int32_t mndProcessAlterDbReq(SMnodeMsg *pMsg) {
dbObj
.
cfgVersion
++
;
dbObj
.
updateTime
=
taosGetTimestampMs
();
code
=
mndUpdateDb
(
pMnode
,
p
Msg
,
pDb
,
&
dbObj
);
code
=
mndUpdateDb
(
pMnode
,
p
Req
,
pDb
,
&
dbObj
);
mndReleaseDb
(
pMnode
,
pDb
);
if
(
code
!=
0
)
{
...
...
@@ -757,14 +733,15 @@ static int32_t mndBuildDropVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *
action
.
epSet
=
mndGetDnodeEpset
(
pDnode
);
mndReleaseDnode
(
pMnode
,
pDnode
);
SDropVnodeReq
*
p
Msg
=
mndBuildDropVnodeMsg
(
pMnode
,
pDnode
,
pDb
,
pVgroup
);
if
(
p
Msg
==
NULL
)
return
-
1
;
SDropVnodeReq
*
p
Req
=
mndBuildDropVnodeReq
(
pMnode
,
pDnode
,
pDb
,
pVgroup
);
if
(
p
Req
==
NULL
)
return
-
1
;
action
.
pCont
=
p
Msg
;
action
.
pCont
=
p
Req
;
action
.
contLen
=
sizeof
(
SCreateVnodeReq
);
action
.
msgType
=
TDMT_DND_DROP_VNODE
;
action
.
acceptableCode
=
TSDB_CODE_DND_VNODE_NOT_DEPLOYED
;
if
(
mndTransAppendRedoAction
(
pTrans
,
&
action
)
!=
0
)
{
free
(
p
Msg
);
free
(
p
Req
);
return
-
1
;
}
}
...
...
@@ -795,35 +772,17 @@ static int32_t mndSetDropDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *p
return
0
;
}
static
int32_t
mndDropDb
(
SMnode
*
pMnode
,
SMnodeMsg
*
p
Msg
,
SDbObj
*
pDb
)
{
static
int32_t
mndDropDb
(
SMnode
*
pMnode
,
SMnodeMsg
*
p
Req
,
SDbObj
*
pDb
)
{
int32_t
code
=
-
1
;
STrans
*
pTrans
=
mndTransCreate
(
pMnode
,
TRN_POLICY_RETRY
,
&
pMsg
->
rpcMsg
);
if
(
pTrans
==
NULL
)
{
mError
(
"db:%s, failed to drop since %s"
,
pDb
->
name
,
terrstr
());
return
-
1
;
}
STrans
*
pTrans
=
mndTransCreate
(
pMnode
,
TRN_POLICY_RETRY
,
&
pReq
->
rpcMsg
);
if
(
pTrans
==
NULL
)
goto
DROP_DB_OVER
;
mDebug
(
"trans:%d, used to drop db:%s"
,
pTrans
->
id
,
pDb
->
name
);
if
(
mndSetDropDbRedoLogs
(
pMnode
,
pTrans
,
pDb
)
!=
0
)
{
mError
(
"trans:%d, failed to set redo log since %s"
,
pTrans
->
id
,
terrstr
());
goto
DROP_DB_OVER
;
}
if
(
mndSetDropDbCommitLogs
(
pMnode
,
pTrans
,
pDb
)
!=
0
)
{
mError
(
"trans:%d, failed to set commit log since %s"
,
pTrans
->
id
,
terrstr
());
goto
DROP_DB_OVER
;
}
if
(
mndSetDropDbRedoActions
(
pMnode
,
pTrans
,
pDb
)
!=
0
)
{
mError
(
"trans:%d, failed to set redo actions since %s"
,
pTrans
->
id
,
terrstr
());
goto
DROP_DB_OVER
;
}
if
(
mndTransPrepare
(
pMnode
,
pTrans
)
!=
0
)
{
mError
(
"trans:%d, failed to prepare since %s"
,
pTrans
->
id
,
terrstr
());
goto
DROP_DB_OVER
;
}
if
(
mndSetDropDbRedoLogs
(
pMnode
,
pTrans
,
pDb
)
!=
0
)
goto
DROP_DB_OVER
;
if
(
mndSetDropDbCommitLogs
(
pMnode
,
pTrans
,
pDb
)
!=
0
)
goto
DROP_DB_OVER
;
if
(
mndSetDropDbRedoActions
(
pMnode
,
pTrans
,
pDb
)
!=
0
)
goto
DROP_DB_OVER
;
if
(
mndTransPrepare
(
pMnode
,
pTrans
)
!=
0
)
goto
DROP_DB_OVER
;
code
=
0
;
...
...
@@ -832,9 +791,9 @@ DROP_DB_OVER:
return
code
;
}
static
int32_t
mndProcessDropDbReq
(
SMnodeMsg
*
p
Msg
)
{
SMnode
*
pMnode
=
p
Msg
->
pMnode
;
SDropDbMsg
*
pDrop
=
p
Msg
->
rpcMsg
.
pCont
;
static
int32_t
mndProcessDropDbReq
(
SMnodeMsg
*
p
Req
)
{
SMnode
*
pMnode
=
p
Req
->
pMnode
;
SDropDbMsg
*
pDrop
=
p
Req
->
rpcMsg
.
pCont
;
mDebug
(
"db:%s, start to drop"
,
pDrop
->
db
);
...
...
@@ -850,7 +809,7 @@ static int32_t mndProcessDropDbReq(SMnodeMsg *pMsg) {
}
}
int32_t
code
=
mndDropDb
(
pMnode
,
p
Msg
,
pDb
);
int32_t
code
=
mndDropDb
(
pMnode
,
p
Req
,
pDb
);
mndReleaseDb
(
pMnode
,
pDb
);
if
(
code
!=
0
)
{
...
...
@@ -861,16 +820,16 @@ static int32_t mndProcessDropDbReq(SMnodeMsg *pMsg) {
return
TSDB_CODE_MND_ACTION_IN_PROGRESS
;
}
static
int32_t
mndProcessUseDbReq
(
SMnodeMsg
*
p
Msg
)
{
SMnode
*
pMnode
=
p
Msg
->
pMnode
;
static
int32_t
mndProcessUseDbReq
(
SMnodeMsg
*
p
Req
)
{
SMnode
*
pMnode
=
p
Req
->
pMnode
;
SSdb
*
pSdb
=
pMnode
->
pSdb
;
SUseDbMsg
*
pUse
=
p
Msg
->
rpcMsg
.
pCont
;
SUseDbMsg
*
pUse
=
p
Req
->
rpcMsg
.
pCont
;
pUse
->
vgVersion
=
htonl
(
pUse
->
vgVersion
);
SDbObj
*
pDb
=
mndAcquireDb
(
pMnode
,
pUse
->
db
);
if
(
pDb
==
NULL
)
{
terrno
=
TSDB_CODE_MND_DB_NOT_EXIST
;
mError
(
"db:%s, failed to process use db
msg
since %s"
,
pUse
->
db
,
terrstr
());
mError
(
"db:%s, failed to process use db
req
since %s"
,
pUse
->
db
,
terrstr
());
return
-
1
;
}
...
...
@@ -922,19 +881,19 @@ static int32_t mndProcessUseDbReq(SMnodeMsg *pMsg) {
pRsp
->
vgNum
=
htonl
(
vindex
);
pRsp
->
hashMethod
=
pDb
->
hashMethod
;
p
Msg
->
pCont
=
pRsp
;
p
Msg
->
contLen
=
contLen
;
p
Req
->
pCont
=
pRsp
;
p
Req
->
contLen
=
contLen
;
mndReleaseDb
(
pMnode
,
pDb
);
return
0
;
}
static
int32_t
mndProcessSyncDbReq
(
SMnodeMsg
*
p
Msg
)
{
SMnode
*
pMnode
=
p
Msg
->
pMnode
;
SSyncDbMsg
*
pSync
=
p
Msg
->
rpcMsg
.
pCont
;
static
int32_t
mndProcessSyncDbReq
(
SMnodeMsg
*
p
Req
)
{
SMnode
*
pMnode
=
p
Req
->
pMnode
;
SSyncDbMsg
*
pSync
=
p
Req
->
rpcMsg
.
pCont
;
SDbObj
*
pDb
=
mndAcquireDb
(
pMnode
,
pSync
->
db
);
if
(
pDb
==
NULL
)
{
mError
(
"db:%s, failed to process sync db
msg
since %s"
,
pSync
->
db
,
terrstr
());
mError
(
"db:%s, failed to process sync db
req
since %s"
,
pSync
->
db
,
terrstr
());
return
-
1
;
}
...
...
@@ -942,12 +901,12 @@ static int32_t mndProcessSyncDbReq(SMnodeMsg *pMsg) {
return
0
;
}
static
int32_t
mndProcessCompactDbReq
(
SMnodeMsg
*
p
Msg
)
{
SMnode
*
pMnode
=
p
Msg
->
pMnode
;
SCompactDbMsg
*
pCompact
=
p
Msg
->
rpcMsg
.
pCont
;
static
int32_t
mndProcessCompactDbReq
(
SMnodeMsg
*
p
Req
)
{
SMnode
*
pMnode
=
p
Req
->
pMnode
;
SCompactDbMsg
*
pCompact
=
p
Req
->
rpcMsg
.
pCont
;
SDbObj
*
pDb
=
mndAcquireDb
(
pMnode
,
pCompact
->
db
);
if
(
pDb
==
NULL
)
{
mError
(
"db:%s, failed to process compact db
msg
since %s"
,
pCompact
->
db
,
terrstr
());
mError
(
"db:%s, failed to process compact db
req
since %s"
,
pCompact
->
db
,
terrstr
());
return
-
1
;
}
...
...
@@ -955,8 +914,8 @@ static int32_t mndProcessCompactDbReq(SMnodeMsg *pMsg) {
return
0
;
}
static
int32_t
mndGetDbMeta
(
SMnodeMsg
*
p
Msg
,
SShowObj
*
pShow
,
STableMetaMsg
*
pMeta
)
{
SMnode
*
pMnode
=
p
Msg
->
pMnode
;
static
int32_t
mndGetDbMeta
(
SMnodeMsg
*
p
Req
,
SShowObj
*
pShow
,
STableMetaMsg
*
pMeta
)
{
SMnode
*
pMnode
=
p
Req
->
pMnode
;
SSdb
*
pSdb
=
pMnode
->
pSdb
;
int32_t
cols
=
0
;
...
...
source/dnode/mnode/impl/src/mndDnode.c
浏览文件 @
51509110
...
...
@@ -183,7 +183,7 @@ static int32_t mndDnodeActionDelete(SSdb *pSdb, SDnodeObj *pDnode) {
}
static
int32_t
mndDnodeActionUpdate
(
SSdb
*
pSdb
,
SDnodeObj
*
pOld
,
SDnodeObj
*
pNew
)
{
mTrace
(
"dnode:%d, perform update action, old
_row:%p new_
row:%p"
,
pOld
->
id
,
pOld
,
pNew
);
mTrace
(
"dnode:%d, perform update action, old
row:%p new
row:%p"
,
pOld
->
id
,
pOld
,
pNew
);
pOld
->
updateTime
=
pNew
->
updateTime
;
return
0
;
}
...
...
source/dnode/mnode/impl/src/mndFunc.c
浏览文件 @
51509110
...
...
@@ -152,7 +152,7 @@ static int32_t mndFuncActionDelete(SSdb *pSdb, SFuncObj *pFunc) {
}
static
int32_t
mndFuncActionUpdate
(
SSdb
*
pSdb
,
SFuncObj
*
pOldFunc
,
SFuncObj
*
pNewFunc
)
{
mTrace
(
"func:%s, perform update action, old
_row:%p new_
row:%p"
,
pOldFunc
->
name
,
pOldFunc
,
pNewFunc
);
mTrace
(
"func:%s, perform update action, old
row:%p new
row:%p"
,
pOldFunc
->
name
,
pOldFunc
,
pNewFunc
);
return
0
;
}
...
...
source/dnode/mnode/impl/src/mndMnode.c
浏览文件 @
51509110
...
...
@@ -208,7 +208,7 @@ static int32_t mndMnodeActionDelete(SSdb *pSdb, SMnodeObj *pObj) {
}
static
int32_t
mndMnodeActionUpdate
(
SSdb
*
pSdb
,
SMnodeObj
*
pOld
,
SMnodeObj
*
pNew
)
{
mTrace
(
"mnode:%d, perform update action, old
_row:%p new_
row:%p"
,
pOld
->
id
,
pOld
,
pNew
);
mTrace
(
"mnode:%d, perform update action, old
row:%p new
row:%p"
,
pOld
->
id
,
pOld
,
pNew
);
pOld
->
updateTime
=
pNew
->
updateTime
;
return
0
;
}
...
...
source/dnode/mnode/impl/src/mndQnode.c
浏览文件 @
51509110
...
...
@@ -155,7 +155,7 @@ static int32_t mndQnodeActionDelete(SSdb *pSdb, SQnodeObj *pObj) {
}
static
int32_t
mndQnodeActionUpdate
(
SSdb
*
pSdb
,
SQnodeObj
*
pOld
,
SQnodeObj
*
pNew
)
{
mTrace
(
"qnode:%d, perform update action, old
_row:%p new_
row:%p"
,
pOld
->
id
,
pOld
,
pNew
);
mTrace
(
"qnode:%d, perform update action, old
row:%p new
row:%p"
,
pOld
->
id
,
pOld
,
pNew
);
pOld
->
updateTime
=
pNew
->
updateTime
;
return
0
;
}
...
...
source/dnode/mnode/impl/src/mndSnode.c
浏览文件 @
51509110
...
...
@@ -155,7 +155,7 @@ static int32_t mndSnodeActionDelete(SSdb *pSdb, SSnodeObj *pObj) {
}
static
int32_t
mndSnodeActionUpdate
(
SSdb
*
pSdb
,
SSnodeObj
*
pOld
,
SSnodeObj
*
pNew
)
{
mTrace
(
"snode:%d, perform update action, old
_row:%p new_
row:%p"
,
pOld
->
id
,
pOld
,
pNew
);
mTrace
(
"snode:%d, perform update action, old
row:%p new
row:%p"
,
pOld
->
id
,
pOld
,
pNew
);
pOld
->
updateTime
=
pNew
->
updateTime
;
return
0
;
}
...
...
source/dnode/mnode/impl/src/mndStb.c
浏览文件 @
51509110
...
...
@@ -178,7 +178,7 @@ static int32_t mndStbActionDelete(SSdb *pSdb, SStbObj *pStb) {
}
static
int32_t
mndStbActionUpdate
(
SSdb
*
pSdb
,
SStbObj
*
pOldStb
,
SStbObj
*
pNewStb
)
{
mTrace
(
"stb:%s, perform update action, old
_row:%p new_
row:%p"
,
pOldStb
->
name
,
pOldStb
,
pNewStb
);
mTrace
(
"stb:%s, perform update action, old
row:%p new
row:%p"
,
pOldStb
->
name
,
pOldStb
,
pNewStb
);
atomic_exchange_32
(
&
pOldStb
->
updateTime
,
pNewStb
->
updateTime
);
atomic_exchange_32
(
&
pOldStb
->
version
,
pNewStb
->
version
);
...
...
source/dnode/mnode/impl/src/mndUser.c
浏览文件 @
51509110
...
...
@@ -192,7 +192,7 @@ static int32_t mndUserActionDelete(SSdb *pSdb, SUserObj *pUser) {
}
static
int32_t
mndUserActionUpdate
(
SSdb
*
pSdb
,
SUserObj
*
pOld
,
SUserObj
*
pNew
)
{
mTrace
(
"user:%s, perform update action, old
_row:%p new_
row:%p"
,
pOld
->
user
,
pOld
,
pNew
);
mTrace
(
"user:%s, perform update action, old
row:%p new
row:%p"
,
pOld
->
user
,
pOld
,
pNew
);
memcpy
(
pOld
->
pass
,
pNew
->
pass
,
TSDB_PASSWORD_LEN
);
pOld
->
updateTime
=
pNew
->
updateTime
;
return
0
;
...
...
source/dnode/mnode/impl/src/mndVgroup.c
浏览文件 @
51509110
...
...
@@ -165,7 +165,7 @@ static int32_t mndVgroupActionDelete(SSdb *pSdb, SVgObj *pVgroup) {
}
static
int32_t
mndVgroupActionUpdate
(
SSdb
*
pSdb
,
SVgObj
*
pOldVgroup
,
SVgObj
*
pNewVgroup
)
{
mTrace
(
"vgId:%d, perform update action, old
_row:%p new_
row:%p"
,
pOldVgroup
->
vgId
,
pOldVgroup
,
pNewVgroup
);
mTrace
(
"vgId:%d, perform update action, old
row:%p new
row:%p"
,
pOldVgroup
->
vgId
,
pOldVgroup
,
pNewVgroup
);
pOldVgroup
->
updateTime
=
pNewVgroup
->
updateTime
;
pOldVgroup
->
version
=
pNewVgroup
->
version
;
pOldVgroup
->
hashBegin
=
pNewVgroup
->
hashBegin
;
...
...
@@ -189,7 +189,7 @@ void mndReleaseVgroup(SMnode *pMnode, SVgObj *pVgroup) {
sdbRelease
(
pSdb
,
pVgroup
);
}
SCreateVnodeReq
*
mndBuildCreateVnode
Msg
(
SMnode
*
pMnode
,
SDnodeObj
*
pDnode
,
SDbObj
*
pDb
,
SVgObj
*
pVgroup
)
{
SCreateVnodeReq
*
mndBuildCreateVnode
Req
(
SMnode
*
pMnode
,
SDnodeObj
*
pDnode
,
SDbObj
*
pDb
,
SVgObj
*
pVgroup
)
{
SCreateVnodeReq
*
pCreate
=
calloc
(
1
,
sizeof
(
SCreateVnodeReq
));
if
(
pCreate
==
NULL
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
...
...
@@ -248,7 +248,7 @@ SCreateVnodeReq *mndBuildCreateVnodeMsg(SMnode *pMnode, SDnodeObj *pDnode, SDbOb
return
pCreate
;
}
SDropVnodeReq
*
mndBuildDropVnode
Msg
(
SMnode
*
pMnode
,
SDnodeObj
*
pDnode
,
SDbObj
*
pDb
,
SVgObj
*
pVgroup
)
{
SDropVnodeReq
*
mndBuildDropVnode
Req
(
SMnode
*
pMnode
,
SDnodeObj
*
pDnode
,
SDbObj
*
pDb
,
SVgObj
*
pVgroup
)
{
SDropVnodeReq
*
pDrop
=
calloc
(
1
,
sizeof
(
SDropVnodeReq
));
if
(
pDrop
==
NULL
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
...
...
source/libs/parser/test/mockCatalogService.cpp
浏览文件 @
51509110
...
...
@@ -112,6 +112,9 @@ public:
int32_t
catalogGetTableHashVgroup
(
const
SName
*
pTableName
,
SVgroupInfo
*
vgInfo
)
const
{
// todo
vgInfo
->
vgId
=
1
;
vgInfo
->
numOfEps
=
1
;
vgInfo
->
epAddr
[
0
].
port
=
6030
;
strcpy
(
vgInfo
->
epAddr
[
0
].
fqdn
,
"node1"
);
return
0
;
}
...
...
source/libs/planner/src/physicalPlan.c
浏览文件 @
51509110
...
...
@@ -234,7 +234,7 @@ static SSubplan* initSubplan(SPlanContext* pCxt, int32_t type) {
static
void
vgroupInfoToEpSet
(
const
SVgroupInfo
*
vg
,
SQueryNodeAddr
*
execNode
)
{
execNode
->
nodeId
=
vg
->
vgId
;
execNode
->
inUse
=
0
;
// todo
execNode
->
inUse
=
vg
->
inUse
;
execNode
->
numOfEps
=
vg
->
numOfEps
;
for
(
int8_t
i
=
0
;
i
<
vg
->
numOfEps
;
++
i
)
{
execNode
->
epAddr
[
i
]
=
vg
->
epAddr
[
i
];
...
...
source/libs/planner/src/physicalPlanJson.c
浏览文件 @
51509110
...
...
@@ -62,7 +62,7 @@ static bool fromObjectWithAlloc(const cJSON* json, const char* name, FFromJson f
return
func
(
jObj
,
*
obj
);
}
static
bool
add
Array
(
cJSON
*
json
,
const
char
*
name
,
FToJson
func
,
const
SArray
*
array
)
{
static
bool
add
Tarray
(
cJSON
*
json
,
const
char
*
name
,
FToJson
func
,
const
SArray
*
array
,
bool
isPoint
)
{
size_t
size
=
(
NULL
==
array
)
?
0
:
taosArrayGetSize
(
array
);
if
(
size
>
0
)
{
cJSON
*
jArray
=
cJSON_AddArrayToObject
(
json
,
name
);
...
...
@@ -70,7 +70,7 @@ static bool addArray(cJSON* json, const char* name, FToJson func, const SArray*
return
false
;
}
for
(
size_t
i
=
0
;
i
<
size
;
++
i
)
{
if
(
!
addItem
(
jArray
,
func
,
taosArrayGetP
(
array
,
i
)))
{
if
(
!
addItem
(
jArray
,
func
,
isPoint
?
taosArrayGetP
(
array
,
i
)
:
taosArrayGet
(
array
,
i
)))
{
return
false
;
}
}
...
...
@@ -78,11 +78,19 @@ static bool addArray(cJSON* json, const char* name, FToJson func, const SArray*
return
true
;
}
static
bool
fromArray
(
const
cJSON
*
json
,
const
char
*
name
,
FFromJson
func
,
SArray
**
array
,
int32_t
itemSize
)
{
static
bool
addInlineArray
(
cJSON
*
json
,
const
char
*
name
,
FToJson
func
,
const
SArray
*
array
)
{
return
addTarray
(
json
,
name
,
func
,
array
,
false
);
}
static
bool
addArray
(
cJSON
*
json
,
const
char
*
name
,
FToJson
func
,
const
SArray
*
array
)
{
return
addTarray
(
json
,
name
,
func
,
array
,
true
);
}
static
bool
fromTarray
(
const
cJSON
*
json
,
const
char
*
name
,
FFromJson
func
,
SArray
**
array
,
int32_t
itemSize
,
bool
isPoint
)
{
const
cJSON
*
jArray
=
cJSON_GetObjectItem
(
json
,
name
);
int32_t
size
=
(
NULL
==
jArray
?
0
:
cJSON_GetArraySize
(
jArray
));
if
(
size
>
0
)
{
*
array
=
taosArrayInit
(
size
,
POINTER_BYTES
);
*
array
=
taosArrayInit
(
size
,
isPoint
?
POINTER_BYTES
:
itemSize
);
if
(
NULL
==
*
array
)
{
return
false
;
}
...
...
@@ -92,11 +100,19 @@ static bool fromArray(const cJSON* json, const char* name, FFromJson func, SArra
if
(
NULL
==
item
||
!
func
(
cJSON_GetArrayItem
(
jArray
,
i
),
item
))
{
return
false
;
}
taosArrayPush
(
*
array
,
&
item
);
taosArrayPush
(
*
array
,
isPoint
?
&
item
:
item
);
}
return
true
;
}
static
bool
fromInlineArray
(
const
cJSON
*
json
,
const
char
*
name
,
FFromJson
func
,
SArray
**
array
,
int32_t
itemSize
)
{
return
fromTarray
(
json
,
name
,
func
,
array
,
itemSize
,
false
);
}
static
bool
fromArray
(
const
cJSON
*
json
,
const
char
*
name
,
FFromJson
func
,
SArray
**
array
,
int32_t
itemSize
)
{
return
fromTarray
(
json
,
name
,
func
,
array
,
itemSize
,
true
);
}
static
bool
addRawArray
(
cJSON
*
json
,
const
char
*
name
,
FToJson
func
,
const
void
*
array
,
int32_t
itemSize
,
int32_t
size
)
{
if
(
size
>
0
)
{
cJSON
*
jArray
=
cJSON_AddArrayToObject
(
json
,
name
);
...
...
@@ -556,6 +572,32 @@ static bool epAddrFromJson(const cJSON* json, void* obj) {
return
true
;
}
static
const
char
*
jkNodeAddrId
=
"NodeId"
;
static
const
char
*
jkNodeAddrInUse
=
"InUse"
;
static
const
char
*
jkNodeAddrEpAddrs
=
"EpAddrs"
;
static
bool
nodeAddrToJson
(
const
void
*
obj
,
cJSON
*
json
)
{
const
SQueryNodeAddr
*
ep
=
(
const
SQueryNodeAddr
*
)
obj
;
bool
res
=
cJSON_AddNumberToObject
(
json
,
jkNodeAddrId
,
ep
->
nodeId
);
if
(
res
)
{
res
=
cJSON_AddNumberToObject
(
json
,
jkNodeAddrInUse
,
ep
->
inUse
);
}
if
(
res
)
{
res
=
addRawArray
(
json
,
jkNodeAddrEpAddrs
,
epAddrToJson
,
ep
->
epAddr
,
ep
->
numOfEps
,
sizeof
(
SEpAddrMsg
));
}
return
res
;
}
static
bool
nodeAddrFromJson
(
const
cJSON
*
json
,
void
*
obj
)
{
SQueryNodeAddr
*
ep
=
(
SQueryNodeAddr
*
)
obj
;
ep
->
nodeId
=
getNumber
(
json
,
jkNodeAddrId
);
ep
->
inUse
=
getNumber
(
json
,
jkNodeAddrInUse
);
int32_t
numOfEps
=
0
;
bool
res
=
fromRawArray
(
json
,
jkNodeAddrEpAddrs
,
nodeAddrFromJson
,
&
ep
->
epAddr
,
sizeof
(
SEpAddrMsg
),
&
numOfEps
);
ep
->
numOfEps
=
numOfEps
;
return
res
;
}
static
const
char
*
jkExchangeNodeSrcTemplateId
=
"SrcTemplateId"
;
static
const
char
*
jkExchangeNodeSrcEndPoints
=
"SrcEndPoints"
;
...
...
@@ -563,7 +605,7 @@ static bool exchangeNodeToJson(const void* obj, cJSON* json) {
const
SExchangePhyNode
*
exchange
=
(
const
SExchangePhyNode
*
)
obj
;
bool
res
=
cJSON_AddNumberToObject
(
json
,
jkExchangeNodeSrcTemplateId
,
exchange
->
srcTemplateId
);
if
(
res
)
{
res
=
add
Array
(
json
,
jkExchangeNodeSrcEndPoints
,
ep
AddrToJson
,
exchange
->
pSrcEndPoints
);
res
=
add
InlineArray
(
json
,
jkExchangeNodeSrcEndPoints
,
node
AddrToJson
,
exchange
->
pSrcEndPoints
);
}
return
res
;
}
...
...
@@ -571,7 +613,7 @@ static bool exchangeNodeToJson(const void* obj, cJSON* json) {
static
bool
exchangeNodeFromJson
(
const
cJSON
*
json
,
void
*
obj
)
{
SExchangePhyNode
*
exchange
=
(
SExchangePhyNode
*
)
obj
;
exchange
->
srcTemplateId
=
getNumber
(
json
,
jkExchangeNodeSrcTemplateId
);
return
from
Array
(
json
,
jkExchangeNodeSrcEndPoints
,
epAddrFromJson
,
&
exchange
->
pSrcEndPoints
,
sizeof
(
SEpAddrMsg
));
return
from
InlineArray
(
json
,
jkExchangeNodeSrcEndPoints
,
nodeAddrFromJson
,
&
exchange
->
pSrcEndPoints
,
sizeof
(
SQueryNodeAddr
));
}
static
bool
specificPhyNodeToJson
(
const
void
*
obj
,
cJSON
*
json
)
{
...
...
@@ -803,7 +845,6 @@ static cJSON* subplanToJson(const SSubplan* subplan) {
if
(
res
)
{
res
=
addObject
(
jSubplan
,
jkSubplanDataSink
,
dataSinkToJson
,
subplan
->
pDataSink
);
}
if
(
!
res
)
{
cJSON_Delete
(
jSubplan
);
return
NULL
;
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录