Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
cfbd7704
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看板
提交
cfbd7704
编写于
3月 16, 2022
作者:
C
Cary Xu
浏览文件
操作
浏览文件
下载
差异文件
Merge branch '3.0' into feature/TD-11463-3.0
上级
575ab144
b50ecc4e
变更
30
显示空白变更内容
内联
并排
Showing
30 changed file
with
2543 addition
and
1778 deletion
+2543
-1778
include/common/tmsg.h
include/common/tmsg.h
+71
-32
include/common/tmsgdef.h
include/common/tmsgdef.h
+1
-0
include/common/ttokendef.h
include/common/ttokendef.h
+48
-47
include/dnode/snode/snode.h
include/dnode/snode/snode.h
+2
-1
include/libs/nodes/cmdnodes.h
include/libs/nodes/cmdnodes.h
+31
-0
include/libs/nodes/nodes.h
include/libs/nodes/nodes.h
+5
-0
source/client/src/tmq.c
source/client/src/tmq.c
+2
-2
source/common/src/tmsg.c
source/common/src/tmsg.c
+68
-45
source/dnode/mgmt/impl/src/dndSnode.c
source/dnode/mgmt/impl/src/dndSnode.c
+20
-7
source/dnode/mnode/impl/inc/mndDef.h
source/dnode/mnode/impl/inc/mndDef.h
+3
-7
source/dnode/mnode/impl/inc/mndScheduler.h
source/dnode/mnode/impl/inc/mndScheduler.h
+2
-0
source/dnode/mnode/impl/src/mndScheduler.c
source/dnode/mnode/impl/src/mndScheduler.c
+47
-28
source/dnode/mnode/impl/src/mndStream.c
source/dnode/mnode/impl/src/mndStream.c
+7
-0
source/dnode/mnode/impl/src/mndTopic.c
source/dnode/mnode/impl/src/mndTopic.c
+38
-2
source/dnode/mnode/impl/test/topic/topic.cpp
source/dnode/mnode/impl/test/topic/topic.cpp
+1
-2
source/dnode/snode/CMakeLists.txt
source/dnode/snode/CMakeLists.txt
+2
-1
source/dnode/snode/inc/sndInt.h
source/dnode/snode/inc/sndInt.h
+8
-23
source/dnode/snode/src/snode.c
source/dnode/snode/src/snode.c
+63
-12
source/libs/nodes/src/nodesCodeFuncs.c
source/libs/nodes/src/nodesCodeFuncs.c
+5
-2
source/libs/nodes/src/nodesUtilFuncs.c
source/libs/nodes/src/nodesUtilFuncs.c
+10
-0
source/libs/parser/inc/parAst.h
source/libs/parser/inc/parAst.h
+5
-0
source/libs/parser/inc/sql.y
source/libs/parser/inc/sql.y
+22
-10
source/libs/parser/src/parAstCreater.c
source/libs/parser/src/parAstCreater.c
+49
-0
source/libs/parser/src/parTokenizer.c
source/libs/parser/src/parTokenizer.c
+1
-1
source/libs/parser/src/parTranslater.c
source/libs/parser/src/parTranslater.c
+167
-8
source/libs/parser/src/sql.c
source/libs/parser/src/sql.c
+1588
-1537
source/libs/parser/test/parserAstTest.cpp
source/libs/parser/test/parserAstTest.cpp
+57
-0
source/libs/transport/src/transSrv.c
source/libs/transport/src/transSrv.c
+3
-1
source/util/src/tuuid.c
source/util/src/tuuid.c
+10
-10
tests/script/tsim/query/interval.sim
tests/script/tsim/query/interval.sim
+207
-0
未找到文件。
include/common/tmsg.h
浏览文件 @
cfbd7704
...
...
@@ -24,6 +24,7 @@
#include "thash.h"
#include "tlist.h"
#include "trow.h"
#include "tuuid.h"
#ifdef __cplusplus
extern
"C"
{
...
...
@@ -553,15 +554,13 @@ int32_t tSerializeSQnodeListReq(void* buf, int32_t bufLen, SQnodeListReq* pReq);
int32_t
tDeserializeSQnodeListReq
(
void
*
buf
,
int32_t
bufLen
,
SQnodeListReq
*
pReq
);
typedef
struct
{
SArray
*
epSetList
;
// SArray<SEpSet>
SArray
*
epSetList
;
// SArray<SEpSet>
}
SQnodeListRsp
;
int32_t
tSerializeSQnodeListRsp
(
void
*
buf
,
int32_t
bufLen
,
SQnodeListRsp
*
pRsp
);
int32_t
tDeserializeSQnodeListRsp
(
void
*
buf
,
int32_t
bufLen
,
SQnodeListRsp
*
pRsp
);
void
tFreeSQnodeListRsp
(
SQnodeListRsp
*
pRsp
);
typedef
struct
{
SArray
*
pArray
;
// Array of SUseDbRsp
}
SUseDbBatchRsp
;
...
...
@@ -777,7 +776,6 @@ typedef struct SVgroupInfo {
int32_t
numOfTable
;
// unit is TSDB_TABLE_NUM_UNIT
}
SVgroupInfo
;
typedef
struct
{
int32_t
numOfVgroups
;
SVgroupInfo
vgroups
[];
...
...
@@ -1063,7 +1061,7 @@ typedef struct {
typedef
struct
{
int64_t
refId
;
SArray
*
taskStatus
;
//
SArray<STaskStatus>
SArray
*
taskStatus
;
//
SArray<STaskStatus>
}
SSchedulerStatusRsp
;
typedef
struct
{
...
...
@@ -1072,35 +1070,31 @@ typedef struct {
int8_t
action
;
}
STaskAction
;
typedef
struct
SQueryNodeEpId
{
int32_t
nodeId
;
// vgId or qnodeId
SEp
ep
;
}
SQueryNodeEpId
;
typedef
struct
{
SMsgHead
header
;
uint64_t
sId
;
SQueryNodeEpId
epId
;
SArray
*
taskAction
;
//
SArray<STaskAction>
SArray
*
taskAction
;
//
SArray<STaskAction>
}
SSchedulerHbReq
;
int32_t
tSerializeSSchedulerHbReq
(
void
*
buf
,
int32_t
bufLen
,
SSchedulerHbReq
*
pReq
);
int32_t
tDeserializeSSchedulerHbReq
(
void
*
buf
,
int32_t
bufLen
,
SSchedulerHbReq
*
pReq
);
void
tFreeSSchedulerHbReq
(
SSchedulerHbReq
*
pReq
);
int32_t
tSerializeSSchedulerHbReq
(
void
*
buf
,
int32_t
bufLen
,
SSchedulerHbReq
*
pReq
);
int32_t
tDeserializeSSchedulerHbReq
(
void
*
buf
,
int32_t
bufLen
,
SSchedulerHbReq
*
pReq
);
void
tFreeSSchedulerHbReq
(
SSchedulerHbReq
*
pReq
);
typedef
struct
{
uint64_t
seqId
;
SQueryNodeEpId
epId
;
SArray
*
taskStatus
;
//
SArray<STaskStatus>
SArray
*
taskStatus
;
//
SArray<STaskStatus>
}
SSchedulerHbRsp
;
int32_t
tSerializeSSchedulerHbRsp
(
void
*
buf
,
int32_t
bufLen
,
SSchedulerHbRsp
*
pRsp
);
int32_t
tDeserializeSSchedulerHbRsp
(
void
*
buf
,
int32_t
bufLen
,
SSchedulerHbRsp
*
pRsp
);
void
tFreeSSchedulerHbRsp
(
SSchedulerHbRsp
*
pRsp
);
int32_t
tSerializeSSchedulerHbRsp
(
void
*
buf
,
int32_t
bufLen
,
SSchedulerHbRsp
*
pRsp
);
int32_t
tDeserializeSSchedulerHbRsp
(
void
*
buf
,
int32_t
bufLen
,
SSchedulerHbRsp
*
pRsp
);
void
tFreeSSchedulerHbRsp
(
SSchedulerHbRsp
*
pRsp
);
typedef
struct
{
SMsgHead
header
;
...
...
@@ -1157,8 +1151,8 @@ typedef struct {
char
name
[
TSDB_TOPIC_FNAME_LEN
];
int8_t
igExists
;
char
*
sql
;
char
*
physicalPlan
;
char
*
logicalPlan
;
char
*
ast
;
char
subscribeDbName
[
TSDB_DB_NAME_LEN
]
;
}
SCMCreateTopicReq
;
int32_t
tSerializeSCMCreateTopicReq
(
void
*
buf
,
int32_t
bufLen
,
const
SCMCreateTopicReq
*
pReq
);
...
...
@@ -2252,6 +2246,51 @@ static FORCE_INLINE void* tDecodeSMqCMGetSubEpRsp(void* buf, SMqCMGetSubEpRsp* p
return
buf
;
}
enum
{
STREAM_TASK_STATUS__RUNNING
=
1
,
STREAM_TASK_STATUS__STOP
,
};
typedef
struct
{
int64_t
streamId
;
int32_t
taskId
;
int32_t
level
;
int8_t
status
;
char
*
qmsg
;
void
*
executor
;
// void* stateStore;
// storage handle
}
SStreamTask
;
static
FORCE_INLINE
SStreamTask
*
streamTaskNew
(
int64_t
streamId
,
int32_t
level
)
{
SStreamTask
*
pTask
=
(
SStreamTask
*
)
calloc
(
1
,
sizeof
(
SStreamTask
));
if
(
pTask
==
NULL
)
{
return
NULL
;
}
pTask
->
taskId
=
tGenIdPI32
();
pTask
->
status
=
STREAM_TASK_STATUS__RUNNING
;
pTask
->
qmsg
=
NULL
;
return
pTask
;
}
int32_t
tEncodeSStreamTask
(
SCoder
*
pEncoder
,
const
SStreamTask
*
pTask
);
int32_t
tDecodeSStreamTask
(
SCoder
*
pDecoder
,
SStreamTask
*
pTask
);
void
tFreeSStreamTask
(
SStreamTask
*
pTask
);
typedef
struct
{
SMsgHead
head
;
SStreamTask
*
task
;
}
SStreamTaskDeployReq
;
typedef
struct
{
int32_t
reserved
;
}
SStreamTaskDeployRsp
;
typedef
struct
{
SMsgHead
head
;
// TODO: other info needed by task
}
SStreamTaskExecReq
;
#pragma pack(pop)
#ifdef __cplusplus
...
...
include/common/tmsgdef.h
浏览文件 @
cfbd7704
...
...
@@ -199,6 +199,7 @@ enum {
// Requests handled by SNODE
TD_NEW_MSG_SEG
(
TDMT_SND_MSG
)
TD_DEF_MSG_TYPE
(
TDMT_SND_TASK_DEPLOY
,
"snode-task-deploy"
,
SStreamTaskDeployReq
,
SStreamTaskDeployRsp
)
#if defined(TD_MSG_NUMBER_)
TDMT_MAX
...
...
include/common/ttokendef.h
浏览文件 @
cfbd7704
...
...
@@ -108,53 +108,54 @@
#define TK_FULLTEXT 90
#define TK_FUNCTION 91
#define TK_INTERVAL 92
#define TK_MNODES 93
#define TK_NK_FLOAT 94
#define TK_NK_BOOL 95
#define TK_NK_VARIABLE 96
#define TK_BETWEEN 97
#define TK_IS 98
#define TK_NULL 99
#define TK_NK_LT 100
#define TK_NK_GT 101
#define TK_NK_LE 102
#define TK_NK_GE 103
#define TK_NK_NE 104
#define TK_NK_EQ 105
#define TK_LIKE 106
#define TK_MATCH 107
#define TK_NMATCH 108
#define TK_IN 109
#define TK_FROM 110
#define TK_AS 111
#define TK_JOIN 112
#define TK_INNER 113
#define TK_SELECT 114
#define TK_DISTINCT 115
#define TK_WHERE 116
#define TK_PARTITION 117
#define TK_BY 118
#define TK_SESSION 119
#define TK_STATE_WINDOW 120
#define TK_SLIDING 121
#define TK_FILL 122
#define TK_VALUE 123
#define TK_NONE 124
#define TK_PREV 125
#define TK_LINEAR 126
#define TK_NEXT 127
#define TK_GROUP 128
#define TK_HAVING 129
#define TK_ORDER 130
#define TK_SLIMIT 131
#define TK_SOFFSET 132
#define TK_LIMIT 133
#define TK_OFFSET 134
#define TK_ASC 135
#define TK_DESC 136
#define TK_NULLS 137
#define TK_FIRST 138
#define TK_LAST 139
#define TK_TOPIC 93
#define TK_AS 94
#define TK_MNODES 95
#define TK_NK_FLOAT 96
#define TK_NK_BOOL 97
#define TK_NK_VARIABLE 98
#define TK_BETWEEN 99
#define TK_IS 100
#define TK_NULL 101
#define TK_NK_LT 102
#define TK_NK_GT 103
#define TK_NK_LE 104
#define TK_NK_GE 105
#define TK_NK_NE 106
#define TK_NK_EQ 107
#define TK_LIKE 108
#define TK_MATCH 109
#define TK_NMATCH 110
#define TK_IN 111
#define TK_FROM 112
#define TK_JOIN 113
#define TK_INNER 114
#define TK_SELECT 115
#define TK_DISTINCT 116
#define TK_WHERE 117
#define TK_PARTITION 118
#define TK_BY 119
#define TK_SESSION 120
#define TK_STATE_WINDOW 121
#define TK_SLIDING 122
#define TK_FILL 123
#define TK_VALUE 124
#define TK_NONE 125
#define TK_PREV 126
#define TK_LINEAR 127
#define TK_NEXT 128
#define TK_GROUP 129
#define TK_HAVING 130
#define TK_ORDER 131
#define TK_SLIMIT 132
#define TK_SOFFSET 133
#define TK_LIMIT 134
#define TK_OFFSET 135
#define TK_ASC 136
#define TK_DESC 137
#define TK_NULLS 138
#define TK_FIRST 139
#define TK_LAST 140
#define TK_NK_SPACE 300
#define TK_NK_COMMENT 301
...
...
include/dnode/snode/snode.h
浏览文件 @
cfbd7704
...
...
@@ -16,6 +16,7 @@
#ifndef _TD_SNODE_H_
#define _TD_SNODE_H_
#include "tcommon.h"
#include "tmsg.h"
#include "trpc.h"
...
...
@@ -78,7 +79,7 @@ int32_t sndGetLoad(SSnode *pSnode, SSnodeLoad *pLoad);
* @param pRsp The response message
* @return int32_t 0 for success, -1 for failure
*/
int32_t
sndProcessMsg
(
SSnode
*
pSnode
,
SRpcMsg
*
pMsg
,
SRpcMsg
**
pRsp
);
//
int32_t sndProcessMsg(SSnode *pSnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
int32_t
sndProcessUMsg
(
SSnode
*
pSnode
,
SRpcMsg
*
pMsg
);
...
...
include/libs/nodes/cmdnodes.h
浏览文件 @
cfbd7704
...
...
@@ -61,6 +61,12 @@ typedef struct SDropDatabaseStmt {
bool
ignoreNotExists
;
}
SDropDatabaseStmt
;
typedef
struct
SAlterDatabaseStmt
{
ENodeType
type
;
char
dbName
[
TSDB_DB_NAME_LEN
];
SDatabaseOptions
*
pOptions
;
}
SAlterDatabaseStmt
;
typedef
struct
STableOptions
{
ENodeType
type
;
int32_t
keep
;
...
...
@@ -179,11 +185,36 @@ typedef struct SCreateIndexStmt {
SIndexOptions
*
pOptions
;
}
SCreateIndexStmt
;
typedef
struct
SDropIndexStmt
{
ENodeType
type
;
char
indexName
[
TSDB_INDEX_NAME_LEN
];
char
tableName
[
TSDB_TABLE_NAME_LEN
];
}
SDropIndexStmt
;
typedef
struct
SCreateQnodeStmt
{
ENodeType
type
;
int32_t
dnodeId
;
}
SCreateQnodeStmt
;
typedef
struct
SDropQnodeStmt
{
ENodeType
type
;
int32_t
dnodeId
;
}
SDropQnodeStmt
;
typedef
struct
SCreateTopicStmt
{
ENodeType
type
;
char
topicName
[
TSDB_TABLE_NAME_LEN
];
char
subscribeDbName
[
TSDB_DB_NAME_LEN
];
bool
ignoreExists
;
SNode
*
pQuery
;
}
SCreateTopicStmt
;
typedef
struct
SDropTopicStmt
{
ENodeType
type
;
char
topicName
[
TSDB_TABLE_NAME_LEN
];
bool
ignoreNotExists
;
}
SDropTopicStmt
;
#ifdef __cplusplus
}
#endif
...
...
include/libs/nodes/nodes.h
浏览文件 @
cfbd7704
...
...
@@ -77,6 +77,7 @@ typedef enum ENodeType {
QUERY_NODE_VNODE_MODIF_STMT
,
QUERY_NODE_CREATE_DATABASE_STMT
,
QUERY_NODE_DROP_DATABASE_STMT
,
QUERY_NODE_ALTER_DATABASE_STMT
,
QUERY_NODE_SHOW_DATABASES_STMT
,
// temp
QUERY_NODE_CREATE_TABLE_STMT
,
QUERY_NODE_CREATE_SUBTABLE_CLAUSE
,
...
...
@@ -98,7 +99,11 @@ typedef enum ENodeType {
QUERY_NODE_SHOW_MNODES_STMT
,
QUERY_NODE_SHOW_QNODES_STMT
,
QUERY_NODE_CREATE_INDEX_STMT
,
QUERY_NODE_DROP_INDEX_STMT
,
QUERY_NODE_CREATE_QNODE_STMT
,
QUERY_NODE_DROP_QNODE_STMT
,
QUERY_NODE_CREATE_TOPIC_STMT
,
QUERY_NODE_DROP_TOPIC_STMT
,
// logic plan node
QUERY_NODE_LOGIC_PLAN_SCAN
,
...
...
source/client/src/tmq.c
浏览文件 @
cfbd7704
...
...
@@ -482,7 +482,7 @@ TAOS_RES* tmq_create_topic(TAOS* taos, const char* topicName, const char* sql, i
}
tscDebug
(
"start to create topic, %s"
,
topicName
);
#if 0
CHECK_CODE_GOTO(buildRequest(pTscObj, sql, sqlLen, &pRequest), _return);
CHECK_CODE_GOTO(parseSql(pRequest, &pQueryNode), _return);
...
...
@@ -536,7 +536,7 @@ TAOS_RES* tmq_create_topic(TAOS* taos, const char* topicName, const char* sql, i
asyncSendMsgToServer(pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, sendInfo);
tsem_wait(&pRequest->body.rspSem);
#endif
_return:
qDestroyQuery
(
pQueryNode
);
/*if (sendInfo != NULL) {*/
...
...
source/common/src/tmsg.c
浏览文件 @
cfbd7704
...
...
@@ -1467,8 +1467,7 @@ int32_t tDeserializeSUseDbReq(void *buf, int32_t bufLen, SUseDbReq *pReq) {
return
0
;
}
int32_t
tSerializeSQnodeListReq
(
void
*
buf
,
int32_t
bufLen
,
SQnodeListReq
*
pReq
)
{
int32_t
tSerializeSQnodeListReq
(
void
*
buf
,
int32_t
bufLen
,
SQnodeListReq
*
pReq
)
{
SCoder
encoder
=
{
0
};
tCoderInit
(
&
encoder
,
TD_LITTLE_ENDIAN
,
buf
,
bufLen
,
TD_ENCODER
);
...
...
@@ -1990,11 +1989,9 @@ int32_t tDeserializeSMDropTopicReq(void *buf, int32_t bufLen, SMDropTopicReq *pR
int32_t
tSerializeSCMCreateTopicReq
(
void
*
buf
,
int32_t
bufLen
,
const
SCMCreateTopicReq
*
pReq
)
{
int32_t
sqlLen
=
0
;
int32_t
physicalPlanLen
=
0
;
int32_t
logicalPlanLen
=
0
;
int32_t
astLen
=
0
;
if
(
pReq
->
sql
!=
NULL
)
sqlLen
=
(
int32_t
)
strlen
(
pReq
->
sql
);
if
(
pReq
->
physicalPlan
!=
NULL
)
physicalPlanLen
=
(
int32_t
)
strlen
(
pReq
->
physicalPlan
);
if
(
pReq
->
logicalPlan
!=
NULL
)
logicalPlanLen
=
(
int32_t
)
strlen
(
pReq
->
logicalPlan
);
if
(
pReq
->
ast
!=
NULL
)
astLen
=
(
int32_t
)
strlen
(
pReq
->
ast
);
SCoder
encoder
=
{
0
};
tCoderInit
(
&
encoder
,
TD_LITTLE_ENDIAN
,
buf
,
bufLen
,
TD_ENCODER
);
...
...
@@ -2003,11 +2000,9 @@ int32_t tSerializeSCMCreateTopicReq(void *buf, int32_t bufLen, const SCMCreateTo
if
(
tEncodeCStr
(
&
encoder
,
pReq
->
name
)
<
0
)
return
-
1
;
if
(
tEncodeI8
(
&
encoder
,
pReq
->
igExists
)
<
0
)
return
-
1
;
if
(
tEncodeI32
(
&
encoder
,
sqlLen
)
<
0
)
return
-
1
;
if
(
tEncodeI32
(
&
encoder
,
physicalPlanLen
)
<
0
)
return
-
1
;
if
(
tEncodeI32
(
&
encoder
,
logicalPlanLen
)
<
0
)
return
-
1
;
if
(
tEncodeCStr
(
&
encoder
,
pReq
->
sql
)
<
0
)
return
-
1
;
if
(
tEncodeCStr
(
&
encoder
,
pReq
->
physicalPlan
)
<
0
)
return
-
1
;
if
(
tEncodeCStr
(
&
encoder
,
pReq
->
logicalPlan
)
<
0
)
return
-
1
;
if
(
tEncodeI32
(
&
encoder
,
astLen
)
<
0
)
return
-
1
;
if
(
sqlLen
>
0
&&
tEncodeCStr
(
&
encoder
,
pReq
->
sql
)
<
0
)
return
-
1
;
if
(
astLen
>
0
&&
tEncodeCStr
(
&
encoder
,
pReq
->
ast
)
<
0
)
return
-
1
;
tEndEncode
(
&
encoder
);
...
...
@@ -2018,8 +2013,7 @@ int32_t tSerializeSCMCreateTopicReq(void *buf, int32_t bufLen, const SCMCreateTo
int32_t
tDeserializeSCMCreateTopicReq
(
void
*
buf
,
int32_t
bufLen
,
SCMCreateTopicReq
*
pReq
)
{
int32_t
sqlLen
=
0
;
int32_t
physicalPlanLen
=
0
;
int32_t
logicalPlanLen
=
0
;
int32_t
astLen
=
0
;
SCoder
decoder
=
{
0
};
tCoderInit
(
&
decoder
,
TD_LITTLE_ENDIAN
,
buf
,
bufLen
,
TD_DECODER
);
...
...
@@ -2028,17 +2022,20 @@ int32_t tDeserializeSCMCreateTopicReq(void *buf, int32_t bufLen, SCMCreateTopicR
if
(
tDecodeCStrTo
(
&
decoder
,
pReq
->
name
)
<
0
)
return
-
1
;
if
(
tDecodeI8
(
&
decoder
,
&
pReq
->
igExists
)
<
0
)
return
-
1
;
if
(
tDecodeI32
(
&
decoder
,
&
sqlLen
)
<
0
)
return
-
1
;
if
(
tDecodeI32
(
&
decoder
,
&
physicalPlanLen
)
<
0
)
return
-
1
;
if
(
tDecodeI32
(
&
decoder
,
&
logicalPlanLen
)
<
0
)
return
-
1
;
if
(
tDecodeI32
(
&
decoder
,
&
astLen
)
<
0
)
return
-
1
;
if
(
sqlLen
>
0
)
{
pReq
->
sql
=
calloc
(
1
,
sqlLen
+
1
);
pReq
->
physicalPlan
=
calloc
(
1
,
physicalPlanLen
+
1
);
pReq
->
logicalPlan
=
calloc
(
1
,
logicalPlanLen
+
1
);
if
(
pReq
->
sql
==
NULL
||
pReq
->
physicalPlan
==
NULL
||
pReq
->
logicalPlan
==
NULL
)
return
-
1
;
if
(
pReq
->
sql
==
NULL
)
return
-
1
;
if
(
tDecodeCStrTo
(
&
decoder
,
pReq
->
sql
)
<
0
)
return
-
1
;
if
(
tDecodeCStrTo
(
&
decoder
,
pReq
->
physicalPlan
)
<
0
)
return
-
1
;
if
(
tDecodeCStrTo
(
&
decoder
,
pReq
->
logicalPlan
)
<
0
)
return
-
1
;
}
if
(
astLen
>
0
)
{
pReq
->
ast
=
calloc
(
1
,
astLen
+
1
);
if
(
pReq
->
ast
==
NULL
)
return
-
1
;
if
(
tDecodeCStrTo
(
&
decoder
,
pReq
->
ast
)
<
0
)
return
-
1
;
}
tEndDecode
(
&
decoder
);
tCoderClear
(
&
decoder
);
...
...
@@ -2047,8 +2044,7 @@ int32_t tDeserializeSCMCreateTopicReq(void *buf, int32_t bufLen, SCMCreateTopicR
void
tFreeSCMCreateTopicReq
(
SCMCreateTopicReq
*
pReq
)
{
tfree
(
pReq
->
sql
);
tfree
(
pReq
->
physicalPlan
);
tfree
(
pReq
->
logicalPlan
);
tfree
(
pReq
->
ast
);
}
int32_t
tSerializeSCMCreateTopicRsp
(
void
*
buf
,
int32_t
bufLen
,
const
SCMCreateTopicRsp
*
pRsp
)
{
...
...
@@ -2559,8 +2555,6 @@ int32_t tDeserializeSSchedulerHbReq(void *buf, int32_t bufLen, SSchedulerHbReq *
void
tFreeSSchedulerHbReq
(
SSchedulerHbReq
*
pReq
)
{
taosArrayDestroy
(
pReq
->
taskAction
);
}
int32_t
tSerializeSSchedulerHbRsp
(
void
*
buf
,
int32_t
bufLen
,
SSchedulerHbRsp
*
pRsp
)
{
SCoder
encoder
=
{
0
};
tCoderInit
(
&
encoder
,
TD_LITTLE_ENDIAN
,
buf
,
bufLen
,
TD_ENCODER
);
...
...
@@ -2694,3 +2688,32 @@ void tFreeSCMCreateStreamReq(SCMCreateStreamReq *pReq) {
tfree
(
pReq
->
physicalPlan
);
tfree
(
pReq
->
logicalPlan
);
}
int32_t
tEncodeSStreamTask
(
SCoder
*
pEncoder
,
const
SStreamTask
*
pTask
)
{
if
(
tStartEncode
(
pEncoder
)
<
0
)
return
-
1
;
if
(
tEncodeI64
(
pEncoder
,
pTask
->
streamId
)
<
0
)
return
-
1
;
if
(
tEncodeI32
(
pEncoder
,
pTask
->
taskId
)
<
0
)
return
-
1
;
if
(
tEncodeI32
(
pEncoder
,
pTask
->
level
)
<
0
)
return
-
1
;
if
(
tEncodeI8
(
pEncoder
,
pTask
->
status
)
<
0
)
return
-
1
;
if
(
tEncodeCStr
(
pEncoder
,
pTask
->
qmsg
)
<
0
)
return
-
1
;
tEndEncode
(
pEncoder
);
return
pEncoder
->
pos
;
}
int32_t
tDecodeSStreamTask
(
SCoder
*
pDecoder
,
SStreamTask
*
pTask
)
{
if
(
tStartDecode
(
pDecoder
)
<
0
)
return
-
1
;
if
(
tDecodeI64
(
pDecoder
,
&
pTask
->
streamId
)
<
0
)
return
-
1
;
if
(
tDecodeI32
(
pDecoder
,
&
pTask
->
taskId
)
<
0
)
return
-
1
;
if
(
tDecodeI32
(
pDecoder
,
&
pTask
->
level
)
<
0
)
return
-
1
;
if
(
tDecodeI8
(
pDecoder
,
&
pTask
->
status
)
<
0
)
return
-
1
;
if
(
tDecodeCStr
(
pDecoder
,
(
const
char
**
)
&
pTask
->
qmsg
)
<
0
)
return
-
1
;
tEndDecode
(
pDecoder
);
return
0
;
}
void
tFreeSStreamTask
(
SStreamTask
*
pTask
)
{
// TODO
/*free(pTask->qmsg);*/
/*free(pTask->executor);*/
/*free(pTask);*/
}
source/dnode/mgmt/impl/src/dndSnode.c
浏览文件 @
cfbd7704
...
...
@@ -323,7 +323,7 @@ int32_t dndProcessDropSnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
}
static
void
dndProcessSnodeUniqueQueue
(
SDnode
*
pDnode
,
STaosQall
*
qall
,
int32_t
numOfMsgs
)
{
SSnodeMgmt
*
pMgmt
=
&
pDnode
->
smgmt
;
/*SSnodeMgmt *pMgmt = &pDnode->smgmt;*/
int32_t
code
=
TSDB_CODE_DND_SNODE_NOT_DEPLOYED
;
SSnode
*
pSnode
=
dndAcquireSnode
(
pDnode
);
...
...
@@ -337,19 +337,32 @@ static void dndProcessSnodeUniqueQueue(SDnode *pDnode, STaosQall *qall, int32_t
rpcFreeCont
(
pMsg
->
pCont
);
taosFreeQitem
(
pMsg
);
}
}
dndReleaseSnode
(
pDnode
,
pSnode
);
}
else
{
for
(
int32_t
i
=
0
;
i
<
numOfMsgs
;
i
++
)
{
SRpcMsg
*
pMsg
=
NULL
;
taosGetQitem
(
qall
,
(
void
**
)
&
pMsg
);
SRpcMsg
rpcRsp
=
{.
handle
=
pMsg
->
handle
,
.
ahandle
=
pMsg
->
ahandle
,
.
code
=
code
};
rpcSendResponse
(
&
rpcRsp
);
rpcFreeCont
(
pMsg
->
pCont
);
taosFreeQitem
(
pMsg
);
}
}
}
static
void
dndProcessSnodeSharedQueue
(
SDnode
*
pDnode
,
SRpcMsg
*
pMsg
)
{
SSnodeMgmt
*
pMgmt
=
&
pDnode
->
smgmt
;
/*SSnodeMgmt *pMgmt = &pDnode->smgmt;*/
int32_t
code
=
TSDB_CODE_DND_SNODE_NOT_DEPLOYED
;
SSnode
*
pSnode
=
dndAcquireSnode
(
pDnode
);
if
(
pSnode
!=
NULL
)
{
code
=
sndProcessSMsg
(
pSnode
,
pMsg
);
}
sndProcessSMsg
(
pSnode
,
pMsg
);
dndReleaseSnode
(
pDnode
,
pSnode
);
}
else
{
SRpcMsg
rpcRsp
=
{.
handle
=
pMsg
->
handle
,
.
ahandle
=
pMsg
->
ahandle
,
.
code
=
code
};
rpcSendResponse
(
&
rpcRsp
);
}
#if 0
if (pMsg->msgType & 1u) {
...
...
source/dnode/mnode/impl/inc/mndDef.h
浏览文件 @
cfbd7704
...
...
@@ -85,6 +85,8 @@ typedef enum {
TRN_TYPE_REBALANCE
=
1017
,
TRN_TYPE_COMMIT_OFFSET
=
1018
,
TRN_TYPE_CREATE_STREAM
=
1019
,
TRN_TYPE_DROP_STREAM
=
1020
,
TRN_TYPE_ALTER_STREAM
=
1021
,
TRN_TYPE_BASIC_SCOPE_END
,
TRN_TYPE_GLOBAL_SCOPE
=
2000
,
TRN_TYPE_CREATE_DNODE
=
2001
,
...
...
@@ -679,12 +681,6 @@ static FORCE_INLINE void* tDecodeSMqConsumerObj(void* buf, SMqConsumerObj* pCons
return
buf
;
}
typedef
struct
{
int32_t
taskId
;
int32_t
level
;
SSubplan
*
plan
;
}
SStreamTaskMeta
;
typedef
struct
{
char
name
[
TSDB_TOPIC_FNAME_LEN
];
char
db
[
TSDB_DB_FNAME_LEN
];
...
...
@@ -700,7 +696,7 @@ typedef struct {
char
*
sql
;
char
*
logicalPlan
;
char
*
physicalPlan
;
SArray
*
tasks
;
// SArray<SArray<SStreamTask
Meta
>>
SArray
*
tasks
;
// SArray<SArray<SStreamTask>>
}
SStreamObj
;
int32_t
tEncodeSStreamObj
(
SCoder
*
pEncoder
,
const
SStreamObj
*
pObj
);
...
...
source/dnode/mnode/impl/inc/mndScheduler.h
浏览文件 @
cfbd7704
...
...
@@ -27,6 +27,8 @@ void mndCleanupScheduler(SMnode* pMnode);
int32_t
mndSchedInitSubEp
(
SMnode
*
pMnode
,
const
SMqTopicObj
*
pTopic
,
SMqSubscribeObj
*
pSub
);
int32_t
mndScheduleStream
(
SMnode
*
pMnode
,
STrans
*
pTrans
,
SStreamObj
*
pStream
);
#ifdef __cplusplus
}
#endif
...
...
source/dnode/mnode/impl/src/mndScheduler.c
浏览文件 @
cfbd7704
...
...
@@ -31,7 +31,7 @@
#include "tname.h"
#include "tuuid.h"
int32_t
mndScheduleStream
(
SMnode
*
pMnode
,
SStreamObj
*
pStream
)
{
int32_t
mndScheduleStream
(
SMnode
*
pMnode
,
S
Trans
*
pTrans
,
S
StreamObj
*
pStream
)
{
SSdb
*
pSdb
=
pMnode
->
pSdb
;
SVgObj
*
pVgroup
=
NULL
;
SQueryPlan
*
pPlan
=
qStringToQueryPlan
(
pStream
->
physicalPlan
);
...
...
@@ -41,17 +41,18 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
}
ASSERT
(
pStream
->
vgNum
==
0
);
int32_t
levelNum
=
LIST_LENGTH
(
pPlan
->
pSubplans
);
pStream
->
tasks
=
taosArrayInit
(
levelNum
,
sizeof
(
SArray
));
int32_t
totLevel
=
LIST_LENGTH
(
pPlan
->
pSubplans
);
pStream
->
tasks
=
taosArrayInit
(
totLevel
,
sizeof
(
SArray
));
for
(
int32_t
i
=
0
;
i
<
levelNum
;
i
++
)
{
SArray
*
taskOneLevel
=
taosArrayInit
(
0
,
sizeof
(
SStreamTaskMeta
));
SNodeListNode
*
inner
=
nodesListGetNode
(
pPlan
->
pSubplans
,
i
);
int32_t
msgLen
;
for
(
int32_t
level
=
0
;
level
<
totLevel
;
level
++
)
{
SArray
*
taskOneLevel
=
taosArrayInit
(
0
,
sizeof
(
SStreamTask
));
SNodeListNode
*
inner
=
nodesListGetNode
(
pPlan
->
pSubplans
,
level
);
int32_t
opNum
=
LIST_LENGTH
(
inner
->
pNodeList
);
ASSERT
(
opNum
==
1
);
SSubplan
*
plan
=
nodesListGetNode
(
inner
->
pNodeList
,
0
);
if
(
i
==
0
)
{
SSubplan
*
plan
=
nodesListGetNode
(
inner
->
pNodeList
,
level
);
if
(
level
==
0
)
{
ASSERT
(
plan
->
type
==
SUBPLAN_TYPE_SCAN
);
void
*
pIter
=
NULL
;
while
(
1
)
{
...
...
@@ -63,15 +64,19 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
}
pStream
->
vgNum
++
;
// send to vnode
SStreamTask
*
pTask
=
streamTaskNew
(
pStream
->
uid
,
level
);
plan
->
execNode
.
nodeId
=
pVgroup
->
vgId
;
plan
->
execNode
.
epSet
=
mndGetVgroupEpset
(
pMnode
,
pVgroup
);
SStreamTaskMeta
task
=
{
.
taskId
=
tGenIdPI32
(),
.
level
=
i
,
.
plan
=
plan
,
}
;
// send to vnode
taosArrayPush
(
taskOneLevel
,
&
t
ask
);
if
(
qSubPlanToString
(
plan
,
&
pTask
->
qmsg
,
&
msgLen
)
<
0
)
{
sdbRelease
(
pSdb
,
pVgroup
);
qDestroyQueryPlan
(
pPlan
);
terrno
=
TSDB_CODE_QRY_INVALID_INPUT
;
return
-
1
;
}
taosArrayPush
(
taskOneLevel
,
pT
ask
);
}
}
else
if
(
plan
->
subplanType
==
SUBPLAN_TYPE_SCAN
)
{
// duplicatable
...
...
@@ -82,22 +87,36 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
// if has snode, set to shared thread num in snode
parallel
=
SND_SHARED_THREAD_NUM
;
for
(
int32_t
j
=
0
;
j
<
parallel
;
j
++
)
{
SStreamTaskMeta
task
=
{
.
taskId
=
tGenIdPI32
(),
.
level
=
i
,
.
plan
=
plan
,
};
taosArrayPush
(
taskOneLevel
,
&
task
);
for
(
int32_t
i
=
0
;
i
<
parallel
;
i
++
)
{
SStreamTask
*
pTask
=
streamTaskNew
(
pStream
->
uid
,
level
);
// TODO:get snode id and ep
plan
->
execNode
.
nodeId
=
pVgroup
->
vgId
;
plan
->
execNode
.
epSet
=
mndGetVgroupEpset
(
pMnode
,
pVgroup
);
if
(
qSubPlanToString
(
plan
,
&
pTask
->
qmsg
,
&
msgLen
)
<
0
)
{
qDestroyQueryPlan
(
pPlan
);
terrno
=
TSDB_CODE_QRY_INVALID_INPUT
;
return
-
1
;
}
taosArrayPush
(
taskOneLevel
,
pTask
);
}
}
else
{
// not duplicatable
SStreamTaskMeta
task
=
{
.
taskId
=
tGenIdPI32
(),
.
level
=
i
,
.
plan
=
plan
,
};
taosArrayPush
(
taskOneLevel
,
&
task
);
SStreamTask
*
pTask
=
streamTaskNew
(
pStream
->
uid
,
level
);
// TODO:get snode id and ep
plan
->
execNode
.
nodeId
=
pVgroup
->
vgId
;
plan
->
execNode
.
epSet
=
mndGetVgroupEpset
(
pMnode
,
pVgroup
);
if
(
qSubPlanToString
(
plan
,
&
pTask
->
qmsg
,
&
msgLen
)
<
0
)
{
sdbRelease
(
pSdb
,
pVgroup
);
qDestroyQueryPlan
(
pPlan
);
terrno
=
TSDB_CODE_QRY_INVALID_INPUT
;
return
-
1
;
}
taosArrayPush
(
taskOneLevel
,
pTask
);
}
taosArrayPush
(
pStream
->
tasks
,
taskOneLevel
);
}
...
...
source/dnode/mnode/impl/src/mndStream.c
浏览文件 @
cfbd7704
...
...
@@ -18,6 +18,7 @@
#include "mndDb.h"
#include "mndDnode.h"
#include "mndMnode.h"
#include "mndScheduler.h"
#include "mndShow.h"
#include "mndStb.h"
#include "mndTrans.h"
...
...
@@ -237,6 +238,12 @@ static int32_t mndCreateStream(SMnode *pMnode, SMnodeMsg *pReq, SCMCreateStreamR
}
sdbSetRawStatus
(
pRedoRaw
,
SDB_STATUS_READY
);
if
(
mndScheduleStream
(
pMnode
,
pTrans
,
&
streamObj
)
<
0
)
{
mError
(
"stream:%ld, schedule stream since %s"
,
streamObj
.
uid
,
terrstr
());
mndTransDrop
(
pTrans
);
return
-
1
;
}
if
(
mndTransPrepare
(
pMnode
,
pTrans
)
!=
0
)
{
mError
(
"trans:%d, failed to prepare since %s"
,
pTrans
->
id
,
terrstr
());
mndTransDrop
(
pTrans
);
...
...
source/dnode/mnode/impl/src/mndTopic.c
浏览文件 @
cfbd7704
...
...
@@ -236,6 +236,29 @@ static int32_t mndCheckCreateTopicReq(SCMCreateTopicReq *pCreate) {
return
0
;
}
static
int32_t
mndGetPlanString
(
SCMCreateTopicReq
*
pCreate
,
char
**
pStr
)
{
if
(
NULL
==
pCreate
->
ast
)
{
return
TSDB_CODE_SUCCESS
;
}
SNode
*
pAst
=
NULL
;
int32_t
code
=
nodesStringToNode
(
pCreate
->
ast
,
&
pAst
);
SQueryPlan
*
pPlan
=
NULL
;
if
(
TSDB_CODE_SUCCESS
==
code
)
{
SPlanContext
cxt
=
{
.
pAstRoot
=
pAst
,
.
streamQuery
=
true
};
code
=
qCreateQueryPlan
(
&
cxt
,
&
pPlan
,
NULL
);
}
if
(
TSDB_CODE_SUCCESS
==
code
)
{
code
=
nodesNodeToString
(
pPlan
,
false
,
pStr
,
NULL
);
}
nodesDestroyNode
(
pAst
);
nodesDestroyNode
(
pPlan
);
terrno
=
code
;
return
code
;
}
static
int32_t
mndCreateTopic
(
SMnode
*
pMnode
,
SMnodeMsg
*
pReq
,
SCMCreateTopicReq
*
pCreate
,
SDbObj
*
pDb
)
{
mDebug
(
"topic:%s to create"
,
pCreate
->
name
);
SMqTopicObj
topicObj
=
{
0
};
...
...
@@ -247,13 +270,23 @@ static int32_t mndCreateTopic(SMnode *pMnode, SMnodeMsg *pReq, SCMCreateTopicReq
topicObj
.
dbUid
=
pDb
->
uid
;
topicObj
.
version
=
1
;
topicObj
.
sql
=
pCreate
->
sql
;
topicObj
.
physicalPlan
=
pCreate
->
physicalPlan
;
topicObj
.
logicalPlan
=
pCreate
->
logicalPlan
;
topicObj
.
physicalPlan
=
""
;
topicObj
.
logicalPlan
=
""
;
topicObj
.
sqlLen
=
strlen
(
pCreate
->
sql
);
char
*
pPlanStr
=
NULL
;
if
(
TSDB_CODE_SUCCESS
!=
mndGetPlanString
(
pCreate
,
&
pPlanStr
))
{
mError
(
"topic:%s, failed to get plan since %s"
,
pCreate
->
name
,
terrstr
());
return
-
1
;
}
if
(
NULL
!=
pPlanStr
)
{
topicObj
.
physicalPlan
=
pPlanStr
;
}
STrans
*
pTrans
=
mndTransCreate
(
pMnode
,
TRN_POLICY_ROLLBACK
,
TRN_TYPE_CREATE_TOPIC
,
&
pReq
->
rpcMsg
);
if
(
pTrans
==
NULL
)
{
mError
(
"topic:%s, failed to create since %s"
,
pCreate
->
name
,
terrstr
());
tfree
(
pPlanStr
);
return
-
1
;
}
mDebug
(
"trans:%d, used to create topic:%s"
,
pTrans
->
id
,
pCreate
->
name
);
...
...
@@ -261,6 +294,7 @@ static int32_t mndCreateTopic(SMnode *pMnode, SMnodeMsg *pReq, SCMCreateTopicReq
SSdbRaw
*
pRedoRaw
=
mndTopicActionEncode
(
&
topicObj
);
if
(
pRedoRaw
==
NULL
||
mndTransAppendRedolog
(
pTrans
,
pRedoRaw
)
!=
0
)
{
mError
(
"trans:%d, failed to append redo log since %s"
,
pTrans
->
id
,
terrstr
());
tfree
(
pPlanStr
);
mndTransDrop
(
pTrans
);
return
-
1
;
}
...
...
@@ -268,10 +302,12 @@ static int32_t mndCreateTopic(SMnode *pMnode, SMnodeMsg *pReq, SCMCreateTopicReq
if
(
mndTransPrepare
(
pMnode
,
pTrans
)
!=
0
)
{
mError
(
"trans:%d, failed to prepare since %s"
,
pTrans
->
id
,
terrstr
());
tfree
(
pPlanStr
);
mndTransDrop
(
pTrans
);
return
-
1
;
}
tfree
(
pPlanStr
);
mndTransDrop
(
pTrans
);
return
0
;
}
...
...
source/dnode/mnode/impl/test/topic/topic.cpp
浏览文件 @
cfbd7704
...
...
@@ -65,8 +65,7 @@ void* MndTestTopic::BuildCreateTopicReq(const char* topicName, const char* sql,
strcpy
(
createReq
.
name
,
topicName
);
createReq
.
igExists
=
0
;
createReq
.
sql
=
(
char
*
)
sql
;
createReq
.
physicalPlan
=
(
char
*
)
"physicalPlan"
;
createReq
.
logicalPlan
=
(
char
*
)
"logicalPlan"
;
createReq
.
ast
=
NULL
;
int32_t
contLen
=
tSerializeSCMCreateTopicReq
(
NULL
,
0
,
&
createReq
);
void
*
pReq
=
rpcMallocCont
(
contLen
);
...
...
source/dnode/snode/CMakeLists.txt
浏览文件 @
cfbd7704
...
...
@@ -7,6 +7,7 @@ target_include_directories(
)
target_link_libraries
(
snode
PRIVATE executor
PRIVATE transport
PRIVATE os
PRIVATE common
...
...
source/dnode/snode/inc/sndInt.h
浏览文件 @
cfbd7704
...
...
@@ -38,13 +38,8 @@ enum {
STREAM_STATUS__DELETING
,
};
enum
{
STREAM_TASK_STATUS__RUNNING
=
1
,
STREAM_TASK_STATUS__STOP
,
};
typedef
struct
{
SHashObj
*
pHash
;
// taskId ->
s
treamTask
SHashObj
*
pHash
;
// taskId ->
SS
treamTask
}
SStreamMeta
;
typedef
struct
SSnode
{
...
...
@@ -52,26 +47,16 @@ typedef struct SSnode {
SSnodeOpt
cfg
;
}
SSnode
;
typedef
struct
{
int64_t
streamId
;
int32_t
taskId
;
int32_t
IdxInLevel
;
int32_t
level
;
}
SStreamTaskInfo
;
SStreamMeta
*
sndMetaNew
();
void
sndMetaDelete
(
SStreamMeta
*
pMeta
);
typedef
struct
{
SStreamTaskInfo
meta
;
int8_t
status
;
void
*
executor
;
void
*
stateStore
;
// storage handle
}
SStreamTask
;
int32_t
sndMetaDeployTask
(
SStreamMeta
*
pMeta
,
SStreamTask
*
pTask
);
int32_t
sndMetaRemoveTask
(
SStreamMeta
*
pMeta
,
int32_t
taskId
);
int32_t
sndCreateTask
();
int32_t
sndDropTaskOfStream
(
int64_t
streamId
);
int32_t
sndDropTaskOfStream
(
SStreamMeta
*
pMeta
,
int64_t
streamId
);
int32_t
sndStopTaskOfStream
(
int64_t
streamId
);
int32_t
sndResumeTaskOfStream
(
int64_t
streamId
);
int32_t
sndStopTaskOfStream
(
SStreamMeta
*
pMeta
,
int64_t
streamId
);
int32_t
sndResumeTaskOfStream
(
SStreamMeta
*
pMeta
,
int64_t
streamId
);
#ifdef __cplusplus
}
...
...
source/dnode/snode/src/snode.c
浏览文件 @
cfbd7704
...
...
@@ -13,40 +13,91 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "executor.h"
#include "sndInt.h"
#include "tuuid.h"
SSnode
*
sndOpen
(
const
char
*
path
,
const
SSnodeOpt
*
pOption
)
{
SSnode
*
pSnode
=
calloc
(
1
,
sizeof
(
SSnode
));
if
(
pSnode
==
NULL
)
{
return
NULL
;
}
memcpy
(
&
pSnode
->
cfg
,
pOption
,
sizeof
(
SSnodeOpt
));
pSnode
->
pMeta
=
sndMetaNew
();
if
(
pSnode
->
pMeta
==
NULL
)
{
free
(
pSnode
);
return
NULL
;
}
return
pSnode
;
}
void
sndClose
(
SSnode
*
pSnode
)
{
free
(
pSnode
);
}
void
sndClose
(
SSnode
*
pSnode
)
{
sndMetaDelete
(
pSnode
->
pMeta
);
free
(
pSnode
);
}
int32_t
sndGetLoad
(
SSnode
*
pSnode
,
SSnodeLoad
*
pLoad
)
{
return
0
;
}
int32_t
sndProcessMsg
(
SSnode
*
pSnode
,
SRpcMsg
*
pMsg
,
SRpcMsg
**
pRsp
)
{
*
pRsp
=
NULL
;
return
0
;
}
/*int32_t sndProcessMsg(SSnode *pSnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {*/
/**pRsp = NULL;*/
/*return 0;*/
/*}*/
void
sndDestroy
(
const
char
*
path
)
{}
static
int32_t
sndDeployTask
(
SSnode
*
pSnode
,
SRpcMsg
*
pMsg
)
{
SStreamTask
*
task
=
malloc
(
sizeof
(
SStreamTask
));
if
(
task
==
NULL
)
{
SStreamMeta
*
sndMetaNew
()
{
SStreamMeta
*
pMeta
=
calloc
(
1
,
sizeof
(
SStreamMeta
));
if
(
pMeta
==
NULL
)
{
return
NULL
;
}
pMeta
->
pHash
=
taosHashInit
(
64
,
taosGetDefaultHashFunction
(
TSDB_DATA_TYPE_INT
),
true
,
HASH_NO_LOCK
);
if
(
pMeta
->
pHash
==
NULL
)
{
free
(
pMeta
);
return
NULL
;
}
return
pMeta
;
}
void
sndMetaDelete
(
SStreamMeta
*
pMeta
)
{
taosHashCleanup
(
pMeta
->
pHash
);
free
(
pMeta
);
}
int32_t
sndMetaDeployTask
(
SStreamMeta
*
pMeta
,
SStreamTask
*
pTask
)
{
pTask
->
executor
=
qCreateStreamExecTaskInfo
(
pTask
->
qmsg
,
NULL
);
return
taosHashPut
(
pMeta
->
pHash
,
&
pTask
->
taskId
,
sizeof
(
int32_t
),
pTask
,
sizeof
(
void
*
));
}
int32_t
sndMetaRemoveTask
(
SStreamMeta
*
pMeta
,
int32_t
taskId
)
{
SStreamTask
*
pTask
=
taosHashGet
(
pMeta
->
pHash
,
&
taskId
,
sizeof
(
int32_t
));
if
(
pTask
==
NULL
)
{
return
-
1
;
}
task
->
meta
.
taskId
=
tGenIdPI32
();
taosHashPut
(
pSnode
->
pMeta
->
pHash
,
&
task
->
meta
.
taskId
,
sizeof
(
int32_t
),
&
task
,
sizeof
(
void
*
));
return
0
;
free
(
pTask
->
qmsg
);
// TODO:free executor
free
(
pTask
);
return
taosHashRemove
(
pMeta
->
pHash
,
&
taskId
,
sizeof
(
int32_t
));
}
int32_t
sndProcessUMsg
(
SSnode
*
pSnode
,
SRpcMsg
*
pMsg
)
{
// stream deploy
ment
// stream deploy
// stream stop/resume
// operator exec
if
(
pMsg
->
msgType
==
TDMT_SND_TASK_DEPLOY
)
{
void
*
msg
=
POINTER_SHIFT
(
pMsg
->
pCont
,
sizeof
(
SMsgHead
));
SStreamTask
*
pTask
=
malloc
(
sizeof
(
SStreamTask
));
if
(
pTask
==
NULL
)
{
return
-
1
;
}
SCoder
decoder
;
tCoderInit
(
&
decoder
,
TD_LITTLE_ENDIAN
,
msg
,
pMsg
->
contLen
-
sizeof
(
SMsgHead
),
TD_DECODER
);
tDecodeSStreamTask
(
&
decoder
,
pTask
);
tCoderClear
(
&
decoder
);
sndMetaDeployTask
(
pSnode
->
pMeta
,
pTask
);
}
else
{
//
}
return
0
;
}
...
...
source/libs/nodes/src/nodesCodeFuncs.c
浏览文件 @
cfbd7704
...
...
@@ -1730,7 +1730,7 @@ static int32_t jsonToNodeObject(const SJson* pJson, const char* pName, SNode** p
}
int32_t
nodesNodeToString
(
const
SNodeptr
pNode
,
bool
format
,
char
**
pStr
,
int32_t
*
pLen
)
{
if
(
NULL
==
pNode
||
NULL
==
pStr
||
NULL
==
pLen
)
{
if
(
NULL
==
pNode
||
NULL
==
pStr
)
{
terrno
=
TSDB_CODE_FAILED
;
return
TSDB_CODE_FAILED
;
}
...
...
@@ -1750,7 +1750,10 @@ int32_t nodesNodeToString(const SNodeptr pNode, bool format, char** pStr, int32_
*
pStr
=
format
?
tjsonToString
(
pJson
)
:
tjsonToUnformattedString
(
pJson
);
tjsonDelete
(
pJson
);
if
(
NULL
!=
pLen
)
{
*
pLen
=
strlen
(
*
pStr
)
+
1
;
}
return
TSDB_CODE_SUCCESS
;
}
...
...
source/libs/nodes/src/nodesUtilFuncs.c
浏览文件 @
cfbd7704
...
...
@@ -92,6 +92,8 @@ SNodeptr nodesMakeNode(ENodeType type) {
return
makeNode
(
type
,
sizeof
(
SCreateDatabaseStmt
));
case
QUERY_NODE_DROP_DATABASE_STMT
:
return
makeNode
(
type
,
sizeof
(
SDropDatabaseStmt
));
case
QUERY_NODE_ALTER_DATABASE_STMT
:
return
makeNode
(
type
,
sizeof
(
SAlterDatabaseStmt
));
case
QUERY_NODE_SHOW_DATABASES_STMT
:
return
makeNode
(
type
,
sizeof
(
SShowStmt
));
case
QUERY_NODE_CREATE_TABLE_STMT
:
...
...
@@ -131,8 +133,16 @@ SNodeptr nodesMakeNode(ENodeType type) {
return
makeNode
(
type
,
sizeof
(
SShowStmt
));
case
QUERY_NODE_CREATE_INDEX_STMT
:
return
makeNode
(
type
,
sizeof
(
SCreateIndexStmt
));
case
QUERY_NODE_DROP_INDEX_STMT
:
return
makeNode
(
type
,
sizeof
(
SDropIndexStmt
));
case
QUERY_NODE_CREATE_QNODE_STMT
:
return
makeNode
(
type
,
sizeof
(
SCreateQnodeStmt
));
case
QUERY_NODE_DROP_QNODE_STMT
:
return
makeNode
(
type
,
sizeof
(
SDropQnodeStmt
));
case
QUERY_NODE_CREATE_TOPIC_STMT
:
return
makeNode
(
type
,
sizeof
(
SCreateTopicStmt
));
case
QUERY_NODE_DROP_TOPIC_STMT
:
return
makeNode
(
type
,
sizeof
(
SDropTopicStmt
));
case
QUERY_NODE_LOGIC_PLAN_SCAN
:
return
makeNode
(
type
,
sizeof
(
SScanLogicNode
));
case
QUERY_NODE_LOGIC_PLAN_JOIN
:
...
...
source/libs/parser/inc/parAst.h
浏览文件 @
cfbd7704
...
...
@@ -113,6 +113,7 @@ SNode* createDefaultDatabaseOptions(SAstCreateContext* pCxt);
SNode
*
setDatabaseOption
(
SAstCreateContext
*
pCxt
,
SNode
*
pOptions
,
EDatabaseOptionType
type
,
const
SToken
*
pVal
);
SNode
*
createCreateDatabaseStmt
(
SAstCreateContext
*
pCxt
,
bool
ignoreExists
,
const
SToken
*
pDbName
,
SNode
*
pOptions
);
SNode
*
createDropDatabaseStmt
(
SAstCreateContext
*
pCxt
,
bool
ignoreNotExists
,
const
SToken
*
pDbName
);
SNode
*
createAlterDatabaseStmt
(
SAstCreateContext
*
pCxt
,
const
SToken
*
pDbName
,
SNode
*
pOptions
);
SNode
*
createDefaultTableOptions
(
SAstCreateContext
*
pCxt
);
SNode
*
setTableOption
(
SAstCreateContext
*
pCxt
,
SNode
*
pOptions
,
ETableOptionType
type
,
const
SToken
*
pVal
);
SNode
*
setTableSmaOption
(
SAstCreateContext
*
pCxt
,
SNode
*
pOptions
,
SNodeList
*
pSma
);
...
...
@@ -134,7 +135,11 @@ SNode* createCreateDnodeStmt(SAstCreateContext* pCxt, const SToken* pFqdn, const
SNode
*
createDropDnodeStmt
(
SAstCreateContext
*
pCxt
,
const
SToken
*
pDnode
);
SNode
*
createCreateIndexStmt
(
SAstCreateContext
*
pCxt
,
EIndexType
type
,
const
SToken
*
pIndexName
,
const
SToken
*
pTableName
,
SNodeList
*
pCols
,
SNode
*
pOptions
);
SNode
*
createIndexOption
(
SAstCreateContext
*
pCxt
,
SNodeList
*
pFuncs
,
SNode
*
pInterval
,
SNode
*
pOffset
,
SNode
*
pSliding
);
SNode
*
createDropIndexStmt
(
SAstCreateContext
*
pCxt
,
const
SToken
*
pIndexName
,
const
SToken
*
pTableName
);
SNode
*
createCreateQnodeStmt
(
SAstCreateContext
*
pCxt
,
const
SToken
*
pDnodeId
);
SNode
*
createDropQnodeStmt
(
SAstCreateContext
*
pCxt
,
const
SToken
*
pDnodeId
);
SNode
*
createCreateTopicStmt
(
SAstCreateContext
*
pCxt
,
bool
ignoreExists
,
const
SToken
*
pTopicName
,
SNode
*
pQuery
,
const
SToken
*
pSubscribeDbName
);
SNode
*
createDropTopicStmt
(
SAstCreateContext
*
pCxt
,
bool
ignoreNotExists
,
const
SToken
*
pTopicName
);
#ifdef __cplusplus
}
...
...
source/libs/parser/inc/sql.y
浏览文件 @
cfbd7704
...
...
@@ -42,17 +42,17 @@
//%right NK_BITNOT.
/************************************************ create/alter/drop/show user *****************************************/
cmd ::= CREATE USER user_name(A) PASS NK_STRING(B). { pCxt->pRootNode = createCreateUserStmt(pCxt, &A, &B);}
cmd ::= ALTER USER user_name(A) PASS NK_STRING(B). { pCxt->pRootNode = createAlterUserStmt(pCxt, &A, TSDB_ALTER_USER_PASSWD, &B);}
cmd ::= ALTER USER user_name(A) PRIVILEGE NK_STRING(B). { pCxt->pRootNode = createAlterUserStmt(pCxt, &A, TSDB_ALTER_USER_PRIVILEGES, &B);}
cmd ::= CREATE USER user_name(A) PASS NK_STRING(B). { pCxt->pRootNode = createCreateUserStmt(pCxt, &A, &B);
}
cmd ::= ALTER USER user_name(A) PASS NK_STRING(B). { pCxt->pRootNode = createAlterUserStmt(pCxt, &A, TSDB_ALTER_USER_PASSWD, &B);
}
cmd ::= ALTER USER user_name(A) PRIVILEGE NK_STRING(B). { pCxt->pRootNode = createAlterUserStmt(pCxt, &A, TSDB_ALTER_USER_PRIVILEGES, &B);
}
cmd ::= DROP USER user_name(A). { pCxt->pRootNode = createDropUserStmt(pCxt, &A); }
cmd ::= SHOW USERS. { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_USERS_STMT, NULL); }
/************************************************ create/drop/show dnode **********************************************/
cmd ::= CREATE DNODE dnode_endpoint(A). { pCxt->pRootNode = createCreateDnodeStmt(pCxt, &A, NULL);}
cmd ::= CREATE DNODE dnode_host_name(A) PORT NK_INTEGER(B). { pCxt->pRootNode = createCreateDnodeStmt(pCxt, &A, &B);}
cmd ::= DROP DNODE NK_INTEGER(A). { pCxt->pRootNode = createDropDnodeStmt(pCxt, &A);}
cmd ::= DROP DNODE dnode_endpoint(A). { pCxt->pRootNode = createDropDnodeStmt(pCxt, &A);}
cmd ::= CREATE DNODE dnode_endpoint(A). { pCxt->pRootNode = createCreateDnodeStmt(pCxt, &A, NULL);
}
cmd ::= CREATE DNODE dnode_host_name(A) PORT NK_INTEGER(B). { pCxt->pRootNode = createCreateDnodeStmt(pCxt, &A, &B);
}
cmd ::= DROP DNODE NK_INTEGER(A). { pCxt->pRootNode = createDropDnodeStmt(pCxt, &A);
}
cmd ::= DROP DNODE dnode_endpoint(A). { pCxt->pRootNode = createDropDnodeStmt(pCxt, &A);
}
cmd ::= SHOW DNODES. { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_DNODES_STMT, NULL); }
%type dnode_endpoint { SToken }
...
...
@@ -64,15 +64,17 @@ dnode_endpoint(A) ::= NK_STRING(B).
dnode_host_name(A) ::= NK_ID(B). { A = B; }
dnode_host_name(A) ::= NK_IPTOKEN(B). { A = B; }
/************************************************ create
qnode *****
***************************************************/
/************************************************ create
/drop qnode
***************************************************/
cmd ::= CREATE QNODE ON DNODE NK_INTEGER(A). { pCxt->pRootNode = createCreateQnodeStmt(pCxt, &A); }
cmd ::= DROP QNODE ON DNODE NK_INTEGER(A). { pCxt->pRootNode = createDropQnodeStmt(pCxt, &A); }
cmd ::= SHOW QNODES. { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_QNODES_STMT, NULL); }
/************************************************ create/drop/show/use database ***************************************/
cmd ::= CREATE DATABASE not_exists_opt(A) db_name(B) db_options(C). { pCxt->pRootNode = createCreateDatabaseStmt(pCxt, A, &B, C);}
cmd ::= CREATE DATABASE not_exists_opt(A) db_name(B) db_options(C). { pCxt->pRootNode = createCreateDatabaseStmt(pCxt, A, &B, C);
}
cmd ::= DROP DATABASE exists_opt(A) db_name(B). { pCxt->pRootNode = createDropDatabaseStmt(pCxt, A, &B); }
cmd ::= SHOW DATABASES. { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_DATABASES_STMT, NULL); }
cmd ::= USE db_name(A). { pCxt->pRootNode = createUseDatabaseStmt(pCxt, &A);}
cmd ::= USE db_name(A). { pCxt->pRootNode = createUseDatabaseStmt(pCxt, &A); }
cmd ::= ALTER DATABASE db_name(A) db_options(B). { pCxt->pRootNode = createAlterDatabaseStmt(pCxt, &A, B); }
%type not_exists_opt { bool }
%destructor not_exists_opt { }
...
...
@@ -198,6 +200,7 @@ col_name(A) ::= column_name(B).
cmd ::= CREATE SMA INDEX index_name(A) ON table_name(B) index_options(C). { pCxt->pRootNode = createCreateIndexStmt(pCxt, INDEX_TYPE_SMA, &A, &B, NULL, C); }
cmd ::= CREATE FULLTEXT INDEX
index_name(A) ON table_name(B) NK_LP col_name_list(C) NK_RP. { pCxt->pRootNode = createCreateIndexStmt(pCxt, INDEX_TYPE_FULLTEXT, &A, &B, C, NULL); }
cmd ::= DROP INDEX index_name(A) ON table_name(B). { pCxt->pRootNode = createDropIndexStmt(pCxt, &A, &B); }
index_options(A) ::= . { A = NULL; }
index_options(A) ::= FUNCTION NK_LP func_list(B) NK_RP INTERVAL
...
...
@@ -212,6 +215,11 @@ func_list(A) ::= func_list(B) NK_COMMA func(C).
func(A) ::= function_name(B) NK_LP expression_list(C) NK_RP. { A = createFunctionNode(pCxt, &B, C); }
/************************************************ create/drop topic ***************************************************/
cmd ::= CREATE TOPIC not_exists_opt(A) topic_name(B) AS query_expression(C). { pCxt->pRootNode = createCreateTopicStmt(pCxt, A, &B, C, NULL); }
cmd ::= CREATE TOPIC not_exists_opt(A) topic_name(B) AS db_name(C). { pCxt->pRootNode = createCreateTopicStmt(pCxt, A, &B, NULL, &C); }
cmd ::= DROP TOPIC exists_opt(A) topic_name(B). { pCxt->pRootNode = createDropTopicStmt(pCxt, A, &B); }
/************************************************ show vgroups ********************************************************/
cmd ::= SHOW VGROUPS. { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_VGROUPS_STMT, NULL); }
cmd ::= SHOW db_name(B) NK_DOT VGROUPS. { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_VGROUPS_STMT, &B); }
...
...
@@ -270,6 +278,10 @@ user_name(A) ::= NK_ID(B).
%destructor index_name { }
index_name(A) ::= NK_ID(B). { A = B; }
%type topic_name { SToken }
%destructor topic_name { }
topic_name(A) ::= NK_ID(B). { A = B; }
/************************************************ expression **********************************************************/
expression(A) ::= literal(B). { A = B; }
//expression(A) ::= NK_QUESTION(B). { A = B; }
...
...
source/libs/parser/src/parAstCreater.c
浏览文件 @
cfbd7704
...
...
@@ -799,6 +799,17 @@ SNode* createDropDatabaseStmt(SAstCreateContext* pCxt, bool ignoreNotExists, con
return
(
SNode
*
)
pStmt
;
}
SNode
*
createAlterDatabaseStmt
(
SAstCreateContext
*
pCxt
,
const
SToken
*
pDbName
,
SNode
*
pOptions
)
{
if
(
!
checkDbName
(
pCxt
,
pDbName
))
{
return
NULL
;
}
SAlterDatabaseStmt
*
pStmt
=
nodesMakeNode
(
QUERY_NODE_ALTER_DATABASE_STMT
);
CHECK_OUT_OF_MEM
(
pStmt
);
strncpy
(
pStmt
->
dbName
,
pDbName
->
z
,
pDbName
->
n
);
pStmt
->
pOptions
=
(
SDatabaseOptions
*
)
pOptions
;
return
(
SNode
*
)
pStmt
;
}
SNode
*
createDefaultTableOptions
(
SAstCreateContext
*
pCxt
)
{
STableOptions
*
pOptions
=
nodesMakeNode
(
QUERY_NODE_TABLE_OPTIONS
);
CHECK_OUT_OF_MEM
(
pOptions
);
...
...
@@ -1022,9 +1033,47 @@ SNode* createIndexOption(SAstCreateContext* pCxt, SNodeList* pFuncs, SNode* pInt
return
(
SNode
*
)
pOptions
;
}
SNode
*
createDropIndexStmt
(
SAstCreateContext
*
pCxt
,
const
SToken
*
pIndexName
,
const
SToken
*
pTableName
)
{
if
(
!
checkIndexName
(
pCxt
,
pIndexName
)
||
!
checkTableName
(
pCxt
,
pTableName
))
{
return
NULL
;
}
SDropIndexStmt
*
pStmt
=
nodesMakeNode
(
QUERY_NODE_DROP_INDEX_STMT
);
CHECK_OUT_OF_MEM
(
pStmt
);
strncpy
(
pStmt
->
indexName
,
pIndexName
->
z
,
pIndexName
->
n
);
strncpy
(
pStmt
->
tableName
,
pTableName
->
z
,
pTableName
->
n
);
return
(
SNode
*
)
pStmt
;
}
SNode
*
createCreateQnodeStmt
(
SAstCreateContext
*
pCxt
,
const
SToken
*
pDnodeId
)
{
SCreateQnodeStmt
*
pStmt
=
nodesMakeNode
(
QUERY_NODE_CREATE_QNODE_STMT
);
CHECK_OUT_OF_MEM
(
pStmt
);
pStmt
->
dnodeId
=
strtol
(
pDnodeId
->
z
,
NULL
,
10
);;
return
(
SNode
*
)
pStmt
;
}
SNode
*
createDropQnodeStmt
(
SAstCreateContext
*
pCxt
,
const
SToken
*
pDnodeId
)
{
SDropQnodeStmt
*
pStmt
=
nodesMakeNode
(
QUERY_NODE_DROP_QNODE_STMT
);
CHECK_OUT_OF_MEM
(
pStmt
);
pStmt
->
dnodeId
=
strtol
(
pDnodeId
->
z
,
NULL
,
10
);;
return
(
SNode
*
)
pStmt
;
}
SNode
*
createCreateTopicStmt
(
SAstCreateContext
*
pCxt
,
bool
ignoreExists
,
const
SToken
*
pTopicName
,
SNode
*
pQuery
,
const
SToken
*
pSubscribeDbName
)
{
SCreateTopicStmt
*
pStmt
=
nodesMakeNode
(
QUERY_NODE_CREATE_TOPIC_STMT
);
CHECK_OUT_OF_MEM
(
pStmt
);
strncpy
(
pStmt
->
topicName
,
pTopicName
->
z
,
pTopicName
->
n
);
pStmt
->
ignoreExists
=
ignoreExists
;
pStmt
->
pQuery
=
pQuery
;
if
(
NULL
!=
pSubscribeDbName
)
{
strncpy
(
pStmt
->
subscribeDbName
,
pSubscribeDbName
->
z
,
pSubscribeDbName
->
n
);
}
return
(
SNode
*
)
pStmt
;
}
SNode
*
createDropTopicStmt
(
SAstCreateContext
*
pCxt
,
bool
ignoreNotExists
,
const
SToken
*
pTopicName
)
{
SDropTopicStmt
*
pStmt
=
nodesMakeNode
(
QUERY_NODE_DROP_TOPIC_STMT
);
CHECK_OUT_OF_MEM
(
pStmt
);
strncpy
(
pStmt
->
topicName
,
pTopicName
->
z
,
pTopicName
->
n
);
pStmt
->
ignoreNotExists
=
ignoreNotExists
;
return
(
SNode
*
)
pStmt
;
}
source/libs/parser/src/parTokenizer.c
浏览文件 @
cfbd7704
...
...
@@ -121,6 +121,7 @@ static SKeyword keywordTable[] = {
{
"TAGS"
,
TK_TAGS
},
{
"TIMESTAMP"
,
TK_TIMESTAMP
},
{
"TINYINT"
,
TK_TINYINT
},
{
"TOPIC"
,
TK_TOPIC
},
{
"TTL"
,
TK_TTL
},
{
"UNION"
,
TK_UNION
},
{
"UNSIGNED"
,
TK_UNSIGNED
},
...
...
@@ -230,7 +231,6 @@ static SKeyword keywordTable[] = {
// {"TBNAME", TK_TBNAME},
// {"VNODES", TK_VNODES},
// {"PARTITIONS", TK_PARTITIONS},
// {"TOPIC", TK_TOPIC},
// {"TOPICS", TK_TOPICS},
// {"COMPACT", TK_COMPACT},
// {"MODIFY", TK_MODIFY},
...
...
source/libs/parser/src/parTranslater.c
浏览文件 @
cfbd7704
...
...
@@ -21,14 +21,6 @@
#include "parUtil.h"
#include "ttime.h"
static
bool
afterGroupBy
(
ESqlClause
clause
)
{
return
clause
>
SQL_CLAUSE_GROUP_BY
;
}
static
bool
beforeHaving
(
ESqlClause
clause
)
{
return
clause
<
SQL_CLAUSE_HAVING
;
}
typedef
struct
STranslateContext
{
SParseContext
*
pParseCxt
;
int32_t
errCode
;
...
...
@@ -41,6 +33,15 @@ typedef struct STranslateContext {
}
STranslateContext
;
static
int32_t
translateSubquery
(
STranslateContext
*
pCxt
,
SNode
*
pNode
);
static
int32_t
translateQuery
(
STranslateContext
*
pCxt
,
SNode
*
pNode
);
static
bool
afterGroupBy
(
ESqlClause
clause
)
{
return
clause
>
SQL_CLAUSE_GROUP_BY
;
}
static
bool
beforeHaving
(
ESqlClause
clause
)
{
return
clause
<
SQL_CLAUSE_HAVING
;
}
static
EDealRes
generateDealNodeErrMsg
(
STranslateContext
*
pCxt
,
int32_t
errCode
,
...)
{
va_list
vArgList
;
...
...
@@ -833,6 +834,41 @@ static int32_t translateDropDatabase(STranslateContext* pCxt, SDropDatabaseStmt*
return
TSDB_CODE_SUCCESS
;
}
static
void
buildAlterDbReq
(
STranslateContext
*
pCxt
,
SAlterDatabaseStmt
*
pStmt
,
SAlterDbReq
*
pReq
)
{
SName
name
=
{
0
};
tNameSetDbName
(
&
name
,
pCxt
->
pParseCxt
->
acctId
,
pStmt
->
dbName
,
strlen
(
pStmt
->
dbName
));
tNameGetFullDbName
(
&
name
,
pReq
->
db
);
pReq
->
totalBlocks
=
pStmt
->
pOptions
->
numOfBlocks
;
pReq
->
daysToKeep0
=
pStmt
->
pOptions
->
keep
;
pReq
->
daysToKeep1
=
-
1
;
pReq
->
daysToKeep2
=
-
1
;
pReq
->
fsyncPeriod
=
pStmt
->
pOptions
->
fsyncPeriod
;
pReq
->
walLevel
=
pStmt
->
pOptions
->
walLevel
;
pReq
->
quorum
=
pStmt
->
pOptions
->
quorum
;
pReq
->
cacheLastRow
=
pStmt
->
pOptions
->
cachelast
;
return
;
}
static
int32_t
translateAlterDatabase
(
STranslateContext
*
pCxt
,
SAlterDatabaseStmt
*
pStmt
)
{
SAlterDbReq
alterReq
=
{
0
};
buildAlterDbReq
(
pCxt
,
pStmt
,
&
alterReq
);
pCxt
->
pCmdMsg
=
malloc
(
sizeof
(
SCmdMsgInfo
));
if
(
NULL
==
pCxt
->
pCmdMsg
)
{
return
TSDB_CODE_OUT_OF_MEMORY
;
}
pCxt
->
pCmdMsg
->
epSet
=
pCxt
->
pParseCxt
->
mgmtEpSet
;
pCxt
->
pCmdMsg
->
msgType
=
TDMT_MND_ALTER_DB
;
pCxt
->
pCmdMsg
->
msgLen
=
tSerializeSAlterDbReq
(
NULL
,
0
,
&
alterReq
);
pCxt
->
pCmdMsg
->
pMsg
=
malloc
(
pCxt
->
pCmdMsg
->
msgLen
);
if
(
NULL
==
pCxt
->
pCmdMsg
->
pMsg
)
{
return
TSDB_CODE_OUT_OF_MEMORY
;
}
tSerializeSAlterDbReq
(
pCxt
->
pCmdMsg
->
pMsg
,
pCxt
->
pCmdMsg
->
msgLen
,
&
alterReq
);
return
TSDB_CODE_SUCCESS
;
}
static
int32_t
columnNodeToField
(
SNodeList
*
pList
,
SArray
**
pArray
)
{
*
pArray
=
taosArrayInit
(
LIST_LENGTH
(
pList
),
sizeof
(
SField
));
SNode
*
pNode
;
...
...
@@ -1201,6 +1237,27 @@ static int32_t translateCreateIndex(STranslateContext* pCxt, SCreateIndexStmt* p
}
}
static
int32_t
translateDropIndex
(
STranslateContext
*
pCxt
,
SDropIndexStmt
*
pStmt
)
{
SVDropTSmaReq
dropSmaReq
=
{
0
};
strcpy
(
dropSmaReq
.
indexName
,
pStmt
->
indexName
);
pCxt
->
pCmdMsg
=
malloc
(
sizeof
(
SCmdMsgInfo
));
if
(
NULL
==
pCxt
->
pCmdMsg
)
{
return
TSDB_CODE_OUT_OF_MEMORY
;
}
pCxt
->
pCmdMsg
->
epSet
=
pCxt
->
pParseCxt
->
mgmtEpSet
;
pCxt
->
pCmdMsg
->
msgType
=
TDMT_VND_DROP_SMA
;
pCxt
->
pCmdMsg
->
msgLen
=
tSerializeSVDropTSmaReq
(
NULL
,
&
dropSmaReq
);
pCxt
->
pCmdMsg
->
pMsg
=
malloc
(
pCxt
->
pCmdMsg
->
msgLen
);
if
(
NULL
==
pCxt
->
pCmdMsg
->
pMsg
)
{
return
TSDB_CODE_OUT_OF_MEMORY
;
}
void
*
pBuf
=
pCxt
->
pCmdMsg
->
pMsg
;
tSerializeSVDropTSmaReq
(
&
pBuf
,
&
dropSmaReq
);
return
TSDB_CODE_SUCCESS
;
}
static
int32_t
translateCreateQnode
(
STranslateContext
*
pCxt
,
SCreateQnodeStmt
*
pStmt
)
{
SMCreateQnodeReq
createReq
=
{
.
dnodeId
=
pStmt
->
dnodeId
};
...
...
@@ -1220,6 +1277,93 @@ static int32_t translateCreateQnode(STranslateContext* pCxt, SCreateQnodeStmt* p
return
TSDB_CODE_SUCCESS
;
}
static
int32_t
translateDropQnode
(
STranslateContext
*
pCxt
,
SDropQnodeStmt
*
pStmt
)
{
SDDropQnodeReq
dropReq
=
{
.
dnodeId
=
pStmt
->
dnodeId
};
pCxt
->
pCmdMsg
=
malloc
(
sizeof
(
SCmdMsgInfo
));
if
(
NULL
==
pCxt
->
pCmdMsg
)
{
return
TSDB_CODE_OUT_OF_MEMORY
;
}
pCxt
->
pCmdMsg
->
epSet
=
pCxt
->
pParseCxt
->
mgmtEpSet
;
pCxt
->
pCmdMsg
->
msgType
=
TDMT_DND_DROP_QNODE
;
pCxt
->
pCmdMsg
->
msgLen
=
tSerializeSMCreateDropQSBNodeReq
(
NULL
,
0
,
&
dropReq
);
pCxt
->
pCmdMsg
->
pMsg
=
malloc
(
pCxt
->
pCmdMsg
->
msgLen
);
if
(
NULL
==
pCxt
->
pCmdMsg
->
pMsg
)
{
return
TSDB_CODE_OUT_OF_MEMORY
;
}
tSerializeSMCreateDropQSBNodeReq
(
pCxt
->
pCmdMsg
->
pMsg
,
pCxt
->
pCmdMsg
->
msgLen
,
&
dropReq
);
return
TSDB_CODE_SUCCESS
;
}
static
int32_t
translateCreateTopic
(
STranslateContext
*
pCxt
,
SCreateTopicStmt
*
pStmt
)
{
SCMCreateTopicReq
createReq
=
{
0
};
if
(
NULL
!=
pStmt
->
pQuery
)
{
int32_t
code
=
translateQuery
(
pCxt
,
pStmt
->
pQuery
);
if
(
TSDB_CODE_SUCCESS
==
code
)
{
code
=
nodesNodeToString
(
pStmt
->
pQuery
,
false
,
&
createReq
.
ast
,
NULL
);
}
if
(
TSDB_CODE_SUCCESS
!=
code
)
{
return
code
;
}
}
else
{
strcpy
(
createReq
.
subscribeDbName
,
pStmt
->
subscribeDbName
);
}
createReq
.
sql
=
strdup
(
pCxt
->
pParseCxt
->
pSql
);
if
(
NULL
==
createReq
.
sql
)
{
return
TSDB_CODE_OUT_OF_MEMORY
;
}
SName
name
=
{
.
type
=
TSDB_TABLE_NAME_T
,
.
acctId
=
pCxt
->
pParseCxt
->
acctId
};
strcpy
(
name
.
dbname
,
pCxt
->
pParseCxt
->
db
);
strcpy
(
name
.
tname
,
pStmt
->
topicName
);
tNameExtractFullName
(
&
name
,
createReq
.
name
);
createReq
.
igExists
=
pStmt
->
ignoreExists
;
pCxt
->
pCmdMsg
=
malloc
(
sizeof
(
SCmdMsgInfo
));
if
(
NULL
==
pCxt
->
pCmdMsg
)
{
return
TSDB_CODE_OUT_OF_MEMORY
;
}
pCxt
->
pCmdMsg
->
epSet
=
pCxt
->
pParseCxt
->
mgmtEpSet
;
pCxt
->
pCmdMsg
->
msgType
=
TDMT_MND_CREATE_TOPIC
;
pCxt
->
pCmdMsg
->
msgLen
=
tSerializeSCMCreateTopicReq
(
NULL
,
0
,
&
createReq
);
pCxt
->
pCmdMsg
->
pMsg
=
malloc
(
pCxt
->
pCmdMsg
->
msgLen
);
if
(
NULL
==
pCxt
->
pCmdMsg
->
pMsg
)
{
return
TSDB_CODE_OUT_OF_MEMORY
;
}
tSerializeSCMCreateTopicReq
(
pCxt
->
pCmdMsg
->
pMsg
,
pCxt
->
pCmdMsg
->
msgLen
,
&
createReq
);
tFreeSCMCreateTopicReq
(
&
createReq
);
return
TSDB_CODE_SUCCESS
;
}
static
int32_t
translateDropTopic
(
STranslateContext
*
pCxt
,
SDropTopicStmt
*
pStmt
)
{
SMDropTopicReq
dropReq
=
{
0
};
SName
name
=
{
.
type
=
TSDB_TABLE_NAME_T
,
.
acctId
=
pCxt
->
pParseCxt
->
acctId
};
strcpy
(
name
.
dbname
,
pCxt
->
pParseCxt
->
db
);
strcpy
(
name
.
tname
,
pStmt
->
topicName
);
tNameExtractFullName
(
&
name
,
dropReq
.
name
);
dropReq
.
igNotExists
=
pStmt
->
ignoreNotExists
;
pCxt
->
pCmdMsg
=
malloc
(
sizeof
(
SCmdMsgInfo
));
if
(
NULL
==
pCxt
->
pCmdMsg
)
{
return
TSDB_CODE_OUT_OF_MEMORY
;
}
pCxt
->
pCmdMsg
->
epSet
=
pCxt
->
pParseCxt
->
mgmtEpSet
;
pCxt
->
pCmdMsg
->
msgType
=
TDMT_MND_DROP_TOPIC
;
pCxt
->
pCmdMsg
->
msgLen
=
tSerializeSMDropTopicReq
(
NULL
,
0
,
&
dropReq
);
pCxt
->
pCmdMsg
->
pMsg
=
malloc
(
pCxt
->
pCmdMsg
->
msgLen
);
if
(
NULL
==
pCxt
->
pCmdMsg
->
pMsg
)
{
return
TSDB_CODE_OUT_OF_MEMORY
;
}
tSerializeSMDropTopicReq
(
pCxt
->
pCmdMsg
->
pMsg
,
pCxt
->
pCmdMsg
->
msgLen
,
&
dropReq
);
return
TSDB_CODE_SUCCESS
;
}
static
int32_t
translateQuery
(
STranslateContext
*
pCxt
,
SNode
*
pNode
)
{
int32_t
code
=
TSDB_CODE_SUCCESS
;
switch
(
nodeType
(
pNode
))
{
...
...
@@ -1232,6 +1376,9 @@ static int32_t translateQuery(STranslateContext* pCxt, SNode* pNode) {
case
QUERY_NODE_DROP_DATABASE_STMT
:
code
=
translateDropDatabase
(
pCxt
,
(
SDropDatabaseStmt
*
)
pNode
);
break
;
case
QUERY_NODE_ALTER_DATABASE_STMT
:
code
=
translateAlterDatabase
(
pCxt
,
(
SAlterDatabaseStmt
*
)
pNode
);
break
;
case
QUERY_NODE_CREATE_TABLE_STMT
:
code
=
translateCreateSuperTable
(
pCxt
,
(
SCreateTableStmt
*
)
pNode
);
break
;
...
...
@@ -1274,9 +1421,21 @@ static int32_t translateQuery(STranslateContext* pCxt, SNode* pNode) {
case
QUERY_NODE_CREATE_INDEX_STMT
:
code
=
translateCreateIndex
(
pCxt
,
(
SCreateIndexStmt
*
)
pNode
);
break
;
case
QUERY_NODE_DROP_INDEX_STMT
:
code
=
translateDropIndex
(
pCxt
,
(
SDropIndexStmt
*
)
pNode
);
break
;
case
QUERY_NODE_CREATE_QNODE_STMT
:
code
=
translateCreateQnode
(
pCxt
,
(
SCreateQnodeStmt
*
)
pNode
);
break
;
case
QUERY_NODE_DROP_QNODE_STMT
:
code
=
translateDropQnode
(
pCxt
,
(
SDropQnodeStmt
*
)
pNode
);
break
;
case
QUERY_NODE_CREATE_TOPIC_STMT
:
code
=
translateCreateTopic
(
pCxt
,
(
SCreateTopicStmt
*
)
pNode
);
break
;
case
QUERY_NODE_DROP_TOPIC_STMT
:
code
=
translateDropTopic
(
pCxt
,
(
SDropTopicStmt
*
)
pNode
);
break
;
default:
break
;
}
...
...
source/libs/parser/src/sql.c
浏览文件 @
cfbd7704
因为 它太大了无法显示 source diff 。你可以改为
查看blob
。
source/libs/parser/test/parserAstTest.cpp
浏览文件 @
cfbd7704
...
...
@@ -340,6 +340,23 @@ TEST_F(ParserTest, createDatabase) {
ASSERT_TRUE
(
run
());
}
TEST_F
(
ParserTest
,
alterDatabase
)
{
setDatabase
(
"root"
,
"test"
);
bind
(
"alter database wxy_db BLOCKS 200"
);
ASSERT_TRUE
(
run
());
bind
(
"alter database wxy_db "
"BLOCKS 200 "
"CACHELAST 1 "
"FSYNC 200 "
"KEEP 200 "
"QUORUM 2 "
"WAL 1 "
);
ASSERT_TRUE
(
run
());
}
TEST_F
(
ParserTest
,
showDatabase
)
{
setDatabase
(
"root"
,
"test"
);
...
...
@@ -406,9 +423,49 @@ TEST_F(ParserTest, createSmaIndex) {
ASSERT_TRUE
(
run
());
}
TEST_F
(
ParserTest
,
dropIndex
)
{
setDatabase
(
"root"
,
"test"
);
bind
(
"drop index index1 on t1"
);
ASSERT_TRUE
(
run
());
}
TEST_F
(
ParserTest
,
createQnode
)
{
setDatabase
(
"root"
,
"test"
);
bind
(
"create qnode on dnode 1"
);
ASSERT_TRUE
(
run
());
}
TEST_F
(
ParserTest
,
dropQnode
)
{
setDatabase
(
"root"
,
"test"
);
bind
(
"drop qnode on dnode 1"
);
ASSERT_TRUE
(
run
());
}
TEST_F
(
ParserTest
,
createTopic
)
{
setDatabase
(
"root"
,
"test"
);
bind
(
"create topic tp1 as select * from t1"
);
ASSERT_TRUE
(
run
());
bind
(
"create topic if not exists tp1 as select * from t1"
);
ASSERT_TRUE
(
run
());
bind
(
"create topic tp1 as test"
);
ASSERT_TRUE
(
run
());
bind
(
"create topic if not exists tp1 as test"
);
ASSERT_TRUE
(
run
());
}
TEST_F
(
ParserTest
,
dropTopic
)
{
setDatabase
(
"root"
,
"test"
);
bind
(
"drop topic tp1"
);
ASSERT_TRUE
(
run
());
bind
(
"drop topic if exists tp1"
);
ASSERT_TRUE
(
run
());
}
source/libs/transport/src/transSrv.c
浏览文件 @
cfbd7704
...
...
@@ -214,7 +214,9 @@ static void uvHandleReq(SSrvConn* pConn) {
// pHead = rpcDecompresSTransMsg(pHead);
}
else
{
pHead
->
msgLen
=
htonl
(
pHead
->
msgLen
);
// impl later
if
(
pHead
->
secured
==
1
)
{
pHead
->
msgLen
-=
sizeof
(
STransUserMsg
);
}
//
}
...
...
source/util/src/tuuid.c
浏览文件 @
cfbd7704
...
...
@@ -15,43 +15,43 @@
#include "tuuid.h"
static
int64_t
h
ashId
=
0
;
static
int32_t
SerialNo
=
0
;
static
int64_t
tUUIDH
ashId
=
0
;
static
int32_t
tUUID
SerialNo
=
0
;
int32_t
tGenIdPI32
(
void
)
{
if
(
h
ashId
==
0
)
{
if
(
tUUIDH
ashId
==
0
)
{
char
uid
[
64
];
int32_t
code
=
taosGetSystemUUID
(
uid
,
tListLen
(
uid
));
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
terrno
=
TAOS_SYSTEM_ERROR
(
errno
);
}
else
{
h
ashId
=
MurmurHash3_32
(
uid
,
strlen
(
uid
));
tUUIDH
ashId
=
MurmurHash3_32
(
uid
,
strlen
(
uid
));
}
}
int64_t
ts
=
taosGetTimestampMs
();
uint64_t
pid
=
taosGetPId
();
int32_t
val
=
atomic_add_fetch_32
(
&
SerialNo
,
1
);
int32_t
val
=
atomic_add_fetch_32
(
&
tUUID
SerialNo
,
1
);
int32_t
id
=
((
h
ashId
&
0x1F
)
<<
26
)
|
((
pid
&
0x3F
)
<<
20
)
|
((
ts
&
0xFFF
)
<<
8
)
|
(
val
&
0xFF
);
int32_t
id
=
((
tUUIDH
ashId
&
0x1F
)
<<
26
)
|
((
pid
&
0x3F
)
<<
20
)
|
((
ts
&
0xFFF
)
<<
8
)
|
(
val
&
0xFF
);
return
id
;
}
int64_t
tGenIdPI64
(
void
)
{
if
(
h
ashId
==
0
)
{
if
(
tUUIDH
ashId
==
0
)
{
char
uid
[
64
];
int32_t
code
=
taosGetSystemUUID
(
uid
,
tListLen
(
uid
));
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
terrno
=
TAOS_SYSTEM_ERROR
(
errno
);
}
else
{
h
ashId
=
MurmurHash3_32
(
uid
,
strlen
(
uid
));
tUUIDH
ashId
=
MurmurHash3_32
(
uid
,
strlen
(
uid
));
}
}
int64_t
ts
=
taosGetTimestampMs
();
uint64_t
pid
=
taosGetPId
();
int32_t
val
=
atomic_add_fetch_32
(
&
SerialNo
,
1
);
int32_t
val
=
atomic_add_fetch_32
(
&
tUUID
SerialNo
,
1
);
int64_t
id
=
((
h
ashId
&
0x07FF
)
<<
52
)
|
((
pid
&
0x0FFF
)
<<
40
)
|
((
ts
&
0xFFFFFF
)
<<
16
)
|
(
val
&
0xFFFF
);
int64_t
id
=
((
tUUIDH
ashId
&
0x07FF
)
<<
52
)
|
((
pid
&
0x0FFF
)
<<
40
)
|
((
ts
&
0xFFFFFF
)
<<
16
)
|
(
val
&
0xFFFF
);
return
id
;
}
tests/script/tsim/query/interval.sim
0 → 100644
浏览文件 @
cfbd7704
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/cfg.sh -n dnode1 -c wal -v 1
system sh/exec.sh -n dnode1 -s start
sleep 2000
sql connect
$dbPrefix = m_in_db
$tbPrefix = m_in_tb
$mtPrefix = m_in_mt
$tbNum = 10
$rowNum = 20
$totalNum = 200
print =============== step1
$i = 0
$db = $dbPrefix . $i
$mt = $mtPrefix . $i
sql drop database $db -x step1
step1:
sql create database $db
sql use $db
sql create table $mt (ts timestamp, tbcol int) TAGS(tgcol int)
print ====== start create child tables and insert data
$i = 0
while $i < $tbNum
$tb = $tbPrefix . $i
sql create table $tb using $mt tags( $i )
$x = 0
while $x < $rowNum
$cc = $x * 60000
$ms = 1601481600000 + $cc
sql insert into $tb values ($ms , $x )
$x = $x + 1
endw
$i = $i + 1
endw
print =============== step2
$i = 1
$tb = $tbPrefix . $i
sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $tb interval(1m)
print ===> $rows
if $rows < $rowNum then
return -1
endi
if $data01 != 1 then
return -1
endi
if $data05 != 1 then
return -1
endi
print =============== step3
$cc = 4 * 60000
$ms = 1601481600000 + $cc
sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $tb where ts <= $ms interval(1m)
print ===> $rows
if $rows > 10 then
return -1
endi
if $rows < 3 then
return -1
endi
if $data01 != 1 then
return -1
endi
if $data05 != 1 then
return -1
endi
print =============== step4
$cc = 40 * 60000
$ms = 1601481600000 + $cc
$cc = 1 * 60000
$ms2 = 1601481600000 - $cc
sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $tb where ts <= $ms and ts > $ms2 interval(1m)
print ===> $rows
if $rows < 18 then
return -1
endi
if $rows > 22 then
return -1
endi
if $data01 != 1 then
return -1
endi
if $data05 != 1 then
return -1
endi
print =============== step5
$cc = 40 * 60000
$ms = 1601481600000 + $cc
$cc = 1 * 60000
$ms2 = 1601481600000 - $cc
sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $tb where ts <= $ms and ts > $ms2 interval(1m) fill(value,0)
print ===> $rows
if $rows < 30 then
return -1
endi
if $rows > 50 then
return -1
endi
if $data21 != 1 then
return -1
endi
if $data25 != 1 then
return -1
endi
print =============== step6
sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $mt interval(1m)
print ===> $rows
if $rows < 18 then
return -1
endi
if $rows > 22 then
return -1
endi
if $data11 > 15 then
return -1
endi
if $data11 < 5 then
return -1
endi
print =============== step7
$cc = 4 * 60000
$ms = 1601481600000 + $cc
sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $mt where ts <= $ms interval(1m)
print ===> $rows
if $rows < 3 then
return -1
endi
if $rows > 7 then
return -1
endi
if $data11 > 15 then
return -1
endi
if $data11 < 5 then
return -1
endi
print =============== step8
$cc = 40 * 60000
$ms1 = 1601481600000 + $cc
$cc = 1 * 60000
$ms2 = 1601481600000 - $cc
sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $mt where ts <= $ms1 and ts > $ms2 interval(1m)
print ===> $rows
if $rows < 18 then
return -1
endi
if $rows > 22 then
return -1
endi
if $data11 > 15 then
return -1
endi
if $data11 < 5 then
return -1
endi
print =============== step9
$cc = 40 * 60000
$ms1 = 1601481600000 + $cc
$cc = 1 * 60000
$ms2 = 1601481600000 - $cc
sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $mt where ts <= $ms1 and ts > $ms2 interval(1m) fill(value, 0)
if $rows < 30 then
return -1
endi
if $rows > 50 then
return -1
endi
if $data11 > 15 then
return -1
endi
if $data11 < 5 then
return -1
endi
print =============== clear
sql drop database $db
sql show databases
if $rows != 0 then
return -1
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT
\ No newline at end of file
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录