Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
2b6cbbf7
T
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1184
Star
22015
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
2b6cbbf7
编写于
6月 15, 2022
作者:
S
Shengliang Guan
浏览文件
操作
浏览文件
下载
差异文件
Merge remote-tracking branch 'origin/3.0' into fix/mnode
上级
b94772e8
38031553
变更
17
隐藏空白更改
内联
并排
Showing
17 changed file
with
584 addition
and
157 deletion
+584
-157
source/common/src/systable.c
source/common/src/systable.c
+6
-6
source/dnode/mnode/impl/src/mndDb.c
source/dnode/mnode/impl/src/mndDb.c
+7
-7
source/dnode/mnode/impl/src/mndSma.c
source/dnode/mnode/impl/src/mndSma.c
+15
-17
source/libs/planner/src/planSpliter.c
source/libs/planner/src/planSpliter.c
+177
-107
source/libs/planner/test/planGroupByTest.cpp
source/libs/planner/test/planGroupByTest.cpp
+2
-0
source/libs/planner/test/planIntervalTest.cpp
source/libs/planner/test/planIntervalTest.cpp
+2
-0
source/libs/planner/test/planJoinTest.cpp
source/libs/planner/test/planJoinTest.cpp
+6
-0
source/libs/planner/test/planOrderByTest.cpp
source/libs/planner/test/planOrderByTest.cpp
+2
-0
source/libs/planner/test/planOtherTest.cpp
source/libs/planner/test/planOtherTest.cpp
+7
-0
source/libs/planner/test/planTestUtil.h
source/libs/planner/test/planTestUtil.h
+4
-0
tests/script/jenkins/basic.txt
tests/script/jenkins/basic.txt
+1
-1
tests/script/tsim/sma/tsmaCreateInsertData.sim
tests/script/tsim/sma/tsmaCreateInsertData.sim
+8
-0
tests/system-test/6-cluster/5dnode3mnodeDrop.py
tests/system-test/6-cluster/5dnode3mnodeDrop.py
+4
-2
tests/system-test/6-cluster/5dnode3mnodeStop.py
tests/system-test/6-cluster/5dnode3mnodeStop.py
+12
-9
tests/system-test/6-cluster/5dnode3mnodeStopInsert.py
tests/system-test/6-cluster/5dnode3mnodeStopInsert.py
+11
-6
tests/system-test/7-tmq/tmqError.py
tests/system-test/7-tmq/tmqError.py
+315
-0
tests/system-test/fulltest.sh
tests/system-test/fulltest.sh
+5
-2
未找到文件。
source/common/src/systable.c
浏览文件 @
2b6cbbf7
...
@@ -91,8 +91,8 @@ static const SSysDbTableSchema userDBSchema[] = {
...
@@ -91,8 +91,8 @@ static const SSysDbTableSchema userDBSchema[] = {
{.
name
=
"precision"
,
.
bytes
=
2
+
VARSTR_HEADER_SIZE
,
.
type
=
TSDB_DATA_TYPE_VARCHAR
},
{.
name
=
"precision"
,
.
bytes
=
2
+
VARSTR_HEADER_SIZE
,
.
type
=
TSDB_DATA_TYPE_VARCHAR
},
{.
name
=
"single_stable_model"
,
.
bytes
=
1
,
.
type
=
TSDB_DATA_TYPE_BOOL
},
{.
name
=
"single_stable_model"
,
.
bytes
=
1
,
.
type
=
TSDB_DATA_TYPE_BOOL
},
{.
name
=
"status"
,
.
bytes
=
10
+
VARSTR_HEADER_SIZE
,
.
type
=
TSDB_DATA_TYPE_VARCHAR
},
{.
name
=
"status"
,
.
bytes
=
10
+
VARSTR_HEADER_SIZE
,
.
type
=
TSDB_DATA_TYPE_VARCHAR
},
// {.name = "schemaless", .bytes = 1, .type = TSDB_DATA_TYPE_BOOL},
// {.name = "schemaless", .bytes = 1, .type = TSDB_DATA_TYPE_BOOL},
{.
name
=
"reten
s
ion"
,
.
bytes
=
60
+
VARSTR_HEADER_SIZE
,
.
type
=
TSDB_DATA_TYPE_VARCHAR
},
{.
name
=
"reten
t
ion"
,
.
bytes
=
60
+
VARSTR_HEADER_SIZE
,
.
type
=
TSDB_DATA_TYPE_VARCHAR
},
// {.name = "update", .bytes = 1, .type = TSDB_DATA_TYPE_TINYINT}, // disable update
// {.name = "update", .bytes = 1, .type = TSDB_DATA_TYPE_TINYINT}, // disable update
};
};
...
@@ -137,7 +137,7 @@ static const SSysDbTableSchema streamSchema[] = {
...
@@ -137,7 +137,7 @@ static const SSysDbTableSchema streamSchema[] = {
{.
name
=
"target_table"
,
.
bytes
=
SYSTABLE_SCH_TABLE_NAME_LEN
,
.
type
=
TSDB_DATA_TYPE_VARCHAR
},
{.
name
=
"target_table"
,
.
bytes
=
SYSTABLE_SCH_TABLE_NAME_LEN
,
.
type
=
TSDB_DATA_TYPE_VARCHAR
},
{.
name
=
"watermark"
,
.
bytes
=
8
,
.
type
=
TSDB_DATA_TYPE_BIGINT
},
{.
name
=
"watermark"
,
.
bytes
=
8
,
.
type
=
TSDB_DATA_TYPE_BIGINT
},
{.
name
=
"trigger"
,
.
bytes
=
4
,
.
type
=
TSDB_DATA_TYPE_INT
},
{.
name
=
"trigger"
,
.
bytes
=
4
,
.
type
=
TSDB_DATA_TYPE_INT
},
};
};
static
const
SSysDbTableSchema
userTblsSchema
[]
=
{
static
const
SSysDbTableSchema
userTblsSchema
[]
=
{
{.
name
=
"table_name"
,
.
bytes
=
SYSTABLE_SCH_TABLE_NAME_LEN
,
.
type
=
TSDB_DATA_TYPE_VARCHAR
},
{.
name
=
"table_name"
,
.
bytes
=
SYSTABLE_SCH_TABLE_NAME_LEN
,
.
type
=
TSDB_DATA_TYPE_VARCHAR
},
...
@@ -221,7 +221,9 @@ static const SSysDbTableSchema transSchema[] = {
...
@@ -221,7 +221,9 @@ static const SSysDbTableSchema transSchema[] = {
{.
name
=
"db"
,
.
bytes
=
SYSTABLE_SCH_DB_NAME_LEN
,
.
type
=
TSDB_DATA_TYPE_VARCHAR
},
{.
name
=
"db"
,
.
bytes
=
SYSTABLE_SCH_DB_NAME_LEN
,
.
type
=
TSDB_DATA_TYPE_VARCHAR
},
{.
name
=
"failed_times"
,
.
bytes
=
4
,
.
type
=
TSDB_DATA_TYPE_INT
},
{.
name
=
"failed_times"
,
.
bytes
=
4
,
.
type
=
TSDB_DATA_TYPE_INT
},
{.
name
=
"last_exec_time"
,
.
bytes
=
8
,
.
type
=
TSDB_DATA_TYPE_TIMESTAMP
},
{.
name
=
"last_exec_time"
,
.
bytes
=
8
,
.
type
=
TSDB_DATA_TYPE_TIMESTAMP
},
{.
name
=
"last_action_info"
,
.
bytes
=
(
TSDB_TRANS_ERROR_LEN
-
1
)
+
VARSTR_HEADER_SIZE
,
.
type
=
TSDB_DATA_TYPE_VARCHAR
},
{.
name
=
"last_action_info"
,
.
bytes
=
(
TSDB_TRANS_ERROR_LEN
-
1
)
+
VARSTR_HEADER_SIZE
,
.
type
=
TSDB_DATA_TYPE_VARCHAR
},
};
};
static
const
SSysDbTableSchema
configSchema
[]
=
{
static
const
SSysDbTableSchema
configSchema
[]
=
{
...
@@ -314,8 +316,6 @@ static const SSysDbTableSchema querySchema[] = {
...
@@ -314,8 +316,6 @@ static const SSysDbTableSchema querySchema[] = {
{.
name
=
"sql"
,
.
bytes
=
TSDB_SHOW_SQL_LEN
+
VARSTR_HEADER_SIZE
,
.
type
=
TSDB_DATA_TYPE_VARCHAR
},
{.
name
=
"sql"
,
.
bytes
=
TSDB_SHOW_SQL_LEN
+
VARSTR_HEADER_SIZE
,
.
type
=
TSDB_DATA_TYPE_VARCHAR
},
};
};
static
const
SSysTableMeta
perfsMeta
[]
=
{
static
const
SSysTableMeta
perfsMeta
[]
=
{
{
TSDB_PERFS_TABLE_CONNECTIONS
,
connectionsSchema
,
tListLen
(
connectionsSchema
)},
{
TSDB_PERFS_TABLE_CONNECTIONS
,
connectionsSchema
,
tListLen
(
connectionsSchema
)},
{
TSDB_PERFS_TABLE_QUERIES
,
querySchema
,
tListLen
(
querySchema
)},
{
TSDB_PERFS_TABLE_QUERIES
,
querySchema
,
tListLen
(
querySchema
)},
...
...
source/dnode/mnode/impl/src/mndDb.c
浏览文件 @
2b6cbbf7
...
@@ -183,12 +183,12 @@ static SSdbRow *mndDbActionDecode(SSdbRaw *pRaw) {
...
@@ -183,12 +183,12 @@ static SSdbRow *mndDbActionDecode(SSdbRaw *pRaw) {
pDb
->
cfg
.
pRetensions
=
taosArrayInit
(
pDb
->
cfg
.
numOfRetensions
,
sizeof
(
SRetention
));
pDb
->
cfg
.
pRetensions
=
taosArrayInit
(
pDb
->
cfg
.
numOfRetensions
,
sizeof
(
SRetention
));
if
(
pDb
->
cfg
.
pRetensions
==
NULL
)
goto
_OVER
;
if
(
pDb
->
cfg
.
pRetensions
==
NULL
)
goto
_OVER
;
for
(
int32_t
i
=
0
;
i
<
pDb
->
cfg
.
numOfRetensions
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pDb
->
cfg
.
numOfRetensions
;
++
i
)
{
SRetention
reten
s
ion
=
{
0
};
SRetention
reten
t
ion
=
{
0
};
SDB_GET_INT64
(
pRaw
,
dataPos
,
&
reten
s
ion
.
freq
,
_OVER
)
SDB_GET_INT64
(
pRaw
,
dataPos
,
&
reten
t
ion
.
freq
,
_OVER
)
SDB_GET_INT64
(
pRaw
,
dataPos
,
&
reten
s
ion
.
keep
,
_OVER
)
SDB_GET_INT64
(
pRaw
,
dataPos
,
&
reten
t
ion
.
keep
,
_OVER
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
reten
s
ion
.
freqUnit
,
_OVER
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
reten
t
ion
.
freqUnit
,
_OVER
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
reten
s
ion
.
keepUnit
,
_OVER
)
SDB_GET_INT8
(
pRaw
,
dataPos
,
&
reten
t
ion
.
keepUnit
,
_OVER
)
if
(
taosArrayPush
(
pDb
->
cfg
.
pRetensions
,
&
reten
s
ion
)
==
NULL
)
{
if
(
taosArrayPush
(
pDb
->
cfg
.
pRetensions
,
&
reten
t
ion
)
==
NULL
)
{
goto
_OVER
;
goto
_OVER
;
}
}
}
}
...
@@ -1382,7 +1382,7 @@ static void dumpDbInfoData(SSDataBlock *pBlock, SDbObj *pDb, SShowObj *pShow, in
...
@@ -1382,7 +1382,7 @@ static void dumpDbInfoData(SSDataBlock *pBlock, SDbObj *pDb, SShowObj *pShow, in
char
*
status
=
"ready"
;
char
*
status
=
"ready"
;
if
(
objStatus
==
SDB_STATUS_CREATING
)
status
=
"creating"
;
if
(
objStatus
==
SDB_STATUS_CREATING
)
status
=
"creating"
;
if
(
objStatus
==
SDB_STATUS_DROPPING
)
status
=
"dropping"
;
if
(
objStatus
==
SDB_STATUS_DROPPING
)
status
=
"dropping"
;
char
statusB
[
24
]
=
{
0
};
char
statusB
[
24
]
=
{
0
};
STR_WITH_SIZE_TO_VARSTR
(
statusB
,
status
,
strlen
(
status
));
STR_WITH_SIZE_TO_VARSTR
(
statusB
,
status
,
strlen
(
status
));
if
(
sysDb
)
{
if
(
sysDb
)
{
...
...
source/dnode/mnode/impl/src/mndSma.c
浏览文件 @
2b6cbbf7
...
@@ -44,6 +44,7 @@ static int32_t mndProcessGetSmaReq(SRpcMsg *pReq);
...
@@ -44,6 +44,7 @@ static int32_t mndProcessGetSmaReq(SRpcMsg *pReq);
static
int32_t
mndProcessGetTbSmaReq
(
SRpcMsg
*
pReq
);
static
int32_t
mndProcessGetTbSmaReq
(
SRpcMsg
*
pReq
);
static
int32_t
mndRetrieveSma
(
SRpcMsg
*
pReq
,
SShowObj
*
pShow
,
SSDataBlock
*
pBlock
,
int32_t
rows
);
static
int32_t
mndRetrieveSma
(
SRpcMsg
*
pReq
,
SShowObj
*
pShow
,
SSDataBlock
*
pBlock
,
int32_t
rows
);
static
void
mndCancelGetNextSma
(
SMnode
*
pMnode
,
void
*
pIter
);
static
void
mndCancelGetNextSma
(
SMnode
*
pMnode
,
void
*
pIter
);
static
void
mndDestroySmaObj
(
SSmaObj
*
pSmaObj
);
int32_t
mndInitSma
(
SMnode
*
pMnode
)
{
int32_t
mndInitSma
(
SMnode
*
pMnode
)
{
SSdbTable
table
=
{
SSdbTable
table
=
{
...
@@ -390,7 +391,9 @@ static int32_t mndSetUpdateSmaStbCommitLogs(SMnode *pMnode, STrans *pTrans, SStb
...
@@ -390,7 +391,9 @@ static int32_t mndSetUpdateSmaStbCommitLogs(SMnode *pMnode, STrans *pTrans, SStb
taosRLockLatch
(
&
pStb
->
lock
);
taosRLockLatch
(
&
pStb
->
lock
);
memcpy
(
&
stbObj
,
pStb
,
sizeof
(
SStbObj
));
memcpy
(
&
stbObj
,
pStb
,
sizeof
(
SStbObj
));
taosRUnLockLatch
(
&
pStb
->
lock
);
taosRUnLockLatch
(
&
pStb
->
lock
);
stbObj
.
numOfColumns
=
0
;
stbObj
.
pColumns
=
NULL
;
stbObj
.
pColumns
=
NULL
;
stbObj
.
numOfTags
=
0
;
stbObj
.
pTags
=
NULL
;
stbObj
.
pTags
=
NULL
;
stbObj
.
updateTime
=
taosGetTimestampMs
();
stbObj
.
updateTime
=
taosGetTimestampMs
();
stbObj
.
lock
=
0
;
stbObj
.
lock
=
0
;
...
@@ -501,6 +504,13 @@ static int32_t mndSetCreateSmaVgroupRedoActions(SMnode *pMnode, STrans *pTrans,
...
@@ -501,6 +504,13 @@ static int32_t mndSetCreateSmaVgroupRedoActions(SMnode *pMnode, STrans *pTrans,
return
0
;
return
0
;
}
}
static
void
mndDestroySmaObj
(
SSmaObj
*
pSmaObj
)
{
if
(
pSmaObj
)
{
taosMemoryFreeClear
(
pSmaObj
->
schemaRow
.
pSchema
);
taosMemoryFreeClear
(
pSmaObj
->
schemaTag
.
pSchema
);
}
}
static
int32_t
mndCreateSma
(
SMnode
*
pMnode
,
SRpcMsg
*
pReq
,
SMCreateSmaReq
*
pCreate
,
SDbObj
*
pDb
,
SStbObj
*
pStb
)
{
static
int32_t
mndCreateSma
(
SMnode
*
pMnode
,
SRpcMsg
*
pReq
,
SMCreateSmaReq
*
pCreate
,
SDbObj
*
pDb
,
SStbObj
*
pStb
)
{
SSmaObj
smaObj
=
{
0
};
SSmaObj
smaObj
=
{
0
};
memcpy
(
smaObj
.
name
,
pCreate
->
name
,
TSDB_TABLE_FNAME_LEN
);
memcpy
(
smaObj
.
name
,
pCreate
->
name
,
TSDB_TABLE_FNAME_LEN
);
...
@@ -524,29 +534,17 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
...
@@ -524,29 +534,17 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
smaObj
.
tagsFilterLen
=
pCreate
->
tagsFilterLen
;
smaObj
.
tagsFilterLen
=
pCreate
->
tagsFilterLen
;
smaObj
.
sqlLen
=
pCreate
->
sqlLen
;
smaObj
.
sqlLen
=
pCreate
->
sqlLen
;
smaObj
.
astLen
=
pCreate
->
astLen
;
smaObj
.
astLen
=
pCreate
->
astLen
;
if
(
smaObj
.
exprLen
>
0
)
{
if
(
smaObj
.
exprLen
>
0
)
{
smaObj
.
expr
=
taosMemoryMalloc
(
smaObj
.
exprLen
);
smaObj
.
expr
=
pCreate
->
expr
;
if
(
smaObj
.
expr
==
NULL
)
goto
_OVER
;
memcpy
(
smaObj
.
expr
,
pCreate
->
expr
,
smaObj
.
exprLen
);
}
}
if
(
smaObj
.
tagsFilterLen
>
0
)
{
if
(
smaObj
.
tagsFilterLen
>
0
)
{
smaObj
.
tagsFilter
=
taosMemoryMalloc
(
smaObj
.
tagsFilterLen
);
smaObj
.
tagsFilter
=
pCreate
->
tagsFilter
;
if
(
smaObj
.
tagsFilter
==
NULL
)
goto
_OVER
;
memcpy
(
smaObj
.
tagsFilter
,
pCreate
->
tagsFilter
,
smaObj
.
tagsFilterLen
);
}
}
if
(
smaObj
.
sqlLen
>
0
)
{
if
(
smaObj
.
sqlLen
>
0
)
{
smaObj
.
sql
=
taosMemoryMalloc
(
smaObj
.
sqlLen
);
smaObj
.
sql
=
pCreate
->
sql
;
if
(
smaObj
.
sql
==
NULL
)
goto
_OVER
;
memcpy
(
smaObj
.
sql
,
pCreate
->
sql
,
smaObj
.
sqlLen
);
}
}
if
(
smaObj
.
astLen
>
0
)
{
if
(
smaObj
.
astLen
>
0
)
{
smaObj
.
ast
=
taosMemoryMalloc
(
smaObj
.
astLen
);
smaObj
.
ast
=
pCreate
->
ast
;
if
(
smaObj
.
ast
==
NULL
)
goto
_OVER
;
memcpy
(
smaObj
.
ast
,
pCreate
->
ast
,
smaObj
.
astLen
);
}
}
SStreamObj
streamObj
=
{
0
};
SStreamObj
streamObj
=
{
0
};
...
@@ -589,6 +587,7 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
...
@@ -589,6 +587,7 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
code
=
0
;
code
=
0
;
_OVER:
_OVER:
mndDestroySmaObj
(
&
smaObj
);
mndTransDrop
(
pTrans
);
mndTransDrop
(
pTrans
);
return
code
;
return
code
;
}
}
...
@@ -1012,7 +1011,6 @@ int32_t mndGetTableSma(SMnode *pMnode, char *tbFName, STableIndexRsp *rsp, bool
...
@@ -1012,7 +1011,6 @@ int32_t mndGetTableSma(SMnode *pMnode, char *tbFName, STableIndexRsp *rsp, bool
rsp
->
suid
=
pStb
->
uid
;
rsp
->
suid
=
pStb
->
uid
;
rsp
->
version
=
pStb
->
smaVer
;
rsp
->
version
=
pStb
->
smaVer
;
mndReleaseStb
(
pMnode
,
pStb
);
mndReleaseStb
(
pMnode
,
pStb
);
while
(
1
)
{
while
(
1
)
{
pIter
=
sdbFetch
(
pSdb
,
SDB_SMA
,
pIter
,
(
void
**
)
&
pSma
);
pIter
=
sdbFetch
(
pSdb
,
SDB_SMA
,
pIter
,
(
void
**
)
&
pSma
);
...
...
source/libs/planner/src/planSpliter.c
浏览文件 @
2b6cbbf7
...
@@ -15,6 +15,7 @@
...
@@ -15,6 +15,7 @@
#include "functionMgt.h"
#include "functionMgt.h"
#include "planInt.h"
#include "planInt.h"
#include "tglobal.h"
#define SPLIT_FLAG_MASK(n) (1 << n)
#define SPLIT_FLAG_MASK(n) (1 << n)
...
@@ -37,7 +38,8 @@ typedef struct SSplitRule {
...
@@ -37,7 +38,8 @@ typedef struct SSplitRule {
FSplit
splitFunc
;
FSplit
splitFunc
;
}
SSplitRule
;
}
SSplitRule
;
typedef
bool
(
*
FSplFindSplitNode
)(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
void
*
pInfo
);
// typedef bool (*FSplFindSplitNode)(SSplitContext* pCxt, SLogicSubplan* pSubplan, void* pInfo);
typedef
bool
(
*
FSplFindSplitNode
)(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
SLogicNode
*
pNode
,
void
*
pInfo
);
static
void
splSetSubplanVgroups
(
SLogicSubplan
*
pSubplan
,
SLogicNode
*
pNode
)
{
static
void
splSetSubplanVgroups
(
SLogicSubplan
*
pSubplan
,
SLogicNode
*
pNode
)
{
if
(
QUERY_NODE_LOGIC_PLAN_SCAN
==
nodeType
(
pNode
))
{
if
(
QUERY_NODE_LOGIC_PLAN_SCAN
==
nodeType
(
pNode
))
{
...
@@ -95,9 +97,23 @@ static int32_t splCreateExchangeNodeForSubplan(SSplitContext* pCxt, SLogicSubpla
...
@@ -95,9 +97,23 @@ static int32_t splCreateExchangeNodeForSubplan(SSplitContext* pCxt, SLogicSubpla
return
code
;
return
code
;
}
}
static
bool
splMatchByNode
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
SLogicNode
*
pNode
,
FSplFindSplitNode
func
,
void
*
pInfo
)
{
if
(
func
(
pCxt
,
pSubplan
,
pNode
,
pInfo
))
{
return
true
;
}
SNode
*
pChild
;
FOREACH
(
pChild
,
pNode
->
pChildren
)
{
if
(
splMatchByNode
(
pCxt
,
pSubplan
,
(
SLogicNode
*
)
pChild
,
func
,
pInfo
))
{
return
true
;
}
}
return
NULL
;
}
static
bool
splMatch
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
int32_t
flag
,
FSplFindSplitNode
func
,
void
*
pInfo
)
{
static
bool
splMatch
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
int32_t
flag
,
FSplFindSplitNode
func
,
void
*
pInfo
)
{
if
(
!
SPLIT_FLAG_TEST_MASK
(
pSubplan
->
splitFlag
,
flag
))
{
if
(
!
SPLIT_FLAG_TEST_MASK
(
pSubplan
->
splitFlag
,
flag
))
{
if
(
func
(
pCxt
,
pSubplan
,
pInfo
))
{
if
(
splMatchByNode
(
pCxt
,
pSubplan
,
pSubplan
->
pNode
,
func
,
pInfo
))
{
return
true
;
return
true
;
}
}
}
}
...
@@ -110,6 +126,11 @@ static bool splMatch(SSplitContext* pCxt, SLogicSubplan* pSubplan, int32_t flag,
...
@@ -110,6 +126,11 @@ static bool splMatch(SSplitContext* pCxt, SLogicSubplan* pSubplan, int32_t flag,
return
false
;
return
false
;
}
}
static
void
splSetParent
(
SLogicNode
*
pNode
)
{
SNode
*
pChild
=
NULL
;
FOREACH
(
pChild
,
pNode
->
pChildren
)
{
((
SLogicNode
*
)
pChild
)
->
pParent
=
pNode
;
}
}
typedef
struct
SStableSplitInfo
{
typedef
struct
SStableSplitInfo
{
SLogicNode
*
pSplitNode
;
SLogicNode
*
pSplitNode
;
SLogicSubplan
*
pSubplan
;
SLogicSubplan
*
pSubplan
;
...
@@ -136,11 +157,21 @@ static bool stbSplHasMultiTbScan(bool streamQuery, SLogicNode* pNode) {
...
@@ -136,11 +157,21 @@ static bool stbSplHasMultiTbScan(bool streamQuery, SLogicNode* pNode) {
return
false
;
return
false
;
}
}
SNode
*
pChild
=
nodesListGetNode
(
pNode
->
pChildren
,
0
);
SNode
*
pChild
=
nodesListGetNode
(
pNode
->
pChildren
,
0
);
if
(
QUERY_NODE_LOGIC_PLAN_PARTITION
==
nodeType
(
pChild
))
{
if
(
1
!=
LIST_LENGTH
(((
SLogicNode
*
)
pChild
)
->
pChildren
))
{
return
false
;
}
pChild
=
nodesListGetNode
(((
SLogicNode
*
)
pChild
)
->
pChildren
,
0
);
}
return
(
QUERY_NODE_LOGIC_PLAN_SCAN
==
nodeType
(
pChild
)
&&
stbSplIsMultiTbScan
(
streamQuery
,
(
SScanLogicNode
*
)
pChild
));
return
(
QUERY_NODE_LOGIC_PLAN_SCAN
==
nodeType
(
pChild
)
&&
stbSplIsMultiTbScan
(
streamQuery
,
(
SScanLogicNode
*
)
pChild
));
}
}
static
bool
stbSplNeedSplit
(
bool
streamQuery
,
SLogicNode
*
pNode
)
{
static
bool
stbSplNeedSplit
(
bool
streamQuery
,
SLogicNode
*
pNode
)
{
switch
(
nodeType
(
pNode
))
{
switch
(
nodeType
(
pNode
))
{
case
QUERY_NODE_LOGIC_PLAN_SCAN
:
return
stbSplIsMultiTbScan
(
streamQuery
,
(
SScanLogicNode
*
)
pNode
);
// case QUERY_NODE_LOGIC_PLAN_JOIN:
// return !(((SJoinLogicNode*)pNode)->isSingleTableJoin);
case
QUERY_NODE_LOGIC_PLAN_AGG
:
case
QUERY_NODE_LOGIC_PLAN_AGG
:
return
!
stbSplHasGatherExecFunc
(((
SAggLogicNode
*
)
pNode
)
->
pAggFuncs
)
&&
stbSplHasMultiTbScan
(
streamQuery
,
pNode
);
return
!
stbSplHasGatherExecFunc
(((
SAggLogicNode
*
)
pNode
)
->
pAggFuncs
)
&&
stbSplHasMultiTbScan
(
streamQuery
,
pNode
);
case
QUERY_NODE_LOGIC_PLAN_WINDOW
:
{
case
QUERY_NODE_LOGIC_PLAN_WINDOW
:
{
...
@@ -152,35 +183,20 @@ static bool stbSplNeedSplit(bool streamQuery, SLogicNode* pNode) {
...
@@ -152,35 +183,20 @@ static bool stbSplNeedSplit(bool streamQuery, SLogicNode* pNode) {
}
}
case
QUERY_NODE_LOGIC_PLAN_SORT
:
case
QUERY_NODE_LOGIC_PLAN_SORT
:
return
stbSplHasMultiTbScan
(
streamQuery
,
pNode
);
return
stbSplHasMultiTbScan
(
streamQuery
,
pNode
);
case
QUERY_NODE_LOGIC_PLAN_SCAN
:
return
stbSplIsMultiTbScan
(
streamQuery
,
(
SScanLogicNode
*
)
pNode
);
default:
default:
break
;
break
;
}
}
return
false
;
return
false
;
}
}
static
SLogicNode
*
stbSplMatchByNode
(
bool
streamQuery
,
SLogicNode
*
pNode
)
{
static
bool
stbSplFindSplitNode
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
SLogicNode
*
pNode
,
if
(
stbSplNeedSplit
(
streamQuery
,
pNode
))
{
SStableSplitInfo
*
pInfo
)
{
return
pNode
;
if
(
stbSplNeedSplit
(
pCxt
->
pPlanCxt
->
streamQuery
,
pNode
))
{
}
pInfo
->
pSplitNode
=
pNode
;
SNode
*
pChild
;
FOREACH
(
pChild
,
pNode
->
pChildren
)
{
SLogicNode
*
pSplitNode
=
stbSplMatchByNode
(
streamQuery
,
(
SLogicNode
*
)
pChild
);
if
(
NULL
!=
pSplitNode
)
{
return
pSplitNode
;
}
}
return
NULL
;
}
static
bool
stbSplFindSplitNode
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
SStableSplitInfo
*
pInfo
)
{
SLogicNode
*
pSplitNode
=
stbSplMatchByNode
(
pCxt
->
pPlanCxt
->
streamQuery
,
pSubplan
->
pNode
);
if
(
NULL
!=
pSplitNode
)
{
pInfo
->
pSplitNode
=
pSplitNode
;
pInfo
->
pSubplan
=
pSubplan
;
pInfo
->
pSubplan
=
pSubplan
;
return
true
;
}
}
return
NULL
!=
pSplitNod
e
;
return
fals
e
;
}
}
static
int32_t
stbSplRewriteFuns
(
const
SNodeList
*
pFuncs
,
SNodeList
**
pPartialFuncs
,
SNodeList
**
pMergeFuncs
)
{
static
int32_t
stbSplRewriteFuns
(
const
SNodeList
*
pFuncs
,
SNodeList
**
pPartialFuncs
,
SNodeList
**
pMergeFuncs
)
{
...
@@ -258,6 +274,7 @@ static int32_t stbSplCreatePartWindowNode(SWindowLogicNode* pMergeWindow, SLogic
...
@@ -258,6 +274,7 @@ static int32_t stbSplCreatePartWindowNode(SWindowLogicNode* pMergeWindow, SLogic
if
(
TSDB_CODE_SUCCESS
==
code
)
{
if
(
TSDB_CODE_SUCCESS
==
code
)
{
pMergeWindow
->
node
.
pTargets
=
pTargets
;
pMergeWindow
->
node
.
pTargets
=
pTargets
;
pPartWin
->
node
.
pChildren
=
pChildren
;
pPartWin
->
node
.
pChildren
=
pChildren
;
splSetParent
((
SLogicNode
*
)
pPartWin
);
code
=
stbSplRewriteFuns
(
pFunc
,
&
pPartWin
->
pFuncs
,
&
pMergeWindow
->
pFuncs
);
code
=
stbSplRewriteFuns
(
pFunc
,
&
pPartWin
->
pFuncs
,
&
pMergeWindow
->
pFuncs
);
}
}
int32_t
index
=
0
;
int32_t
index
=
0
;
...
@@ -285,13 +302,24 @@ static int32_t stbSplCreatePartWindowNode(SWindowLogicNode* pMergeWindow, SLogic
...
@@ -285,13 +302,24 @@ static int32_t stbSplCreatePartWindowNode(SWindowLogicNode* pMergeWindow, SLogic
return
code
;
return
code
;
}
}
static
int32_t
stbSplGetNumOfVgroups
(
SLogicNode
*
pNode
)
{
if
(
QUERY_NODE_LOGIC_PLAN_SCAN
==
nodeType
(
pNode
))
{
return
((
SScanLogicNode
*
)
pNode
)
->
pVgroupList
->
numOfVgroups
;
}
else
{
if
(
1
==
LIST_LENGTH
(
pNode
->
pChildren
))
{
return
stbSplGetNumOfVgroups
((
SLogicNode
*
)
nodesListGetNode
(
pNode
->
pChildren
,
0
));
}
}
return
0
;
}
static
int32_t
stbSplCreateMergeNode
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
SLogicNode
*
pSplitNode
,
static
int32_t
stbSplCreateMergeNode
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
SLogicNode
*
pSplitNode
,
SNodeList
*
pMergeKeys
,
SLogicNode
*
pPartChild
)
{
SNodeList
*
pMergeKeys
,
SLogicNode
*
pPartChild
)
{
SMergeLogicNode
*
pMerge
=
(
SMergeLogicNode
*
)
nodesMakeNode
(
QUERY_NODE_LOGIC_PLAN_MERGE
);
SMergeLogicNode
*
pMerge
=
(
SMergeLogicNode
*
)
nodesMakeNode
(
QUERY_NODE_LOGIC_PLAN_MERGE
);
if
(
NULL
==
pMerge
)
{
if
(
NULL
==
pMerge
)
{
return
TSDB_CODE_OUT_OF_MEMORY
;
return
TSDB_CODE_OUT_OF_MEMORY
;
}
}
pMerge
->
numOfChannels
=
((
SScanLogicNode
*
)
nodesListGetNode
(
pPartChild
->
pChildren
,
0
))
->
pVgroupList
->
numOfVgroups
;
pMerge
->
numOfChannels
=
stbSplGetNumOfVgroups
(
pPartChild
)
;
pMerge
->
srcGroupId
=
pCxt
->
groupId
;
pMerge
->
srcGroupId
=
pCxt
->
groupId
;
pMerge
->
node
.
precision
=
pPartChild
->
precision
;
pMerge
->
node
.
precision
=
pPartChild
->
precision
;
pMerge
->
pMergeKeys
=
pMergeKeys
;
pMerge
->
pMergeKeys
=
pMergeKeys
;
...
@@ -329,12 +357,12 @@ static int32_t stbSplCreateExchangeNode(SSplitContext* pCxt, SLogicNode* pParent
...
@@ -329,12 +357,12 @@ static int32_t stbSplCreateExchangeNode(SSplitContext* pCxt, SLogicNode* pParent
return
code
;
return
code
;
}
}
static
int32_t
stbSplCreateMergeKeys
ForInterval
(
SNode
*
pWStartTs
,
SNodeList
**
pMergeKeys
)
{
static
int32_t
stbSplCreateMergeKeys
ByPrimaryKey
(
SNode
*
pPrimaryKey
,
SNodeList
**
pMergeKeys
)
{
SOrderByExprNode
*
pMergeKey
=
(
SOrderByExprNode
*
)
nodesMakeNode
(
QUERY_NODE_ORDER_BY_EXPR
);
SOrderByExprNode
*
pMergeKey
=
(
SOrderByExprNode
*
)
nodesMakeNode
(
QUERY_NODE_ORDER_BY_EXPR
);
if
(
NULL
==
pMergeKey
)
{
if
(
NULL
==
pMergeKey
)
{
return
TSDB_CODE_OUT_OF_MEMORY
;
return
TSDB_CODE_OUT_OF_MEMORY
;
}
}
pMergeKey
->
pExpr
=
nodesCloneNode
(
p
WStartTs
);
pMergeKey
->
pExpr
=
nodesCloneNode
(
p
PrimaryKey
);
if
(
NULL
==
pMergeKey
->
pExpr
)
{
if
(
NULL
==
pMergeKey
->
pExpr
)
{
nodesDestroyNode
((
SNode
*
)
pMergeKey
);
nodesDestroyNode
((
SNode
*
)
pMergeKey
);
return
TSDB_CODE_OUT_OF_MEMORY
;
return
TSDB_CODE_OUT_OF_MEMORY
;
...
@@ -351,7 +379,7 @@ static int32_t stbSplSplitIntervalForBatch(SSplitContext* pCxt, SStableSplitInfo
...
@@ -351,7 +379,7 @@ static int32_t stbSplSplitIntervalForBatch(SSplitContext* pCxt, SStableSplitInfo
((
SWindowLogicNode
*
)
pPartWindow
)
->
intervalAlgo
=
INTERVAL_ALGO_HASH
;
((
SWindowLogicNode
*
)
pPartWindow
)
->
intervalAlgo
=
INTERVAL_ALGO_HASH
;
((
SWindowLogicNode
*
)
pInfo
->
pSplitNode
)
->
intervalAlgo
=
INTERVAL_ALGO_MERGE
;
((
SWindowLogicNode
*
)
pInfo
->
pSplitNode
)
->
intervalAlgo
=
INTERVAL_ALGO_MERGE
;
SNodeList
*
pMergeKeys
=
NULL
;
SNodeList
*
pMergeKeys
=
NULL
;
code
=
stbSplCreateMergeKeys
ForInterval
(((
SWindowLogicNode
*
)
pInfo
->
pSplitNode
)
->
pTspk
,
&
pMergeKeys
);
code
=
stbSplCreateMergeKeys
ByPrimaryKey
(((
SWindowLogicNode
*
)
pInfo
->
pSplitNode
)
->
pTspk
,
&
pMergeKeys
);
if
(
TSDB_CODE_SUCCESS
==
code
)
{
if
(
TSDB_CODE_SUCCESS
==
code
)
{
code
=
stbSplCreateMergeNode
(
pCxt
,
NULL
,
pInfo
->
pSplitNode
,
pMergeKeys
,
pPartWindow
);
code
=
stbSplCreateMergeNode
(
pCxt
,
NULL
,
pInfo
->
pSplitNode
,
pMergeKeys
,
pPartWindow
);
}
}
...
@@ -439,6 +467,7 @@ static int32_t stbSplCreatePartAggNode(SAggLogicNode* pMergeAgg, SLogicNode** pO
...
@@ -439,6 +467,7 @@ static int32_t stbSplCreatePartAggNode(SAggLogicNode* pMergeAgg, SLogicNode** pO
pMergeAgg
->
node
.
pConditions
=
pConditions
;
pMergeAgg
->
node
.
pConditions
=
pConditions
;
pMergeAgg
->
node
.
pTargets
=
pTargets
;
pMergeAgg
->
node
.
pTargets
=
pTargets
;
pPartAgg
->
node
.
pChildren
=
pChildren
;
pPartAgg
->
node
.
pChildren
=
pChildren
;
splSetParent
((
SLogicNode
*
)
pPartAgg
);
code
=
stbSplRewriteFuns
(
pFunc
,
&
pPartAgg
->
pAggFuncs
,
&
pMergeAgg
->
pAggFuncs
);
code
=
stbSplRewriteFuns
(
pFunc
,
&
pPartAgg
->
pAggFuncs
,
&
pMergeAgg
->
pAggFuncs
);
}
}
...
@@ -553,6 +582,7 @@ static int32_t stbSplCreatePartSortNode(SSortLogicNode* pSort, SLogicNode** pOut
...
@@ -553,6 +582,7 @@ static int32_t stbSplCreatePartSortNode(SSortLogicNode* pSort, SLogicNode** pOut
SNodeList
*
pMergeKeys
=
NULL
;
SNodeList
*
pMergeKeys
=
NULL
;
if
(
TSDB_CODE_SUCCESS
==
code
)
{
if
(
TSDB_CODE_SUCCESS
==
code
)
{
pPartSort
->
node
.
pChildren
=
pChildren
;
pPartSort
->
node
.
pChildren
=
pChildren
;
splSetParent
((
SLogicNode
*
)
pPartSort
);
pPartSort
->
pSortKeys
=
pSortKeys
;
pPartSort
->
pSortKeys
=
pSortKeys
;
code
=
stbSplCreateMergeKeys
(
pPartSort
->
pSortKeys
,
pPartSort
->
node
.
pTargets
,
&
pMergeKeys
);
code
=
stbSplCreateMergeKeys
(
pPartSort
->
pSortKeys
,
pPartSort
->
node
.
pTargets
,
&
pMergeKeys
);
}
}
...
@@ -592,6 +622,56 @@ static int32_t stbSplSplitScanNode(SSplitContext* pCxt, SStableSplitInfo* pInfo)
...
@@ -592,6 +622,56 @@ static int32_t stbSplSplitScanNode(SSplitContext* pCxt, SStableSplitInfo* pInfo)
return
code
;
return
code
;
}
}
static
SNode
*
stbSplFindPrimaryKeyFromScan
(
SScanLogicNode
*
pScan
)
{
SNode
*
pCol
=
NULL
;
FOREACH
(
pCol
,
pScan
->
pScanCols
)
{
if
(
PRIMARYKEY_TIMESTAMP_COL_ID
==
((
SColumnNode
*
)
pCol
)
->
colId
)
{
return
pCol
;
}
}
return
NULL
;
}
static
int32_t
stbSplSplitScanNodeForJoin
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
SScanLogicNode
*
pScan
)
{
SNodeList
*
pMergeKeys
=
NULL
;
int32_t
code
=
stbSplCreateMergeKeysByPrimaryKey
(
stbSplFindPrimaryKeyFromScan
(
pScan
),
&
pMergeKeys
);
if
(
TSDB_CODE_SUCCESS
==
code
)
{
code
=
stbSplCreateMergeNode
(
pCxt
,
pSubplan
,
(
SLogicNode
*
)
pScan
,
pMergeKeys
,
(
SLogicNode
*
)
pScan
);
}
if
(
TSDB_CODE_SUCCESS
==
code
)
{
code
=
nodesListMakeStrictAppend
(
&
pSubplan
->
pChildren
,
(
SNode
*
)
splCreateScanSubplan
(
pCxt
,
(
SLogicNode
*
)
pScan
,
SPLIT_FLAG_STABLE_SPLIT
));
}
return
code
;
}
static
int32_t
stbSplSplitJoinNodeImpl
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
SJoinLogicNode
*
pJoin
)
{
int32_t
code
=
TSDB_CODE_SUCCESS
;
SNode
*
pChild
=
NULL
;
FOREACH
(
pChild
,
pJoin
->
node
.
pChildren
)
{
if
(
QUERY_NODE_LOGIC_PLAN_SCAN
==
nodeType
(
pChild
))
{
code
=
stbSplSplitScanNodeForJoin
(
pCxt
,
pSubplan
,
(
SScanLogicNode
*
)
pChild
);
}
else
if
(
QUERY_NODE_LOGIC_PLAN_JOIN
==
nodeType
(
pChild
))
{
code
=
stbSplSplitJoinNodeImpl
(
pCxt
,
pSubplan
,
(
SJoinLogicNode
*
)
pChild
);
}
else
{
code
=
TSDB_CODE_PLAN_INTERNAL_ERROR
;
}
if
(
TSDB_CODE_SUCCESS
!=
code
)
{
break
;
}
}
return
code
;
}
static
int32_t
stbSplSplitJoinNode
(
SSplitContext
*
pCxt
,
SStableSplitInfo
*
pInfo
)
{
int32_t
code
=
stbSplSplitJoinNodeImpl
(
pCxt
,
pInfo
->
pSubplan
,
(
SJoinLogicNode
*
)
pInfo
->
pSplitNode
);
if
(
TSDB_CODE_SUCCESS
==
code
)
{
pInfo
->
pSubplan
->
subplanType
=
SUBPLAN_TYPE_MERGE
;
SPLIT_FLAG_SET_MASK
(
pInfo
->
pSubplan
->
splitFlag
,
SPLIT_FLAG_STABLE_SPLIT
);
}
return
code
;
}
static
int32_t
stableSplit
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
)
{
static
int32_t
stableSplit
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
)
{
if
(
pCxt
->
pPlanCxt
->
rSmaQuery
)
{
if
(
pCxt
->
pPlanCxt
->
rSmaQuery
)
{
return
TSDB_CODE_SUCCESS
;
return
TSDB_CODE_SUCCESS
;
...
@@ -604,6 +684,12 @@ static int32_t stableSplit(SSplitContext* pCxt, SLogicSubplan* pSubplan) {
...
@@ -604,6 +684,12 @@ static int32_t stableSplit(SSplitContext* pCxt, SLogicSubplan* pSubplan) {
int32_t
code
=
TSDB_CODE_SUCCESS
;
int32_t
code
=
TSDB_CODE_SUCCESS
;
switch
(
nodeType
(
info
.
pSplitNode
))
{
switch
(
nodeType
(
info
.
pSplitNode
))
{
case
QUERY_NODE_LOGIC_PLAN_SCAN
:
code
=
stbSplSplitScanNode
(
pCxt
,
&
info
);
break
;
case
QUERY_NODE_LOGIC_PLAN_JOIN
:
code
=
stbSplSplitJoinNode
(
pCxt
,
&
info
);
break
;
case
QUERY_NODE_LOGIC_PLAN_AGG
:
case
QUERY_NODE_LOGIC_PLAN_AGG
:
code
=
stbSplSplitAggNode
(
pCxt
,
&
info
);
code
=
stbSplSplitAggNode
(
pCxt
,
&
info
);
break
;
break
;
...
@@ -613,9 +699,6 @@ static int32_t stableSplit(SSplitContext* pCxt, SLogicSubplan* pSubplan) {
...
@@ -613,9 +699,6 @@ static int32_t stableSplit(SSplitContext* pCxt, SLogicSubplan* pSubplan) {
case
QUERY_NODE_LOGIC_PLAN_SORT
:
case
QUERY_NODE_LOGIC_PLAN_SORT
:
code
=
stbSplSplitSortNode
(
pCxt
,
&
info
);
code
=
stbSplSplitSortNode
(
pCxt
,
&
info
);
break
;
break
;
case
QUERY_NODE_LOGIC_PLAN_SCAN
:
code
=
stbSplSplitScanNode
(
pCxt
,
&
info
);
break
;
default:
default:
break
;
break
;
}
}
...
@@ -631,7 +714,12 @@ typedef struct SSigTbJoinSplitInfo {
...
@@ -631,7 +714,12 @@ typedef struct SSigTbJoinSplitInfo {
SLogicSubplan
*
pSubplan
;
SLogicSubplan
*
pSubplan
;
}
SSigTbJoinSplitInfo
;
}
SSigTbJoinSplitInfo
;
static
bool
sigTbJoinSplNeedSplit
(
SJoinLogicNode
*
pJoin
)
{
static
bool
sigTbJoinSplNeedSplit
(
SLogicNode
*
pNode
)
{
if
(
QUERY_NODE_LOGIC_PLAN_JOIN
!=
nodeType
(
pNode
))
{
return
false
;
}
SJoinLogicNode
*
pJoin
=
(
SJoinLogicNode
*
)
pNode
;
if
(
!
pJoin
->
isSingleTableJoin
)
{
if
(
!
pJoin
->
isSingleTableJoin
)
{
return
false
;
return
false
;
}
}
...
@@ -639,28 +727,15 @@ static bool sigTbJoinSplNeedSplit(SJoinLogicNode* pJoin) {
...
@@ -639,28 +727,15 @@ static bool sigTbJoinSplNeedSplit(SJoinLogicNode* pJoin) {
QUERY_NODE_LOGIC_PLAN_EXCHANGE
!=
nodeType
(
nodesListGetNode
(
pJoin
->
node
.
pChildren
,
1
));
QUERY_NODE_LOGIC_PLAN_EXCHANGE
!=
nodeType
(
nodesListGetNode
(
pJoin
->
node
.
pChildren
,
1
));
}
}
static
SJoinLogicNode
*
sigTbJoinSplMatchByNode
(
SLogicNode
*
pNode
)
{
static
bool
sigTbJoinSplFindSplitNode
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
SLogicNode
*
pNode
,
if
(
QUERY_NODE_LOGIC_PLAN_JOIN
==
nodeType
(
pNode
)
&&
sigTbJoinSplNeedSplit
((
SJoinLogicNode
*
)
pNode
))
{
SSigTbJoinSplitInfo
*
pInfo
)
{
return
(
SJoinLogicNode
*
)
pNode
;
if
(
sigTbJoinSplNeedSplit
(
pNode
))
{
}
pInfo
->
pJoin
=
(
SJoinLogicNode
*
)
pNode
;
SNode
*
pChild
;
pInfo
->
pSplitNode
=
(
SLogicNode
*
)
nodesListGetNode
(
pNode
->
pChildren
,
1
);
FOREACH
(
pChild
,
pNode
->
pChildren
)
{
SJoinLogicNode
*
pSplitNode
=
sigTbJoinSplMatchByNode
((
SLogicNode
*
)
pChild
);
if
(
NULL
!=
pSplitNode
)
{
return
pSplitNode
;
}
}
return
NULL
;
}
static
bool
sigTbJoinSplFindSplitNode
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
SSigTbJoinSplitInfo
*
pInfo
)
{
SJoinLogicNode
*
pJoin
=
sigTbJoinSplMatchByNode
(
pSubplan
->
pNode
);
if
(
NULL
!=
pJoin
)
{
pInfo
->
pJoin
=
pJoin
;
pInfo
->
pSplitNode
=
(
SLogicNode
*
)
nodesListGetNode
(
pJoin
->
node
.
pChildren
,
1
);
pInfo
->
pSubplan
=
pSubplan
;
pInfo
->
pSubplan
=
pSubplan
;
return
true
;
}
}
return
NULL
!=
pJoin
;
return
false
;
}
}
static
int32_t
singleTableJoinSplit
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
)
{
static
int32_t
singleTableJoinSplit
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
)
{
...
@@ -753,27 +828,14 @@ typedef struct SUnionAllSplitInfo {
...
@@ -753,27 +828,14 @@ typedef struct SUnionAllSplitInfo {
SLogicSubplan
*
pSubplan
;
SLogicSubplan
*
pSubplan
;
}
SUnionAllSplitInfo
;
}
SUnionAllSplitInfo
;
static
SLogicNode
*
unAllSplMatchByNode
(
SLogicNode
*
pNode
)
{
static
bool
unAllSplFindSplitNode
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
SLogicNode
*
pNode
,
SUnionAllSplitInfo
*
pInfo
)
{
if
(
QUERY_NODE_LOGIC_PLAN_PROJECT
==
nodeType
(
pNode
)
&&
LIST_LENGTH
(
pNode
->
pChildren
)
>
1
)
{
if
(
QUERY_NODE_LOGIC_PLAN_PROJECT
==
nodeType
(
pNode
)
&&
LIST_LENGTH
(
pNode
->
pChildren
)
>
1
)
{
return
pNode
;
pInfo
->
pProject
=
(
SProjectLogicNode
*
)
pNode
;
}
SNode
*
pChild
;
FOREACH
(
pChild
,
pNode
->
pChildren
)
{
SLogicNode
*
pSplitNode
=
unAllSplMatchByNode
((
SLogicNode
*
)
pChild
);
if
(
NULL
!=
pSplitNode
)
{
return
pSplitNode
;
}
}
return
NULL
;
}
static
bool
unAllSplFindSplitNode
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
SUnionAllSplitInfo
*
pInfo
)
{
SLogicNode
*
pSplitNode
=
unAllSplMatchByNode
(
pSubplan
->
pNode
);
if
(
NULL
!=
pSplitNode
)
{
pInfo
->
pProject
=
(
SProjectLogicNode
*
)
pSplitNode
;
pInfo
->
pSubplan
=
pSubplan
;
pInfo
->
pSubplan
=
pSubplan
;
return
true
;
}
}
return
NULL
!=
pSplitNod
e
;
return
fals
e
;
}
}
static
int32_t
unAllSplCreateExchangeNode
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
SProjectLogicNode
*
pProject
)
{
static
int32_t
unAllSplCreateExchangeNode
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
SProjectLogicNode
*
pProject
)
{
...
@@ -828,20 +890,6 @@ typedef struct SUnionDistinctSplitInfo {
...
@@ -828,20 +890,6 @@ typedef struct SUnionDistinctSplitInfo {
SLogicSubplan
*
pSubplan
;
SLogicSubplan
*
pSubplan
;
}
SUnionDistinctSplitInfo
;
}
SUnionDistinctSplitInfo
;
static
SLogicNode
*
unDistSplMatchByNode
(
SLogicNode
*
pNode
)
{
if
(
QUERY_NODE_LOGIC_PLAN_AGG
==
nodeType
(
pNode
)
&&
LIST_LENGTH
(
pNode
->
pChildren
)
>
1
)
{
return
pNode
;
}
SNode
*
pChild
;
FOREACH
(
pChild
,
pNode
->
pChildren
)
{
SLogicNode
*
pSplitNode
=
unDistSplMatchByNode
((
SLogicNode
*
)
pChild
);
if
(
NULL
!=
pSplitNode
)
{
return
pSplitNode
;
}
}
return
NULL
;
}
static
int32_t
unDistSplCreateExchangeNode
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
SAggLogicNode
*
pAgg
)
{
static
int32_t
unDistSplCreateExchangeNode
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
SAggLogicNode
*
pAgg
)
{
SExchangeLogicNode
*
pExchange
=
(
SExchangeLogicNode
*
)
nodesMakeNode
(
QUERY_NODE_LOGIC_PLAN_EXCHANGE
);
SExchangeLogicNode
*
pExchange
=
(
SExchangeLogicNode
*
)
nodesMakeNode
(
QUERY_NODE_LOGIC_PLAN_EXCHANGE
);
if
(
NULL
==
pExchange
)
{
if
(
NULL
==
pExchange
)
{
...
@@ -859,13 +907,14 @@ static int32_t unDistSplCreateExchangeNode(SSplitContext* pCxt, SLogicSubplan* p
...
@@ -859,13 +907,14 @@ static int32_t unDistSplCreateExchangeNode(SSplitContext* pCxt, SLogicSubplan* p
return
nodesListMakeAppend
(
&
pAgg
->
node
.
pChildren
,
(
SNode
*
)
pExchange
);
return
nodesListMakeAppend
(
&
pAgg
->
node
.
pChildren
,
(
SNode
*
)
pExchange
);
}
}
static
bool
unDistSplFindSplitNode
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
S
UnionDistinctSplitInfo
*
pInfo
)
{
static
bool
unDistSplFindSplitNode
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
S
LogicNode
*
pNode
,
SLogicNode
*
pSplitNode
=
unDistSplMatchByNode
(
pSubplan
->
pNode
);
SUnionDistinctSplitInfo
*
pInfo
)
{
if
(
NULL
!=
pSplitNode
)
{
if
(
QUERY_NODE_LOGIC_PLAN_AGG
==
nodeType
(
pNode
)
&&
LIST_LENGTH
(
pNode
->
pChildren
)
>
1
)
{
pInfo
->
pAgg
=
(
SAggLogicNode
*
)
p
Split
Node
;
pInfo
->
pAgg
=
(
SAggLogicNode
*
)
pNode
;
pInfo
->
pSubplan
=
pSubplan
;
pInfo
->
pSubplan
=
pSubplan
;
return
true
;
}
}
return
NULL
!=
pSplitNod
e
;
return
fals
e
;
}
}
static
int32_t
unionDistinctSplit
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
)
{
static
int32_t
unionDistinctSplit
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
)
{
...
@@ -888,27 +937,14 @@ typedef struct SSmaIndexSplitInfo {
...
@@ -888,27 +937,14 @@ typedef struct SSmaIndexSplitInfo {
SLogicSubplan
*
pSubplan
;
SLogicSubplan
*
pSubplan
;
}
SSmaIndexSplitInfo
;
}
SSmaIndexSplitInfo
;
static
SLogicNode
*
smaIdxSplMatchByNode
(
SLogicNode
*
pNode
)
{
static
bool
smaIdxSplFindSplitNode
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
SLogicNode
*
pNode
,
SSmaIndexSplitInfo
*
pInfo
)
{
if
(
QUERY_NODE_LOGIC_PLAN_MERGE
==
nodeType
(
pNode
)
&&
LIST_LENGTH
(
pNode
->
pChildren
)
>
1
)
{
if
(
QUERY_NODE_LOGIC_PLAN_MERGE
==
nodeType
(
pNode
)
&&
LIST_LENGTH
(
pNode
->
pChildren
)
>
1
)
{
return
pNode
;
pInfo
->
pMerge
=
(
SMergeLogicNode
*
)
pNode
;
}
SNode
*
pChild
;
FOREACH
(
pChild
,
pNode
->
pChildren
)
{
SLogicNode
*
pSplitNode
=
smaIdxSplMatchByNode
((
SLogicNode
*
)
pChild
);
if
(
NULL
!=
pSplitNode
)
{
return
pSplitNode
;
}
}
return
NULL
;
}
static
bool
smaIdxSplFindSplitNode
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
SSmaIndexSplitInfo
*
pInfo
)
{
SLogicNode
*
pSplitNode
=
smaIdxSplMatchByNode
(
pSubplan
->
pNode
);
if
(
NULL
!=
pSplitNode
)
{
pInfo
->
pMerge
=
(
SMergeLogicNode
*
)
pSplitNode
;
pInfo
->
pSubplan
=
pSubplan
;
pInfo
->
pSubplan
=
pSubplan
;
return
true
;
}
}
return
NULL
!=
pSplitNod
e
;
return
fals
e
;
}
}
static
int32_t
smaIndexSplit
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
)
{
static
int32_t
smaIndexSplit
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
)
{
...
@@ -926,13 +962,47 @@ static int32_t smaIndexSplit(SSplitContext* pCxt, SLogicSubplan* pSubplan) {
...
@@ -926,13 +962,47 @@ static int32_t smaIndexSplit(SSplitContext* pCxt, SLogicSubplan* pSubplan) {
return
code
;
return
code
;
}
}
typedef
struct
SQnodeSplitInfo
{
SLogicNode
*
pSplitNode
;
SLogicSubplan
*
pSubplan
;
}
SQnodeSplitInfo
;
static
bool
qndSplFindSplitNode
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
SLogicNode
*
pNode
,
SQnodeSplitInfo
*
pInfo
)
{
if
(
QUERY_NODE_LOGIC_PLAN_SCAN
==
nodeType
(
pNode
)
&&
NULL
!=
pNode
->
pParent
)
{
pInfo
->
pSplitNode
=
pNode
;
pInfo
->
pSubplan
=
pSubplan
;
return
true
;
}
return
false
;
}
static
int32_t
qnodeSplit
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
)
{
if
(
QUERY_POLICY_QNODE
!=
tsQueryPolicy
)
{
return
TSDB_CODE_SUCCESS
;
}
SQnodeSplitInfo
info
=
{
0
};
if
(
!
splMatch
(
pCxt
,
pSubplan
,
0
,
(
FSplFindSplitNode
)
qndSplFindSplitNode
,
&
info
))
{
return
TSDB_CODE_SUCCESS
;
}
int32_t
code
=
splCreateExchangeNodeForSubplan
(
pCxt
,
info
.
pSubplan
,
info
.
pSplitNode
,
info
.
pSubplan
->
subplanType
);
if
(
TSDB_CODE_SUCCESS
==
code
)
{
code
=
nodesListMakeStrictAppend
(
&
info
.
pSubplan
->
pChildren
,
(
SNode
*
)
splCreateScanSubplan
(
pCxt
,
info
.
pSplitNode
,
0
));
}
++
(
pCxt
->
groupId
);
pCxt
->
split
=
true
;
return
code
;
}
// clang-format off
// clang-format off
static
const
SSplitRule
splitRuleSet
[]
=
{
static
const
SSplitRule
splitRuleSet
[]
=
{
{.
pName
=
"SuperTableSplit"
,
.
splitFunc
=
stableSplit
},
{.
pName
=
"SuperTableSplit"
,
.
splitFunc
=
stableSplit
},
{.
pName
=
"SingleTableJoinSplit"
,
.
splitFunc
=
singleTableJoinSplit
},
{.
pName
=
"SingleTableJoinSplit"
,
.
splitFunc
=
singleTableJoinSplit
},
{.
pName
=
"UnionAllSplit"
,
.
splitFunc
=
unionAllSplit
},
{.
pName
=
"UnionAllSplit"
,
.
splitFunc
=
unionAllSplit
},
{.
pName
=
"UnionDistinctSplit"
,
.
splitFunc
=
unionDistinctSplit
},
{.
pName
=
"UnionDistinctSplit"
,
.
splitFunc
=
unionDistinctSplit
},
{.
pName
=
"SmaIndexSplit"
,
.
splitFunc
=
smaIndexSplit
}
{.
pName
=
"SmaIndexSplit"
,
.
splitFunc
=
smaIndexSplit
},
{.
pName
=
"QnodeSplit"
,
.
splitFunc
=
qnodeSplit
}
};
};
// clang-format on
// clang-format on
...
...
source/libs/planner/test/planGroupByTest.cpp
浏览文件 @
2b6cbbf7
...
@@ -83,5 +83,7 @@ TEST_F(PlanGroupByTest, stable) {
...
@@ -83,5 +83,7 @@ TEST_F(PlanGroupByTest, stable) {
run
(
"SELECT COUNT(*) FROM st1 GROUP BY c1"
);
run
(
"SELECT COUNT(*) FROM st1 GROUP BY c1"
);
run
(
"SELECT COUNT(*) FROM st1 PARTITION BY c2 GROUP BY c1"
);
run
(
"SELECT SUM(c1) FROM st1 GROUP BY c2 HAVING SUM(c1) IS NOT NULL"
);
run
(
"SELECT SUM(c1) FROM st1 GROUP BY c2 HAVING SUM(c1) IS NOT NULL"
);
}
}
source/libs/planner/test/planIntervalTest.cpp
浏览文件 @
2b6cbbf7
...
@@ -60,4 +60,6 @@ TEST_F(PlanIntervalTest, stable) {
...
@@ -60,4 +60,6 @@ TEST_F(PlanIntervalTest, stable) {
run
(
"SELECT COUNT(*) FROM st1 INTERVAL(10s)"
);
run
(
"SELECT COUNT(*) FROM st1 INTERVAL(10s)"
);
run
(
"SELECT _WSTARTTS, COUNT(*) FROM st1 INTERVAL(10s)"
);
run
(
"SELECT _WSTARTTS, COUNT(*) FROM st1 INTERVAL(10s)"
);
run
(
"SELECT _WSTARTTS, COUNT(*) FROM st1 PARTITION BY TBNAME INTERVAL(10s)"
);
}
}
source/libs/planner/test/planJoinTest.cpp
浏览文件 @
2b6cbbf7
...
@@ -50,3 +50,9 @@ TEST_F(PlanJoinTest, multiJoin) {
...
@@ -50,3 +50,9 @@ TEST_F(PlanJoinTest, multiJoin) {
run
(
"SELECT t1.c1, t2.c1 FROM st1s1 t1 JOIN st1s2 t2 ON t1.ts = t2.ts JOIN st1s3 t3 ON t1.ts = t3.ts"
);
run
(
"SELECT t1.c1, t2.c1 FROM st1s1 t1 JOIN st1s2 t2 ON t1.ts = t2.ts JOIN st1s3 t3 ON t1.ts = t3.ts"
);
}
}
TEST_F
(
PlanJoinTest
,
stable
)
{
useDb
(
"root"
,
"test"
);
run
(
"SELECT t1.c1, t2.c1 FROM st1 t1 JOIN st2 t2 ON t1.ts = t2.ts "
);
}
source/libs/planner/test/planOrderByTest.cpp
浏览文件 @
2b6cbbf7
...
@@ -49,4 +49,6 @@ TEST_F(PlanOrderByTest, stable) {
...
@@ -49,4 +49,6 @@ TEST_F(PlanOrderByTest, stable) {
// ORDER BY key is not in the projection list
// ORDER BY key is not in the projection list
run
(
"SELECT c2 FROM st1 ORDER BY c1"
);
run
(
"SELECT c2 FROM st1 ORDER BY c1"
);
run
(
"SELECT c2 FROM st1 PARTITION BY c2 ORDER BY c1"
);
}
}
source/libs/planner/test/planOtherTest.cpp
浏览文件 @
2b6cbbf7
...
@@ -83,3 +83,10 @@ TEST_F(PlanOtherTest, delete) {
...
@@ -83,3 +83,10 @@ TEST_F(PlanOtherTest, delete) {
run
(
"DELETE FROM st1 WHERE ts > now - 2d and ts < now - 1d AND tag1 = 10"
);
run
(
"DELETE FROM st1 WHERE ts > now - 2d and ts < now - 1d AND tag1 = 10"
);
}
}
TEST_F
(
PlanOtherTest
,
queryPolicy
)
{
useDb
(
"root"
,
"test"
);
tsQueryPolicy
=
QUERY_POLICY_QNODE
;
run
(
"SELECT COUNT(*) FROM st1"
);
}
source/libs/planner/test/planTestUtil.h
浏览文件 @
2b6cbbf7
...
@@ -18,6 +18,10 @@
...
@@ -18,6 +18,10 @@
#include <gtest/gtest.h>
#include <gtest/gtest.h>
#define ALLOW_FORBID_FUNC
#include "planInt.h"
class
PlannerTestBaseImpl
;
class
PlannerTestBaseImpl
;
struct
TAOS_MULTI_BIND
;
struct
TAOS_MULTI_BIND
;
...
...
tests/script/jenkins/basic.txt
浏览文件 @
2b6cbbf7
...
@@ -132,7 +132,7 @@
...
@@ -132,7 +132,7 @@
#./test.sh -f tsim/mnode/basic1.sim -m
#./test.sh -f tsim/mnode/basic1.sim -m
# --- sma
# --- sma
#
./test.sh -f tsim/sma/tsmaCreateInsertData.sim
./test.sh -f tsim/sma/tsmaCreateInsertData.sim
./test.sh -f tsim/sma/rsmaCreateInsertQuery.sim
./test.sh -f tsim/sma/rsmaCreateInsertQuery.sim
# --- valgrind
# --- valgrind
...
...
tests/script/tsim/sma/tsmaCreateInsertData.sim
浏览文件 @
2b6cbbf7
...
@@ -37,6 +37,14 @@ print =============== trigger stream to execute sma aggr task and insert sma dat
...
@@ -37,6 +37,14 @@ print =============== trigger stream to execute sma aggr task and insert sma dat
sql insert into ct1 values(now+5s, 20, 20.0, 30.0)
sql insert into ct1 values(now+5s, 20, 20.0, 30.0)
#===================================================================
#===================================================================
print =============== show streams ================================
sql show streams;
print $data00 $data01 $data02
if $data00 != d1 then
return -1
endi
print =============== select * from ct1 from memory
print =============== select * from ct1 from memory
sql select * from ct1;
sql select * from ct1;
print $data00 $data01
print $data00 $data01
...
...
tests/system-test/6-cluster/5dnode3mnodeDrop.py
浏览文件 @
2b6cbbf7
...
@@ -230,7 +230,7 @@ class TDTestCase:
...
@@ -230,7 +230,7 @@ class TDTestCase:
def
five_dnode_three_mnode
(
self
,
dnodenumber
):
def
five_dnode_three_mnode
(
self
):
tdSql
.
query
(
"show dnodes;"
)
tdSql
.
query
(
"show dnodes;"
)
tdSql
.
checkData
(
0
,
1
,
'%s:6030'
%
self
.
host
)
tdSql
.
checkData
(
0
,
1
,
'%s:6030'
%
self
.
host
)
tdSql
.
checkData
(
4
,
1
,
'%s:6430'
%
self
.
host
)
tdSql
.
checkData
(
4
,
1
,
'%s:6430'
%
self
.
host
)
...
@@ -260,7 +260,9 @@ class TDTestCase:
...
@@ -260,7 +260,9 @@ class TDTestCase:
dropcount
=
0
dropcount
=
0
while
dropcount
<=
10
:
while
dropcount
<=
10
:
for
i
in
range
(
1
,
3
):
for
i
in
range
(
1
,
3
):
tdLog
.
debug
(
"drop mnode on dnode %d"
%
(
i
+
1
))
tdSql
.
execute
(
"drop mnode on dnode %d"
%
(
i
+
1
))
tdSql
.
execute
(
"drop mnode on dnode %d"
%
(
i
+
1
))
tdLog
.
debug
(
"create mnode on dnode %d"
%
(
i
+
1
))
tdSql
.
execute
(
"create mnode on dnode %d"
%
(
i
+
1
))
tdSql
.
execute
(
"create mnode on dnode %d"
%
(
i
+
1
))
dropcount
+=
1
dropcount
+=
1
self
.
check3mnode
()
self
.
check3mnode
()
...
@@ -276,7 +278,7 @@ class TDTestCase:
...
@@ -276,7 +278,7 @@ class TDTestCase:
def
run
(
self
):
def
run
(
self
):
# print(self.master_dnode.cfgDict)
# print(self.master_dnode.cfgDict)
self
.
buildcluster
(
5
)
self
.
buildcluster
(
5
)
self
.
five_dnode_three_mnode
(
5
)
self
.
five_dnode_three_mnode
()
def
stop
(
self
):
def
stop
(
self
):
tdSql
.
close
()
tdSql
.
close
()
...
...
tests/system-test/6-cluster/5dnode3mnodeStop.py
浏览文件 @
2b6cbbf7
...
@@ -145,6 +145,7 @@ class TDTestCase:
...
@@ -145,6 +145,7 @@ class TDTestCase:
tdSql
.
checkData
(
2
,
3
,
'ready'
)
tdSql
.
checkData
(
2
,
3
,
'ready'
)
def
check3mnode1off
(
self
):
def
check3mnode1off
(
self
):
tdSql
.
error
(
"drop mnode on dnode 1;"
)
count
=
0
count
=
0
while
count
<
10
:
while
count
<
10
:
time
.
sleep
(
1
)
time
.
sleep
(
1
)
...
@@ -174,6 +175,7 @@ class TDTestCase:
...
@@ -174,6 +175,7 @@ class TDTestCase:
tdSql
.
checkData
(
2
,
3
,
'ready'
)
tdSql
.
checkData
(
2
,
3
,
'ready'
)
def
check3mnode2off
(
self
):
def
check3mnode2off
(
self
):
tdSql
.
error
(
"drop mnode on dnode 2;"
)
count
=
0
count
=
0
while
count
<
10
:
while
count
<
10
:
time
.
sleep
(
1
)
time
.
sleep
(
1
)
...
@@ -201,6 +203,7 @@ class TDTestCase:
...
@@ -201,6 +203,7 @@ class TDTestCase:
tdSql
.
checkData
(
2
,
3
,
'ready'
)
tdSql
.
checkData
(
2
,
3
,
'ready'
)
def
check3mnode3off
(
self
):
def
check3mnode3off
(
self
):
tdSql
.
error
(
"drop mnode on dnode 3;"
)
count
=
0
count
=
0
while
count
<
10
:
while
count
<
10
:
time
.
sleep
(
1
)
time
.
sleep
(
1
)
...
@@ -255,17 +258,17 @@ class TDTestCase:
...
@@ -255,17 +258,17 @@ class TDTestCase:
print
(
tdSql
.
queryResult
)
print
(
tdSql
.
queryResult
)
tdLog
.
debug
(
"stop and follower of mnode"
)
tdLog
.
debug
(
"stop and follower of mnode"
)
#
self.TDDnodes.stoptaosd(2)
self
.
TDDnodes
.
stoptaosd
(
2
)
#
self.check3mnode2off()
self
.
check3mnode2off
()
#
self.TDDnodes.starttaosd(2)
self
.
TDDnodes
.
starttaosd
(
2
)
#
self.TDDnodes.stoptaosd(3)
self
.
TDDnodes
.
stoptaosd
(
3
)
#
self.check3mnode3off()
self
.
check3mnode3off
()
#
self.TDDnodes.starttaosd(2)
self
.
TDDnodes
.
starttaosd
(
2
)
#
self.TDDnodes.stoptaosd(1)
self
.
TDDnodes
.
stoptaosd
(
1
)
#
self.check3mnode1off()
self
.
check3mnode1off
()
#
self.TDDnodes.starttaosd(1)
self
.
TDDnodes
.
starttaosd
(
1
)
# self.check3mnode()
# self.check3mnode()
stopcount
=
0
stopcount
=
0
...
...
tests/system-test/6-cluster/5dnode3mnodeStopInsert.py
浏览文件 @
2b6cbbf7
...
@@ -12,7 +12,8 @@ from util.dnodes import TDDnode
...
@@ -12,7 +12,8 @@ from util.dnodes import TDDnode
import
time
import
time
import
socket
import
socket
import
subprocess
import
subprocess
from
multiprocessing
import
Process
import
threading
as
thd
class
MyDnodes
(
TDDnodes
):
class
MyDnodes
(
TDDnodes
):
def
__init__
(
self
,
dnodes_lists
):
def
__init__
(
self
,
dnodes_lists
):
super
(
MyDnodes
,
self
).
__init__
()
super
(
MyDnodes
,
self
).
__init__
()
...
@@ -49,10 +50,10 @@ class TDTestCase:
...
@@ -49,10 +50,10 @@ class TDTestCase:
buildPath
=
root
[:
len
(
root
)
-
len
(
"/build/bin"
)]
buildPath
=
root
[:
len
(
root
)
-
len
(
"/build/bin"
)]
break
break
return
buildPath
return
buildPath
def
insert_data
(
self
,
count
):
def
insert_data
(
self
,
count
start
,
countstop
):
# fisrt add data : db\stable\childtable\general table
# fisrt add data : db\stable\childtable\general table
for
couti
in
count
:
for
couti
in
range
(
countstart
,
countstop
)
:
tdSql
.
execute
(
"drop database if exists db%d"
%
couti
)
tdSql
.
execute
(
"drop database if exists db%d"
%
couti
)
tdSql
.
execute
(
"create database if not exists db%d replica 1 days 300"
%
couti
)
tdSql
.
execute
(
"create database if not exists db%d replica 1 days 300"
%
couti
)
tdSql
.
execute
(
"use db%d"
%
couti
)
tdSql
.
execute
(
"use db%d"
%
couti
)
...
@@ -258,6 +259,11 @@ class TDTestCase:
...
@@ -258,6 +259,11 @@ class TDTestCase:
stopcount
=
0
stopcount
=
0
while
stopcount
<=
2
:
while
stopcount
<=
2
:
for
i
in
range
(
dnodenumber
):
for
i
in
range
(
dnodenumber
):
threads
=
[]
threads
.
append
(
thd
.
Thread
(
target
=
self
.
insert_data
,
args
=
(
i
*
2
,
i
*
2
+
2
)))
# start_time = time.time()
threads
[
0
].
start
()
# end_time = time.time()
self
.
TDDnodes
.
stoptaosd
(
i
+
1
)
self
.
TDDnodes
.
stoptaosd
(
i
+
1
)
# if i == 1 :
# if i == 1 :
# self.check3mnode2off()
# self.check3mnode2off()
...
@@ -265,13 +271,12 @@ class TDTestCase:
...
@@ -265,13 +271,12 @@ class TDTestCase:
# self.check3mnode3off()
# self.check3mnode3off()
# elif i == 0:
# elif i == 0:
# self.check3mnode1off()
# self.check3mnode1off()
self
.
TDDnodes
.
starttaosd
(
i
+
1
)
self
.
TDDnodes
.
starttaosd
(
i
+
1
)
threads
[
0
].
join
()
# self.check3mnode()
# self.check3mnode()
stopcount
+=
1
stopcount
+=
1
self
.
check3mnode
()
self
.
check3mnode
()
def
getConnection
(
self
,
dnode
):
def
getConnection
(
self
,
dnode
):
host
=
dnode
.
cfgDict
[
"fqdn"
]
host
=
dnode
.
cfgDict
[
"fqdn"
]
port
=
dnode
.
cfgDict
[
"serverPort"
]
port
=
dnode
.
cfgDict
[
"serverPort"
]
...
...
tests/system-test/7-tmq/tmqError.py
0 → 100644
浏览文件 @
2b6cbbf7
import
taos
import
sys
import
time
import
socket
import
os
import
threading
from
enum
import
Enum
from
util.log
import
*
from
util.sql
import
*
from
util.cases
import
*
from
util.dnodes
import
*
class
actionType
(
Enum
):
CREATE_DATABASE
=
0
CREATE_STABLE
=
1
CREATE_CTABLE
=
2
INSERT_DATA
=
3
class
TDTestCase
:
hostname
=
socket
.
gethostname
()
#rpcDebugFlagVal = '143'
#clientCfgDict = {'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''}
#clientCfgDict["rpcDebugFlag"] = rpcDebugFlagVal
#updatecfgDict = {'clientCfg': {}, 'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''}
#updatecfgDict["rpcDebugFlag"] = rpcDebugFlagVal
#print ("===================: ", updatecfgDict)
def
init
(
self
,
conn
,
logSql
):
tdLog
.
debug
(
f
"start to excute
{
__file__
}
"
)
tdSql
.
init
(
conn
.
cursor
())
#tdSql.init(conn.cursor(), logSql) # output sql.txt file
def
getBuildPath
(
self
):
selfPath
=
os
.
path
.
dirname
(
os
.
path
.
realpath
(
__file__
))
if
(
"community"
in
selfPath
):
projPath
=
selfPath
[:
selfPath
.
find
(
"community"
)]
else
:
projPath
=
selfPath
[:
selfPath
.
find
(
"tests"
)]
for
root
,
dirs
,
files
in
os
.
walk
(
projPath
):
if
(
"taosd"
in
files
or
"taosd.exe"
in
files
):
rootRealPath
=
os
.
path
.
dirname
(
os
.
path
.
realpath
(
root
))
if
(
"packaging"
not
in
rootRealPath
):
buildPath
=
root
[:
len
(
root
)
-
len
(
"/build/bin"
)]
break
return
buildPath
def
newcur
(
self
,
cfg
,
host
,
port
):
user
=
"root"
password
=
"taosdata"
con
=
taos
.
connect
(
host
=
host
,
user
=
user
,
password
=
password
,
config
=
cfg
,
port
=
port
)
cur
=
con
.
cursor
()
print
(
cur
)
return
cur
def
initConsumerTable
(
self
,
cdbName
=
'cdb'
):
tdLog
.
info
(
"create consume database, and consume info table, and consume result table"
)
tdSql
.
query
(
"create database if not exists %s vgroups 1"
%
(
cdbName
))
# tdSql.query("drop table if exists %s.consumeinfo "%(cdbName))
# tdSql.query("drop table if exists %s.consumeresult "%(cdbName))
tdSql
.
query
(
"create table %s.consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)"
%
cdbName
)
tdSql
.
query
(
"create table %s.consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)"
%
cdbName
)
def
initConsumerInfoTable
(
self
,
cdbName
=
'cdb'
):
tdLog
.
info
(
"drop consumeinfo table"
)
tdSql
.
query
(
"drop table if exists %s.consumeinfo "
%
(
cdbName
))
tdSql
.
query
(
"create table %s.consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)"
%
cdbName
)
def
insertConsumerInfo
(
self
,
consumerId
,
expectrowcnt
,
topicList
,
keyList
,
ifcheckdata
,
ifmanualcommit
,
cdbName
=
'cdb'
):
sql
=
"insert into %s.consumeinfo values "
%
cdbName
sql
+=
"(now, %d, '%s', '%s', %d, %d, %d)"
%
(
consumerId
,
topicList
,
keyList
,
expectrowcnt
,
ifcheckdata
,
ifmanualcommit
)
tdLog
.
info
(
"consume info sql: %s"
%
sql
)
tdSql
.
query
(
sql
)
def
selectConsumeResult
(
self
,
expectRows
,
cdbName
=
'cdb'
):
resultList
=
[]
while
1
:
tdSql
.
query
(
"select * from %s.consumeresult"
%
cdbName
)
#tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3))
if
tdSql
.
getRows
()
==
expectRows
:
break
else
:
time
.
sleep
(
5
)
for
i
in
range
(
expectRows
):
tdLog
.
info
(
"consume id: %d, consume msgs: %d, consume rows: %d"
%
(
tdSql
.
getData
(
i
,
1
),
tdSql
.
getData
(
i
,
2
),
tdSql
.
getData
(
i
,
3
)))
resultList
.
append
(
tdSql
.
getData
(
i
,
3
))
return
resultList
def
startTmqSimProcess
(
self
,
buildPath
,
cfgPath
,
pollDelay
,
dbName
,
showMsg
=
1
,
showRow
=
1
,
cdbName
=
'cdb'
,
valgrind
=
0
):
if
valgrind
==
1
:
logFile
=
cfgPath
+
'/../log/valgrind-tmq.log'
shellCmd
=
'nohup valgrind --log-file='
+
logFile
shellCmd
+=
'--tool=memcheck --leak-check=full --show-reachable=no --track-origins=yes --show-leak-kinds=all --num-callers=20 -v --workaround-gcc296-bugs=yes '
if
(
platform
.
system
().
lower
()
==
'windows'
):
shellCmd
=
'mintty -h never -w hide '
+
buildPath
+
'
\\
build
\\
bin
\\
tmq_sim.exe -c '
+
cfgPath
shellCmd
+=
" -y %d -d %s -g %d -r %d -w %s "
%
(
pollDelay
,
dbName
,
showMsg
,
showRow
,
cdbName
)
shellCmd
+=
"> nul 2>&1 &"
else
:
shellCmd
=
'nohup '
+
buildPath
+
'/build/bin/tmq_sim -c '
+
cfgPath
shellCmd
+=
" -y %d -d %s -g %d -r %d -w %s "
%
(
pollDelay
,
dbName
,
showMsg
,
showRow
,
cdbName
)
shellCmd
+=
"> /dev/null 2>&1 &"
tdLog
.
info
(
shellCmd
)
os
.
system
(
shellCmd
)
def
create_database
(
self
,
tsql
,
dbName
,
dropFlag
=
1
,
vgroups
=
4
,
replica
=
1
):
if
dropFlag
==
1
:
tsql
.
execute
(
"drop database if exists %s"
%
(
dbName
))
tsql
.
execute
(
"create database if not exists %s vgroups %d replica %d"
%
(
dbName
,
vgroups
,
replica
))
tdLog
.
debug
(
"complete to create database %s"
%
(
dbName
))
return
def
create_stable
(
self
,
tsql
,
dbName
,
stbName
):
tsql
.
execute
(
"create table if not exists %s.%s (ts timestamp, c1 bigint, c2 binary(16)) tags(t1 int)"
%
(
dbName
,
stbName
))
tdLog
.
debug
(
"complete to create %s.%s"
%
(
dbName
,
stbName
))
return
def
create_ctables
(
self
,
tsql
,
dbName
,
stbName
,
ctbNum
):
tsql
.
execute
(
"use %s"
%
dbName
)
pre_create
=
"create table"
sql
=
pre_create
#tdLog.debug("doing create one stable %s and %d child table in %s ..." %(stbname, count ,dbname))
for
i
in
range
(
ctbNum
):
sql
+=
" %s_%d using %s tags(%d)"
%
(
stbName
,
i
,
stbName
,
i
+
1
)
if
(
i
>
0
)
and
(
i
%
100
==
0
):
tsql
.
execute
(
sql
)
sql
=
pre_create
if
sql
!=
pre_create
:
tsql
.
execute
(
sql
)
tdLog
.
debug
(
"complete to create %d child tables in %s.%s"
%
(
ctbNum
,
dbName
,
stbName
))
return
def
insert_data
(
self
,
tsql
,
dbName
,
stbName
,
ctbNum
,
rowsPerTbl
,
batchNum
,
startTs
=
0
):
tdLog
.
debug
(
"start to insert data ............"
)
tsql
.
execute
(
"use %s"
%
dbName
)
pre_insert
=
"insert into "
sql
=
pre_insert
if
startTs
==
0
:
t
=
time
.
time
()
startTs
=
int
(
round
(
t
*
1000
))
#tdLog.debug("doing insert data into stable:%s rows:%d ..."%(stbName, allRows))
rowsOfSql
=
0
for
i
in
range
(
ctbNum
):
sql
+=
" %s_%d values "
%
(
stbName
,
i
)
for
j
in
range
(
rowsPerTbl
):
sql
+=
"(%d, %d, 'tmqrow_%d') "
%
(
startTs
+
j
,
j
,
j
)
rowsOfSql
+=
1
if
(
j
>
0
)
and
((
rowsOfSql
==
batchNum
)
or
(
j
==
rowsPerTbl
-
1
)):
tsql
.
execute
(
sql
)
rowsOfSql
=
0
if
j
<
rowsPerTbl
-
1
:
sql
=
"insert into %s_%d values "
%
(
stbName
,
i
)
else
:
sql
=
"insert into "
#end sql
if
sql
!=
pre_insert
:
#print("insert sql:%s"%sql)
tsql
.
execute
(
sql
)
tdLog
.
debug
(
"insert data ............ [OK]"
)
return
def
prepareEnv
(
self
,
**
parameterDict
):
# create new connector for my thread
tsql
=
self
.
newcur
(
parameterDict
[
'cfg'
],
'localhost'
,
6030
)
if
parameterDict
[
"actionType"
]
==
actionType
.
CREATE_DATABASE
:
self
.
create_database
(
tsql
,
parameterDict
[
"dbName"
])
elif
parameterDict
[
"actionType"
]
==
actionType
.
CREATE_STABLE
:
self
.
create_stable
(
tsql
,
parameterDict
[
"dbName"
],
parameterDict
[
"stbName"
])
elif
parameterDict
[
"actionType"
]
==
actionType
.
CREATE_CTABLE
:
self
.
create_ctables
(
tsql
,
parameterDict
[
"dbName"
],
parameterDict
[
"stbName"
],
parameterDict
[
"ctbNum"
])
elif
parameterDict
[
"actionType"
]
==
actionType
.
INSERT_DATA
:
self
.
insert_data
(
tsql
,
parameterDict
[
"dbName"
],
parameterDict
[
"stbName"
],
parameterDict
[
"ctbNum"
],
\
parameterDict
[
"rowsPerTbl"
],
parameterDict
[
"batchNum"
])
else
:
tdLog
.
exit
(
"not support's action: "
,
parameterDict
[
"actionType"
])
return
def
tmqCase1
(
self
,
cfgPath
,
buildPath
):
'''
Leave a TMQ process. Stop taosd, delete the data directory, restart taosd,
and restart a consumption process to complete a consumption
'''
tdLog
.
printNoPrefix
(
"======== test case 1: "
)
self
.
initConsumerTable
()
# create and start thread
parameterDict
=
{
'cfg'
:
''
,
\
'actionType'
:
0
,
\
'dbName'
:
'db3'
,
\
'dropFlag'
:
1
,
\
'vgroups'
:
4
,
\
'replica'
:
1
,
\
'stbName'
:
'stb1'
,
\
'ctbNum'
:
10
,
\
'rowsPerTbl'
:
20000
,
\
'batchNum'
:
100
,
\
'startTs'
:
1640966400000
}
# 2022-01-01 00:00:00.000
parameterDict
[
'cfg'
]
=
cfgPath
self
.
create_database
(
tdSql
,
parameterDict
[
"dbName"
])
self
.
create_stable
(
tdSql
,
parameterDict
[
"dbName"
],
parameterDict
[
"stbName"
])
self
.
create_ctables
(
tdSql
,
parameterDict
[
"dbName"
],
parameterDict
[
"stbName"
],
parameterDict
[
"ctbNum"
])
self
.
insert_data
(
tdSql
,
parameterDict
[
"dbName"
],
parameterDict
[
"stbName"
],
parameterDict
[
"ctbNum"
],
parameterDict
[
"rowsPerTbl"
],
parameterDict
[
"batchNum"
])
tdLog
.
info
(
"create topics from stb1"
)
topicFromStb1
=
'topic_stb1'
tdSql
.
execute
(
"create topic %s as select ts, c1, c2 from %s.%s"
%
(
topicFromStb1
,
parameterDict
[
'dbName'
],
parameterDict
[
'stbName'
]))
consumerId
=
0
# expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"]
expectrowcnt
=
90000000000
topicList
=
topicFromStb1
ifcheckdata
=
0
ifManualCommit
=
0
keyList
=
'group.id:cgrp1,\
enable.auto.commit:false,\
auto.commit.interval.ms:6000,\
auto.offset.reset:earliest'
self
.
insertConsumerInfo
(
consumerId
,
expectrowcnt
,
topicList
,
keyList
,
ifcheckdata
,
ifManualCommit
)
tdLog
.
info
(
"start consume processor"
)
pollDelay
=
9000000
# Forever loop
showMsg
=
1
showRow
=
1
self
.
startTmqSimProcess
(
buildPath
,
cfgPath
,
pollDelay
,
parameterDict
[
"dbName"
],
showMsg
,
showRow
)
time
.
sleep
(
3
)
tdLog
.
info
(
"================= stop dnode, and remove data file, then start dnode ==========================="
)
tdDnodes
.
stop
(
1
)
# time.sleep(5)
dataPath
=
buildPath
+
"/../sim/dnode1/data/*"
shellCmd
=
'rm -rf '
+
dataPath
tdLog
.
info
(
shellCmd
)
os
.
system
(
shellCmd
)
tdDnodes
.
start
(
1
)
time
.
sleep
(
2
)
######### redo to consume
self
.
initConsumerTable
()
self
.
create_database
(
tdSql
,
parameterDict
[
"dbName"
])
self
.
create_stable
(
tdSql
,
parameterDict
[
"dbName"
],
parameterDict
[
"stbName"
])
self
.
create_ctables
(
tdSql
,
parameterDict
[
"dbName"
],
parameterDict
[
"stbName"
],
parameterDict
[
"ctbNum"
])
self
.
insert_data
(
tdSql
,
parameterDict
[
"dbName"
],
parameterDict
[
"stbName"
],
parameterDict
[
"ctbNum"
],
parameterDict
[
"rowsPerTbl"
],
parameterDict
[
"batchNum"
])
tdLog
.
info
(
"create topics from stb1"
)
topicFromStb1
=
'topic_stb1'
tdSql
.
execute
(
"create topic %s as select ts, c1, c2 from %s.%s"
%
(
topicFromStb1
,
parameterDict
[
'dbName'
],
parameterDict
[
'stbName'
]))
consumerId
=
0
expectrowcnt
=
parameterDict
[
"rowsPerTbl"
]
*
parameterDict
[
"ctbNum"
]
topicList
=
topicFromStb1
ifcheckdata
=
0
ifManualCommit
=
0
keyList
=
'group.id:cgrp1,\
enable.auto.commit:false,\
auto.commit.interval.ms:6000,\
auto.offset.reset:earliest'
self
.
insertConsumerInfo
(
consumerId
,
expectrowcnt
,
topicList
,
keyList
,
ifcheckdata
,
ifManualCommit
)
tdLog
.
info
(
"start consume processor"
)
pollDelay
=
20
showMsg
=
1
showRow
=
1
self
.
startTmqSimProcess
(
buildPath
,
cfgPath
,
pollDelay
,
parameterDict
[
"dbName"
],
showMsg
,
showRow
)
expectRows
=
1
resultList
=
self
.
selectConsumeResult
(
expectRows
)
totalConsumeRows
=
0
for
i
in
range
(
expectRows
):
totalConsumeRows
+=
resultList
[
i
]
tdLog
.
info
(
"act consume rows: %d, expect consume rows: %d"
%
(
totalConsumeRows
,
expectrowcnt
))
if
not
(
totalConsumeRows
==
expectrowcnt
):
tdLog
.
exit
(
"tmq consume rows error!"
)
tdSql
.
query
(
"drop topic %s"
%
topicFromStb1
)
os
.
system
(
'pkill tmq_sim'
)
tdLog
.
printNoPrefix
(
"======== test case 1 end ...... "
)
def
run
(
self
):
tdSql
.
prepare
()
buildPath
=
self
.
getBuildPath
()
if
(
buildPath
==
""
):
tdLog
.
exit
(
"taosd not found!"
)
else
:
tdLog
.
info
(
"taosd found in %s"
%
buildPath
)
cfgPath
=
buildPath
+
"/../sim/psim/cfg"
tdLog
.
info
(
"cfgPath: %s"
%
cfgPath
)
self
.
tmqCase1
(
cfgPath
,
buildPath
)
def
stop
(
self
):
tdSql
.
close
()
tdLog
.
success
(
f
"
{
__file__
}
successfully executed"
)
event
=
threading
.
Event
()
tdCases
.
addLinux
(
__file__
,
TDTestCase
())
tdCases
.
addWindows
(
__file__
,
TDTestCase
())
tests/system-test/fulltest.sh
浏览文件 @
2b6cbbf7
...
@@ -18,7 +18,7 @@ python3 ./test.py -f 0-others/fsync.py
...
@@ -18,7 +18,7 @@ python3 ./test.py -f 0-others/fsync.py
python3 ./test.py
-f
1-insert/influxdb_line_taosc_insert.py
python3 ./test.py
-f
1-insert/influxdb_line_taosc_insert.py
python3 ./test.py
-f
1-insert/opentsdb_telnet_line_taosc_insert.py
python3 ./test.py
-f
1-insert/opentsdb_telnet_line_taosc_insert.py
python3 ./test.py
-f
1-insert/opentsdb_json_taosc_insert.py
python3 ./test.py
-f
1-insert/opentsdb_json_taosc_insert.py
# python3 ./test.py -f 1-insert/test_stmt_muti_insert_query.py
#
BUG
python3 ./test.py -f 1-insert/test_stmt_muti_insert_query.py
python3 ./test.py
-f
1-insert/alter_stable.py
python3 ./test.py
-f
1-insert/alter_stable.py
python3 ./test.py
-f
1-insert/alter_table.py
python3 ./test.py
-f
1-insert/alter_table.py
python3 ./test.py
-f
1-insert/insertWithMoreVgroup.py
python3 ./test.py
-f
1-insert/insertWithMoreVgroup.py
...
@@ -101,7 +101,9 @@ python3 ./test.py -f 2-query/tail.py
...
@@ -101,7 +101,9 @@ python3 ./test.py -f 2-query/tail.py
python3 ./test.py
-f
6-cluster/5dnode1mnode.py
python3 ./test.py
-f
6-cluster/5dnode1mnode.py
python3 ./test.py
-f
6-cluster/5dnode2mnode.py
python3 ./test.py
-f
6-cluster/5dnode2mnode.py
python3 ./test.py
-f
6-cluster/5dnode3mnodeStop.py
# BUG python3 ./test.py -f 6-cluster/5dnode3mnodeStop.py
python3 ./test.py
-f
6-cluster/5dnode3mnodeDrop.py
# BUG python3 ./test.py -f 6-cluster/5dnode3mnodeStopInsert.py
python3 ./test.py
-f
7-tmq/basic5.py
python3 ./test.py
-f
7-tmq/basic5.py
python3 ./test.py
-f
7-tmq/subscribeDb.py
python3 ./test.py
-f
7-tmq/subscribeDb.py
...
@@ -114,3 +116,4 @@ python3 ./test.py -f 7-tmq/subscribeStb2.py
...
@@ -114,3 +116,4 @@ python3 ./test.py -f 7-tmq/subscribeStb2.py
python3 ./test.py
-f
7-tmq/subscribeStb3.py
python3 ./test.py
-f
7-tmq/subscribeStb3.py
python3 ./test.py
-f
7-tmq/subscribeStb4.py
python3 ./test.py
-f
7-tmq/subscribeStb4.py
python3 ./test.py
-f
7-tmq/db.py
python3 ./test.py
-f
7-tmq/db.py
python3 ./test.py
-f
7-tmq/tmqError.py
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录