Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
93962a8c
T
TDengine
项目概览
taosdata
/
TDengine
11 个月 前同步成功
通知
1179
Star
22014
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,发现更多精彩内容 >>
提交
93962a8c
编写于
7月 05, 2022
作者:
S
Shengliang Guan
浏览文件
操作
浏览文件
下载
差异文件
Merge remote-tracking branch 'origin/3.0' into fix/mnode
上级
9aa870e5
cf9a6dea
变更
20
隐藏空白更改
内联
并排
Showing
20 changed file
with
287 addition
and
128 deletion
+287
-128
source/dnode/mnode/impl/src/mndMain.c
source/dnode/mnode/impl/src/mndMain.c
+2
-0
source/libs/function/src/builtins.c
source/libs/function/src/builtins.c
+32
-33
source/libs/parser/src/parAstCreater.c
source/libs/parser/src/parAstCreater.c
+9
-0
source/libs/parser/src/parTranslater.c
source/libs/parser/src/parTranslater.c
+17
-17
source/libs/parser/src/parUtil.c
source/libs/parser/src/parUtil.c
+3
-19
source/libs/planner/src/planOptimizer.c
source/libs/planner/src/planOptimizer.c
+7
-0
source/libs/planner/src/planSpliter.c
source/libs/planner/src/planSpliter.c
+2
-1
source/libs/planner/test/planBasicTest.cpp
source/libs/planner/test/planBasicTest.cpp
+6
-0
source/libs/planner/test/planOtherTest.cpp
source/libs/planner/test/planOtherTest.cpp
+0
-7
source/libs/planner/test/planTestMain.cpp
source/libs/planner/test/planTestMain.cpp
+4
-0
source/libs/planner/test/planTestUtil.cpp
source/libs/planner/test/planTestUtil.cpp
+19
-0
source/libs/planner/test/planTestUtil.h
source/libs/planner/test/planTestUtil.h
+1
-0
source/libs/sync/inc/syncSnapshot.h
source/libs/sync/inc/syncSnapshot.h
+3
-3
source/libs/sync/src/syncMain.c
source/libs/sync/src/syncMain.c
+2
-1
source/libs/sync/src/syncRespMgr.c
source/libs/sync/src/syncRespMgr.c
+57
-1
source/libs/sync/src/syncSnapshot.c
source/libs/sync/src/syncSnapshot.c
+80
-36
source/libs/sync/test/syncIndexTest.cpp
source/libs/sync/test/syncIndexTest.cpp
+7
-1
source/libs/sync/test/syncRespMgrTest.cpp
source/libs/sync/test/syncRespMgrTest.cpp
+31
-4
tests/script/tsim/stable/tag_modify.sim
tests/script/tsim/stable/tag_modify.sim
+2
-2
tests/script/tsim/stable/tag_rename.sim
tests/script/tsim/stable/tag_rename.sim
+3
-3
未找到文件。
source/dnode/mnode/impl/src/mndMain.c
浏览文件 @
93962a8c
...
...
@@ -519,6 +519,8 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
}
}
syncNodeRelease
(
pSyncNode
);
if
(
code
!=
0
)
{
terrno
=
TSDB_CODE_SYN_INTERNAL_ERROR
;
}
...
...
source/libs/function/src/builtins.c
浏览文件 @
93962a8c
...
...
@@ -15,10 +15,10 @@
#include "builtins.h"
#include "builtinsimpl.h"
#include "cJSON.h"
#include "querynodes.h"
#include "scalar.h"
#include "taoserror.h"
#include "cJSON.h"
static
int32_t
buildFuncErrMsg
(
char
*
pErrBuf
,
int32_t
len
,
int32_t
errCode
,
const
char
*
pFormat
,
...)
{
va_list
vArgList
;
...
...
@@ -40,7 +40,7 @@ static int32_t invaildFuncParaValueErrMsg(char* pErrBuf, int32_t len, const char
return
buildFuncErrMsg
(
pErrBuf
,
len
,
TSDB_CODE_FUNC_FUNTION_PARA_VALUE
,
"Invalid parameter value : %s"
,
pFuncName
);
}
#define TIME_UNIT_INVALID 1
#define TIME_UNIT_INVALID
1
#define TIME_UNIT_TOO_SMALL 2
static
int32_t
validateTimeUnitParam
(
uint8_t
dbPrec
,
const
SValueNode
*
pVal
)
{
...
...
@@ -52,10 +52,9 @@ static int32_t validateTimeUnitParam(uint8_t dbPrec, const SValueNode* pVal) {
return
TIME_UNIT_TOO_SMALL
;
}
if
(
pVal
->
literal
[
0
]
!=
'1'
||
(
pVal
->
literal
[
1
]
!=
'u'
&&
pVal
->
literal
[
1
]
!=
'a'
&&
pVal
->
literal
[
1
]
!=
's'
&&
pVal
->
literal
[
1
]
!=
'm'
&&
pVal
->
literal
[
1
]
!=
'h'
&&
pVal
->
literal
[
1
]
!=
'd'
&&
pVal
->
literal
[
1
]
!=
'w'
))
{
if
(
pVal
->
literal
[
0
]
!=
'1'
||
(
pVal
->
literal
[
1
]
!=
'u'
&&
pVal
->
literal
[
1
]
!=
'a'
&&
pVal
->
literal
[
1
]
!=
's'
&&
pVal
->
literal
[
1
]
!=
'm'
&&
pVal
->
literal
[
1
]
!=
'h'
&&
pVal
->
literal
[
1
]
!=
'd'
&&
pVal
->
literal
[
1
]
!=
'w'
))
{
return
TIME_UNIT_INVALID
;
}
...
...
@@ -696,13 +695,14 @@ static int32_t translateElapsed(SFunctionNode* pFunc, char* pErrBuf, int32_t len
uint8_t
dbPrec
=
pFunc
->
node
.
resType
.
precision
;
int32_t
ret
=
validateTimeUnitParam
(
dbPrec
,
(
SValueNode
*
)
nodesListGetNode
(
pFunc
->
pParameterList
,
1
));
int32_t
ret
=
validateTimeUnitParam
(
dbPrec
,
(
SValueNode
*
)
nodesListGetNode
(
pFunc
->
pParameterList
,
1
));
if
(
ret
==
TIME_UNIT_TOO_SMALL
)
{
return
buildFuncErrMsg
(
pErrBuf
,
len
,
TSDB_CODE_FUNC_FUNTION_ERROR
,
"ELAPSED function time unit parameter should be greater than db precision"
);
}
else
if
(
ret
==
TIME_UNIT_INVALID
)
{
return
buildFuncErrMsg
(
pErrBuf
,
len
,
TSDB_CODE_FUNC_FUNTION_ERROR
,
"ELAPSED function time unit parameter should be one of the following: [1u, 1a, 1s, 1m, 1h, 1d, 1w]"
);
return
buildFuncErrMsg
(
pErrBuf
,
len
,
TSDB_CODE_FUNC_FUNTION_ERROR
,
"ELAPSED function time unit parameter should be one of the following: [1u, 1a, 1s, 1m, 1h, 1d, 1w]"
);
}
}
...
...
@@ -815,13 +815,13 @@ static int8_t validateHistogramBinType(char* binTypeStr) {
}
static
bool
validateHistogramBinDesc
(
char
*
binDescStr
,
int8_t
binType
,
char
*
errMsg
,
int32_t
msgLen
)
{
const
char
*
msg1
=
"HISTOGRAM function requires four parameters"
;
const
char
*
msg3
=
"HISTOGRAM function invalid format for binDesc parameter"
;
const
char
*
msg4
=
"HISTOGRAM function binDesc parameter
\"
count
\"
should be in range [1, 1000]"
;
const
char
*
msg5
=
"HISTOGRAM function bin/parameter should be in range [-DBL_MAX, DBL_MAX]"
;
const
char
*
msg6
=
"HISTOGRAM function binDesc parameter
\"
width
\"
cannot be 0"
;
const
char
*
msg7
=
"HISTOGRAM function binDesc parameter
\"
start
\"
cannot be 0 with
\"
log_bin
\"
type"
;
const
char
*
msg8
=
"HISTOGRAM function binDesc parameter
\"
factor
\"
cannot be negative or equal to 0/1"
;
const
char
*
msg1
=
"HISTOGRAM function requires four parameters"
;
const
char
*
msg3
=
"HISTOGRAM function invalid format for binDesc parameter"
;
const
char
*
msg4
=
"HISTOGRAM function binDesc parameter
\"
count
\"
should be in range [1, 1000]"
;
const
char
*
msg5
=
"HISTOGRAM function bin/parameter should be in range [-DBL_MAX, DBL_MAX]"
;
const
char
*
msg6
=
"HISTOGRAM function binDesc parameter
\"
width
\"
cannot be 0"
;
const
char
*
msg7
=
"HISTOGRAM function binDesc parameter
\"
start
\"
cannot be 0 with
\"
log_bin
\"
type"
;
const
char
*
msg8
=
"HISTOGRAM function binDesc parameter
\"
factor
\"
cannot be negative or equal to 0/1"
;
cJSON
*
binDesc
=
cJSON_Parse
(
binDescStr
);
int32_t
numOfBins
;
...
...
@@ -1004,8 +1004,8 @@ static int32_t translateHistogram(SFunctionNode* pFunc, char* pErrBuf, int32_t l
}
if
(
i
==
3
&&
pValue
->
datum
.
i
!=
1
&&
pValue
->
datum
.
i
!=
0
)
{
return
buildFuncErrMsg
(
pErrBuf
,
len
,
TSDB_CODE_FUNC_FUNTION_ERROR
,
"HISTOGRAM function normalized parameter should be 0/1"
);
return
buildFuncErrMsg
(
pErrBuf
,
len
,
TSDB_CODE_FUNC_FUNTION_ERROR
,
"HISTOGRAM function normalized parameter should be 0/1"
);
}
}
...
...
@@ -1062,8 +1062,8 @@ static int32_t translateHistogramImpl(SFunctionNode* pFunc, char* pErrBuf, int32
}
if
(
i
==
3
&&
pValue
->
datum
.
i
!=
1
&&
pValue
->
datum
.
i
!=
0
)
{
return
buildFuncErrMsg
(
pErrBuf
,
len
,
TSDB_CODE_FUNC_FUNTION_ERROR
,
"HISTOGRAM function normalized parameter should be 0/1"
);
return
buildFuncErrMsg
(
pErrBuf
,
len
,
TSDB_CODE_FUNC_FUNTION_ERROR
,
"HISTOGRAM function normalized parameter should be 0/1"
);
}
}
...
...
@@ -1218,13 +1218,14 @@ static int32_t translateStateDuration(SFunctionNode* pFunc, char* pErrBuf, int32
if
(
numOfParams
==
4
)
{
uint8_t
dbPrec
=
pFunc
->
node
.
resType
.
precision
;
int32_t
ret
=
validateTimeUnitParam
(
dbPrec
,
(
SValueNode
*
)
nodesListGetNode
(
pFunc
->
pParameterList
,
3
));
int32_t
ret
=
validateTimeUnitParam
(
dbPrec
,
(
SValueNode
*
)
nodesListGetNode
(
pFunc
->
pParameterList
,
3
));
if
(
ret
==
TIME_UNIT_TOO_SMALL
)
{
return
buildFuncErrMsg
(
pErrBuf
,
len
,
TSDB_CODE_FUNC_FUNTION_ERROR
,
"STATEDURATION function time unit parameter should be greater than db precision"
);
}
else
if
(
ret
==
TIME_UNIT_INVALID
)
{
return
buildFuncErrMsg
(
pErrBuf
,
len
,
TSDB_CODE_FUNC_FUNTION_ERROR
,
"STATEDURATION function time unit parameter should be one of the following: [1u, 1a, 1s, 1m, 1h, 1d, 1w]"
);
return
buildFuncErrMsg
(
pErrBuf
,
len
,
TSDB_CODE_FUNC_FUNTION_ERROR
,
"STATEDURATION function time unit parameter should be one of the following: [1u, 1a, 1s, 1m, 1h, 1d, 1w]"
);
}
}
...
...
@@ -1432,10 +1433,6 @@ static int32_t translateFirstLast(SFunctionNode* pFunc, char* pErrBuf, int32_t l
static
int32_t
translateFirstLastImpl
(
SFunctionNode
*
pFunc
,
char
*
pErrBuf
,
int32_t
len
,
bool
isPartial
)
{
// first(col_list) will be rewritten as first(col)
if
(
2
!=
LIST_LENGTH
(
pFunc
->
pParameterList
))
{
// input has two params c0,ts, is this a bug?
return
TSDB_CODE_SUCCESS
;
}
SNode
*
pPara
=
nodesListGetNode
(
pFunc
->
pParameterList
,
0
);
uint8_t
paraType
=
((
SExprNode
*
)
pPara
)
->
resType
.
type
;
int32_t
paraBytes
=
((
SExprNode
*
)
pPara
)
->
resType
.
bytes
;
...
...
@@ -1733,13 +1730,14 @@ static int32_t translateTimeTruncate(SFunctionNode* pFunc, char* pErrBuf, int32_
// add database precision as param
uint8_t
dbPrec
=
pFunc
->
node
.
resType
.
precision
;
int32_t
ret
=
validateTimeUnitParam
(
dbPrec
,
(
SValueNode
*
)
nodesListGetNode
(
pFunc
->
pParameterList
,
1
));
int32_t
ret
=
validateTimeUnitParam
(
dbPrec
,
(
SValueNode
*
)
nodesListGetNode
(
pFunc
->
pParameterList
,
1
));
if
(
ret
==
TIME_UNIT_TOO_SMALL
)
{
return
buildFuncErrMsg
(
pErrBuf
,
len
,
TSDB_CODE_FUNC_FUNTION_ERROR
,
"TIMETRUNCATE function time unit parameter should be greater than db precision"
);
}
else
if
(
ret
==
TIME_UNIT_INVALID
)
{
return
buildFuncErrMsg
(
pErrBuf
,
len
,
TSDB_CODE_FUNC_FUNTION_ERROR
,
"TIMETRUNCATE function time unit parameter should be one of the following: [1u, 1a, 1s, 1m, 1h, 1d, 1w]"
);
return
buildFuncErrMsg
(
pErrBuf
,
len
,
TSDB_CODE_FUNC_FUNTION_ERROR
,
"TIMETRUNCATE function time unit parameter should be one of the following: [1u, 1a, 1s, 1m, 1h, 1d, 1w]"
);
}
addDbPrecisonParam
(
&
pFunc
->
pParameterList
,
dbPrec
);
...
...
@@ -1772,13 +1770,14 @@ static int32_t translateTimeDiff(SFunctionNode* pFunc, char* pErrBuf, int32_t le
uint8_t
dbPrec
=
pFunc
->
node
.
resType
.
precision
;
if
(
3
==
numOfParams
)
{
int32_t
ret
=
validateTimeUnitParam
(
dbPrec
,
(
SValueNode
*
)
nodesListGetNode
(
pFunc
->
pParameterList
,
2
));
int32_t
ret
=
validateTimeUnitParam
(
dbPrec
,
(
SValueNode
*
)
nodesListGetNode
(
pFunc
->
pParameterList
,
2
));
if
(
ret
==
TIME_UNIT_TOO_SMALL
)
{
return
buildFuncErrMsg
(
pErrBuf
,
len
,
TSDB_CODE_FUNC_FUNTION_ERROR
,
"TIMEDIFF function time unit parameter should be greater than db precision"
);
}
else
if
(
ret
==
TIME_UNIT_INVALID
)
{
return
buildFuncErrMsg
(
pErrBuf
,
len
,
TSDB_CODE_FUNC_FUNTION_ERROR
,
"TIMEDIFF function time unit parameter should be one of the following: [1u, 1a, 1s, 1m, 1h, 1d, 1w]"
);
return
buildFuncErrMsg
(
pErrBuf
,
len
,
TSDB_CODE_FUNC_FUNTION_ERROR
,
"TIMEDIFF function time unit parameter should be one of the following: [1u, 1a, 1s, 1m, 1h, 1d, 1w]"
);
}
}
...
...
source/libs/parser/src/parAstCreater.c
浏览文件 @
93962a8c
...
...
@@ -385,6 +385,15 @@ SNode* createLogicConditionNode(SAstCreateContext* pCxt, ELogicConditionType typ
SNode
*
createOperatorNode
(
SAstCreateContext
*
pCxt
,
EOperatorType
type
,
SNode
*
pLeft
,
SNode
*
pRight
)
{
CHECK_PARSER_STATUS
(
pCxt
);
if
(
OP_TYPE_MINUS
==
type
&&
QUERY_NODE_VALUE
==
nodeType
(
pLeft
))
{
SValueNode
*
pVal
=
(
SValueNode
*
)
pLeft
;
char
*
pNewLiteral
=
taosMemoryCalloc
(
1
,
strlen
(
pVal
->
literal
)
+
2
);
CHECK_OUT_OF_MEM
(
pNewLiteral
);
sprintf
(
pNewLiteral
,
"-%s"
,
pVal
->
literal
);
taosMemoryFree
(
pVal
->
literal
);
pVal
->
literal
=
pNewLiteral
;
return
pLeft
;
}
SOperatorNode
*
op
=
(
SOperatorNode
*
)
nodesMakeNode
(
QUERY_NODE_OPERATOR
);
CHECK_OUT_OF_MEM
(
op
);
op
->
opType
=
type
;
...
...
source/libs/parser/src/parTranslater.c
浏览文件 @
93962a8c
...
...
@@ -558,11 +558,11 @@ static void setColumnInfoByExpr(STempTableNode* pTable, SExprNode* pExpr, SColum
pCol
->
node
.
resType
=
pExpr
->
resType
;
}
static
int32_t
createColumnsByTable
(
STranslateContext
*
pCxt
,
const
STableNode
*
pTable
,
SNodeList
*
pList
)
{
static
int32_t
createColumnsByTable
(
STranslateContext
*
pCxt
,
const
STableNode
*
pTable
,
bool
igTags
,
SNodeList
*
pList
)
{
if
(
QUERY_NODE_REAL_TABLE
==
nodeType
(
pTable
))
{
const
STableMeta
*
pMeta
=
((
SRealTableNode
*
)
pTable
)
->
pMeta
;
int32_t
nums
=
pMeta
->
tableInfo
.
numOfColumns
+
((
TSDB_SUPER_TABLE
==
pMeta
->
tableType
)
?
pMeta
->
tableInfo
.
numOfTags
:
0
);
int32_t
nums
=
pMeta
->
tableInfo
.
numOfColumns
+
(
igTags
?
0
:
((
TSDB_SUPER_TABLE
==
pMeta
->
tableType
)
?
pMeta
->
tableInfo
.
numOfTags
:
0
)
);
for
(
int32_t
i
=
0
;
i
<
nums
;
++
i
)
{
SColumnNode
*
pCol
=
(
SColumnNode
*
)
nodesMakeNode
(
QUERY_NODE_COLUMN
);
if
(
NULL
==
pCol
)
{
...
...
@@ -878,6 +878,9 @@ static EDealRes translateNormalValue(STranslateContext* pCxt, SValueNode* pVal,
}
case
TSDB_DATA_TYPE_VARCHAR
:
case
TSDB_DATA_TYPE_VARBINARY
:
{
if
(
strict
&&
(
pVal
->
node
.
resType
.
bytes
>
targetDt
.
bytes
-
VARSTR_HEADER_SIZE
))
{
return
generateDealNodeErrMsg
(
pCxt
,
TSDB_CODE_PAR_WRONG_VALUE_TYPE
,
pVal
->
literal
);
}
pVal
->
datum
.
p
=
taosMemoryCalloc
(
1
,
targetDt
.
bytes
+
1
);
if
(
NULL
==
pVal
->
datum
.
p
)
{
return
generateDealNodeErrMsg
(
pCxt
,
TSDB_CODE_OUT_OF_MEMORY
);
...
...
@@ -1333,6 +1336,9 @@ static int32_t rewriteSystemInfoFuncImpl(STranslateContext* pCxt, char* pLiteral
pVal
->
isNull
=
true
;
}
else
{
pVal
->
literal
=
pLiteral
;
if
(
IS_VAR_DATA_TYPE
(
pVal
->
node
.
resType
.
type
))
{
pVal
->
node
.
resType
.
bytes
=
strlen
(
pLiteral
);
}
}
if
(
DEAL_RES_ERROR
!=
translateValue
(
pCxt
,
pVal
))
{
*
pNode
=
(
SNode
*
)
pVal
;
...
...
@@ -1928,7 +1934,7 @@ static int32_t translateTable(STranslateContext* pCxt, SNode* pTable) {
return
code
;
}
static
int32_t
createAllColumns
(
STranslateContext
*
pCxt
,
SNodeList
**
pCols
)
{
static
int32_t
createAllColumns
(
STranslateContext
*
pCxt
,
bool
igTags
,
SNodeList
**
pCols
)
{
*
pCols
=
nodesMakeList
();
if
(
NULL
==
*
pCols
)
{
return
generateSyntaxErrMsg
(
&
pCxt
->
msgBuf
,
TSDB_CODE_OUT_OF_MEMORY
);
...
...
@@ -1937,7 +1943,7 @@ static int32_t createAllColumns(STranslateContext* pCxt, SNodeList** pCols) {
size_t
nums
=
taosArrayGetSize
(
pTables
);
for
(
size_t
i
=
0
;
i
<
nums
;
++
i
)
{
STableNode
*
pTable
=
taosArrayGetP
(
pTables
,
i
);
int32_t
code
=
createColumnsByTable
(
pCxt
,
pTable
,
*
pCols
);
int32_t
code
=
createColumnsByTable
(
pCxt
,
pTable
,
igTags
,
*
pCols
);
if
(
TSDB_CODE_SUCCESS
!=
code
)
{
return
code
;
}
...
...
@@ -1974,7 +1980,7 @@ static SNode* createMultiResFunc(SFunctionNode* pSrcFunc, SExprNode* pExpr) {
return
(
SNode
*
)
pFunc
;
}
static
int32_t
createTableAllCols
(
STranslateContext
*
pCxt
,
SColumnNode
*
pCol
,
SNodeList
**
pOutput
)
{
static
int32_t
createTableAllCols
(
STranslateContext
*
pCxt
,
SColumnNode
*
pCol
,
bool
igTags
,
SNodeList
**
pOutput
)
{
STableNode
*
pTable
=
NULL
;
int32_t
code
=
findTable
(
pCxt
,
pCol
->
tableAlias
,
&
pTable
);
if
(
TSDB_CODE_SUCCESS
==
code
&&
NULL
==
*
pOutput
)
{
...
...
@@ -1984,7 +1990,7 @@ static int32_t createTableAllCols(STranslateContext* pCxt, SColumnNode* pCol, SN
}
}
if
(
TSDB_CODE_SUCCESS
==
code
)
{
code
=
createColumnsByTable
(
pCxt
,
pTable
,
*
pOutput
);
code
=
createColumnsByTable
(
pCxt
,
pTable
,
igTags
,
*
pOutput
);
}
return
code
;
}
...
...
@@ -2006,11 +2012,9 @@ static int32_t createMultiResFuncsParas(STranslateContext* pCxt, SNodeList* pSrc
SNode
*
pPara
=
NULL
;
FOREACH
(
pPara
,
pSrcParas
)
{
if
(
isStar
(
pPara
))
{
code
=
createAllColumns
(
pCxt
,
&
pExprs
);
// The syntax definition ensures that * and other parameters do not appear at the same time
break
;
code
=
createAllColumns
(
pCxt
,
true
,
&
pExprs
);
}
else
if
(
isTableStar
(
pPara
))
{
code
=
createTableAllCols
(
pCxt
,
(
SColumnNode
*
)
pPara
,
&
pExprs
);
code
=
createTableAllCols
(
pCxt
,
(
SColumnNode
*
)
pPara
,
true
,
&
pExprs
);
}
else
{
code
=
nodesListMakeStrictAppend
(
&
pExprs
,
nodesCloneNode
(
pPara
));
}
...
...
@@ -2069,7 +2073,7 @@ static int32_t translateStar(STranslateContext* pCxt, SSelectStmt* pSelect) {
int32_t
code
=
TSDB_CODE_SUCCESS
;
if
(
isStar
(
pNode
))
{
SNodeList
*
pCols
=
NULL
;
code
=
createAllColumns
(
pCxt
,
&
pCols
);
code
=
createAllColumns
(
pCxt
,
false
,
&
pCols
);
if
(
TSDB_CODE_SUCCESS
==
code
)
{
INSERT_LIST
(
pSelect
->
pProjectionList
,
pCols
);
ERASE_NODE
(
pSelect
->
pProjectionList
);
...
...
@@ -2085,7 +2089,7 @@ static int32_t translateStar(STranslateContext* pCxt, SSelectStmt* pSelect) {
}
}
else
if
(
isTableStar
(
pNode
))
{
SNodeList
*
pCols
=
NULL
;
code
=
createTableAllCols
(
pCxt
,
(
SColumnNode
*
)
pNode
,
&
pCols
);
code
=
createTableAllCols
(
pCxt
,
(
SColumnNode
*
)
pNode
,
false
,
&
pCols
);
if
(
TSDB_CODE_SUCCESS
==
code
)
{
INSERT_LIST
(
pSelect
->
pProjectionList
,
pCols
);
ERASE_NODE
(
pSelect
->
pProjectionList
);
...
...
@@ -5826,10 +5830,6 @@ static int32_t buildUpdateTagValReq(STranslateContext* pCxt, SAlterTableStmt* pS
return
pCxt
->
errCode
;
}
if
(
IS_VAR_DATA_TYPE
(
pSchema
->
type
)
&&
strlen
(
pStmt
->
pVal
->
literal
)
>
pSchema
->
bytes
)
{
return
generateSyntaxErrMsg
(
&
pCxt
->
msgBuf
,
TSDB_CODE_PAR_WRONG_VALUE_TYPE
,
pStmt
->
pVal
->
literal
);
}
pReq
->
isNull
=
(
TSDB_DATA_TYPE_NULL
==
pStmt
->
pVal
->
node
.
resType
.
type
);
if
(
targetDt
.
type
==
TSDB_DATA_TYPE_JSON
)
{
pReq
->
isNull
=
0
;
...
...
source/libs/parser/src/parUtil.c
浏览文件 @
93962a8c
...
...
@@ -236,7 +236,6 @@ int32_t buildSyntaxErrMsg(SMsgBuf* pBuf, const char* additionalInfo, const char*
const
char
*
prefix
=
"syntax error"
;
if
(
sourceStr
==
NULL
)
{
assert
(
additionalInfo
!=
NULL
);
snprintf
(
pBuf
->
buf
,
pBuf
->
len
,
msgFormat1
,
additionalInfo
);
return
TSDB_CODE_TSC_SQL_SYNTAX_ERROR
;
}
...
...
@@ -254,40 +253,25 @@ int32_t buildSyntaxErrMsg(SMsgBuf* pBuf, const char* additionalInfo, const char*
return
TSDB_CODE_TSC_SQL_SYNTAX_ERROR
;
}
SSchema
*
getTableColumnSchema
(
const
STableMeta
*
pTableMeta
)
{
assert
(
pTableMeta
!=
NULL
);
return
(
SSchema
*
)
pTableMeta
->
schema
;
}
SSchema
*
getTableColumnSchema
(
const
STableMeta
*
pTableMeta
)
{
return
(
SSchema
*
)
pTableMeta
->
schema
;
}
static
SSchema
*
getOneColumnSchema
(
const
STableMeta
*
pTableMeta
,
int32_t
colIndex
)
{
assert
(
pTableMeta
!=
NULL
&&
pTableMeta
->
schema
!=
NULL
&&
colIndex
>=
0
&&
colIndex
<
(
getNumOfColumns
(
pTableMeta
)
+
getNumOfTags
(
pTableMeta
)));
SSchema
*
pSchema
=
(
SSchema
*
)
pTableMeta
->
schema
;
return
&
pSchema
[
colIndex
];
}
SSchema
*
getTableTagSchema
(
const
STableMeta
*
pTableMeta
)
{
assert
(
pTableMeta
!=
NULL
&&
(
pTableMeta
->
tableType
==
TSDB_SUPER_TABLE
||
pTableMeta
->
tableType
==
TSDB_CHILD_TABLE
));
return
getOneColumnSchema
(
pTableMeta
,
getTableInfo
(
pTableMeta
).
numOfColumns
);
}
int32_t
getNumOfColumns
(
const
STableMeta
*
pTableMeta
)
{
assert
(
pTableMeta
!=
NULL
);
// table created according to super table, use data from super table
return
getTableInfo
(
pTableMeta
).
numOfColumns
;
}
int32_t
getNumOfTags
(
const
STableMeta
*
pTableMeta
)
{
assert
(
pTableMeta
!=
NULL
);
return
getTableInfo
(
pTableMeta
).
numOfTags
;
}
int32_t
getNumOfTags
(
const
STableMeta
*
pTableMeta
)
{
return
getTableInfo
(
pTableMeta
).
numOfTags
;
}
STableComInfo
getTableInfo
(
const
STableMeta
*
pTableMeta
)
{
assert
(
pTableMeta
!=
NULL
);
return
pTableMeta
->
tableInfo
;
}
STableComInfo
getTableInfo
(
const
STableMeta
*
pTableMeta
)
{
return
pTableMeta
->
tableInfo
;
}
STableMeta
*
tableMetaDup
(
const
STableMeta
*
pTableMeta
)
{
size_t
size
=
TABLE_META_SIZE
(
pTableMeta
);
...
...
source/libs/planner/src/planOptimizer.c
浏览文件 @
93962a8c
...
...
@@ -75,6 +75,11 @@ static SLogicNode* optFindPossibleNode(SLogicNode* pNode, FMayBeOptimized func)
return
NULL
;
}
static
void
optResetParent
(
SLogicNode
*
pNode
)
{
SNode
*
pChild
=
NULL
;
FOREACH
(
pChild
,
pNode
->
pChildren
)
{
((
SLogicNode
*
)
pChild
)
->
pParent
=
pNode
;
}
}
EDealRes
scanPathOptHaveNormalColImpl
(
SNode
*
pNode
,
void
*
pContext
)
{
if
(
QUERY_NODE_COLUMN
==
nodeType
(
pNode
))
{
// *((bool*)pContext) = (COLUMN_TYPE_TAG != ((SColumnNode*)pNode)->colType);
...
...
@@ -1540,6 +1545,7 @@ static int32_t rewriteTailOptCreateSort(SIndefRowsFuncLogicNode* pIndef, SLogicN
pSort
->
groupSort
=
rewriteTailOptNeedGroupSort
(
pIndef
);
TSWAP
(
pSort
->
node
.
pChildren
,
pIndef
->
node
.
pChildren
);
optResetParent
((
SLogicNode
*
)
pSort
);
pSort
->
node
.
precision
=
pIndef
->
node
.
precision
;
SFunctionNode
*
pTail
=
NULL
;
...
...
@@ -1747,6 +1753,7 @@ static int32_t rewriteUniqueOptCreateAgg(SIndefRowsFuncLogicNode* pIndef, SLogic
}
TSWAP
(
pAgg
->
node
.
pChildren
,
pIndef
->
node
.
pChildren
);
optResetParent
((
SLogicNode
*
)
pAgg
);
pAgg
->
node
.
precision
=
pIndef
->
node
.
precision
;
int32_t
code
=
TSDB_CODE_SUCCESS
;
...
...
source/libs/planner/src/planSpliter.c
浏览文件 @
93962a8c
...
...
@@ -1203,7 +1203,8 @@ typedef struct SQnodeSplitInfo {
static
bool
qndSplFindSplitNode
(
SSplitContext
*
pCxt
,
SLogicSubplan
*
pSubplan
,
SLogicNode
*
pNode
,
SQnodeSplitInfo
*
pInfo
)
{
if
(
QUERY_NODE_LOGIC_PLAN_SCAN
==
nodeType
(
pNode
)
&&
NULL
!=
pNode
->
pParent
)
{
if
(
QUERY_NODE_LOGIC_PLAN_SCAN
==
nodeType
(
pNode
)
&&
NULL
!=
pNode
->
pParent
&&
((
SScanLogicNode
*
)
pNode
)
->
scanSeq
[
0
]
<
1
&&
((
SScanLogicNode
*
)
pNode
)
->
scanSeq
[
1
]
<
1
)
{
pInfo
->
pSplitNode
=
pNode
;
pInfo
->
pSubplan
=
pSubplan
;
return
true
;
...
...
source/libs/planner/test/planBasicTest.cpp
浏览文件 @
93962a8c
...
...
@@ -63,6 +63,10 @@ TEST_F(PlanBasicTest, uniqueFunc) {
run
(
"SELECT UNIQUE(c2 + 10), ts, c2 FROM t1 WHERE c1 > 10"
);
run
(
"SELECT UNIQUE(c1) a FROM t1 ORDER BY a"
);
run
(
"SELECT ts, UNIQUE(c1) FROM st1 PARTITION BY TBNAME"
);
run
(
"SELECT TBNAME, UNIQUE(c1) FROM st1 PARTITION BY TBNAME"
);
}
TEST_F
(
PlanBasicTest
,
tailFunc
)
{
...
...
@@ -81,6 +85,8 @@ TEST_F(PlanBasicTest, tailFunc) {
run
(
"SELECT TAIL(c2 + 10, 10, 80) FROM t1 WHERE c1 > 10 PARTITION BY c1 LIMIT 5"
);
run
(
"SELECT TAIL(c1, 2, 1) FROM st1s1 UNION ALL SELECT c1 FROM st1s2"
);
run
(
"SELECT TAIL(c1, 1) FROM st2 WHERE jtag->'tag1' > 10"
);
}
TEST_F
(
PlanBasicTest
,
interpFunc
)
{
...
...
source/libs/planner/test/planOtherTest.cpp
浏览文件 @
93962a8c
...
...
@@ -91,10 +91,3 @@ TEST_F(PlanOtherTest, delete) {
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/planTestMain.cpp
浏览文件 @
93962a8c
...
...
@@ -78,6 +78,7 @@ static void parseArg(int argc, char* argv[]) {
{
"skipSql"
,
required_argument
,
NULL
,
's'
},
{
"limitSql"
,
required_argument
,
NULL
,
'i'
},
{
"log"
,
required_argument
,
NULL
,
'l'
},
{
"queryPolicy"
,
required_argument
,
NULL
,
'q'
},
{
0
,
0
,
0
,
0
}
};
// clang-format on
...
...
@@ -95,6 +96,9 @@ static void parseArg(int argc, char* argv[]) {
case
'l'
:
setLogLevel
(
optarg
);
break
;
case
'q'
:
setQueryPolicy
(
optarg
);
break
;
default:
break
;
}
...
...
source/libs/planner/test/planTestUtil.cpp
浏览文件 @
93962a8c
...
...
@@ -24,6 +24,7 @@
#include "mockCatalogService.h"
#include "parser.h"
#include "planInt.h"
#include "tglobal.h"
using
namespace
std
;
using
namespace
testing
;
...
...
@@ -53,6 +54,7 @@ DumpModule g_dumpModule = DUMP_MODULE_NOTHING;
int32_t
g_skipSql
=
0
;
int32_t
g_limitSql
=
0
;
int32_t
g_logLevel
=
131
;
int32_t
g_queryPolicy
=
QUERY_POLICY_VNODE
;
void
setDumpModule
(
const
char
*
pModule
)
{
if
(
NULL
==
pModule
)
{
...
...
@@ -79,6 +81,7 @@ void setDumpModule(const char* pModule) {
void
setSkipSqlNum
(
const
char
*
pNum
)
{
g_skipSql
=
stoi
(
pNum
);
}
void
setLimitSqlNum
(
const
char
*
pNum
)
{
g_limitSql
=
stoi
(
pNum
);
}
void
setLogLevel
(
const
char
*
pLogLevel
)
{
g_logLevel
=
stoi
(
pLogLevel
);
}
void
setQueryPolicy
(
const
char
*
pQueryPolicy
)
{
g_queryPolicy
=
stoi
(
pQueryPolicy
);
}
int32_t
getLogLevel
()
{
return
g_logLevel
;
}
...
...
@@ -105,7 +108,23 @@ class PlannerTestBaseImpl {
}
++
sqlNum_
;
switch
(
g_queryPolicy
)
{
case
QUERY_POLICY_VNODE
:
case
QUERY_POLICY_HYBRID
:
case
QUERY_POLICY_QNODE
:
runImpl
(
sql
,
g_queryPolicy
);
break
;
default:
runImpl
(
sql
,
QUERY_POLICY_VNODE
);
runImpl
(
sql
,
QUERY_POLICY_HYBRID
);
runImpl
(
sql
,
QUERY_POLICY_QNODE
);
break
;
}
}
void
runImpl
(
const
string
&
sql
,
int32_t
queryPolicy
)
{
reset
();
tsQueryPolicy
=
queryPolicy
;
try
{
SQuery
*
pQuery
=
nullptr
;
doParseSql
(
sql
,
&
pQuery
);
...
...
source/libs/planner/test/planTestUtil.h
浏览文件 @
93962a8c
...
...
@@ -45,6 +45,7 @@ extern void setDumpModule(const char* pModule);
extern
void
setSkipSqlNum
(
const
char
*
pNum
);
extern
void
setLimitSqlNum
(
const
char
*
pNum
);
extern
void
setLogLevel
(
const
char
*
pLogLevel
);
extern
void
setQueryPolicy
(
const
char
*
pQueryPolicy
);
extern
int32_t
getLogLevel
();
#endif // PLAN_TEST_UTIL_H
source/libs/sync/inc/syncSnapshot.h
浏览文件 @
93962a8c
...
...
@@ -83,9 +83,9 @@ typedef struct SSyncSnapshotReceiver {
SSyncSnapshotReceiver
*
snapshotReceiverCreate
(
SSyncNode
*
pSyncNode
,
SRaftId
fromId
);
void
snapshotReceiverDestroy
(
SSyncSnapshotReceiver
*
pReceiver
);
int32_t
snapshotReceiverStart
(
SSyncSnapshotReceiver
*
pReceiver
,
SyncTerm
privateTerm
,
SyncSnapshotSend
*
pBeginMsg
);
int32_t
snapshotReceiverStop
(
SSyncSnapshotReceiver
*
pReceiver
);
bool
snapshotReceiverIsStart
(
SSyncSnapshotReceiver
*
pReceiver
);
int32_t
snapshotReceiverStart
(
SSyncSnapshotReceiver
*
pReceiver
,
SyncSnapshotSend
*
pBeginMsg
);
int32_t
snapshotReceiverStop
(
SSyncSnapshotReceiver
*
pReceiver
);
bool
snapshotReceiverIsStart
(
SSyncSnapshotReceiver
*
pReceiver
);
cJSON
*
snapshotReceiver2Json
(
SSyncSnapshotReceiver
*
pReceiver
);
char
*
snapshotReceiver2Str
(
SSyncSnapshotReceiver
*
pReceiver
);
...
...
source/libs/sync/src/syncMain.c
浏览文件 @
93962a8c
...
...
@@ -158,13 +158,13 @@ int32_t syncSetStandby(int64_t rid) {
}
if
(
pSyncNode
->
state
!=
TAOS_SYNC_STATE_FOLLOWER
)
{
taosReleaseRef
(
tsNodeRefId
,
pSyncNode
->
rid
);
if
(
pSyncNode
->
state
==
TAOS_SYNC_STATE_LEADER
)
{
terrno
=
TSDB_CODE_SYN_IS_LEADER
;
}
else
{
terrno
=
TSDB_CODE_SYN_STANDBY_NOT_READY
;
}
sError
(
"failed to set standby since it is not follower, state:%s rid:%"
PRId64
,
syncStr
(
pSyncNode
->
state
),
rid
);
taosReleaseRef
(
tsNodeRefId
,
pSyncNode
->
rid
);
return
-
1
;
}
...
...
@@ -620,6 +620,7 @@ int32_t syncPropose(int64_t rid, SRpcMsg* pMsg, bool isWeak) {
SSyncNode
*
pSyncNode
=
taosAcquireRef
(
tsNodeRefId
,
rid
);
if
(
pSyncNode
==
NULL
)
{
taosReleaseRef
(
tsNodeRefId
,
rid
);
terrno
=
TSDB_CODE_SYN_INTERNAL_ERROR
;
return
-
1
;
}
...
...
source/libs/sync/src/syncRespMgr.c
浏览文件 @
93962a8c
...
...
@@ -14,6 +14,7 @@
*/
#include "syncRespMgr.h"
#include "syncRaftEntry.h"
#include "syncRaftStore.h"
SSyncRespMgr
*
syncRespMgrCreate
(
void
*
data
,
int64_t
ttl
)
{
...
...
@@ -116,4 +117,59 @@ void syncRespClean(SSyncRespMgr *pObj) {
taosThreadMutexUnlock
(
&
(
pObj
->
mutex
));
}
void
syncRespCleanByTTL
(
SSyncRespMgr
*
pObj
,
int64_t
ttl
)
{}
void
syncRespCleanByTTL
(
SSyncRespMgr
*
pObj
,
int64_t
ttl
)
{
SRespStub
*
pStub
=
(
SRespStub
*
)
taosHashIterate
(
pObj
->
pRespHash
,
NULL
);
int
cnt
=
0
;
SSyncNode
*
pSyncNode
=
pObj
->
data
;
SArray
*
delIndexArray
=
taosArrayInit
(
0
,
sizeof
(
SyncIndex
));
ASSERT
(
delIndexArray
!=
NULL
);
while
(
pStub
)
{
size_t
len
;
void
*
key
=
taosHashGetKey
(
pStub
,
&
len
);
SyncIndex
*
pIndex
=
(
SyncIndex
*
)
key
;
int64_t
nowMS
=
taosGetTimestampMs
();
if
(
nowMS
-
pStub
->
createTime
>
ttl
)
{
taosArrayPush
(
delIndexArray
,
pIndex
);
cnt
++
;
SSyncRaftEntry
*
pEntry
=
NULL
;
int32_t
code
=
0
;
if
(
pSyncNode
->
pLogStore
!=
NULL
)
{
code
=
pSyncNode
->
pLogStore
->
syncLogGetEntry
(
pSyncNode
->
pLogStore
,
*
pIndex
,
&
pEntry
);
if
(
code
==
0
&&
pEntry
!=
NULL
)
{
SFsmCbMeta
cbMeta
=
{
0
};
cbMeta
.
index
=
pEntry
->
index
;
cbMeta
.
lastConfigIndex
=
syncNodeGetSnapshotConfigIndex
(
pSyncNode
,
cbMeta
.
index
);
cbMeta
.
isWeak
=
pEntry
->
isWeak
;
cbMeta
.
code
=
TSDB_CODE_SYN_TIMEOUT
;
cbMeta
.
state
=
pSyncNode
->
state
;
cbMeta
.
seqNum
=
pEntry
->
seqNum
;
cbMeta
.
term
=
pEntry
->
term
;
cbMeta
.
currentTerm
=
pSyncNode
->
pRaftStore
->
currentTerm
;
cbMeta
.
flag
=
0
;
SRpcMsg
rpcMsg
=
pStub
->
rpcMsg
;
rpcMsg
.
pCont
=
rpcMallocCont
(
pEntry
->
dataLen
);
memcpy
(
rpcMsg
.
pCont
,
pEntry
->
data
,
pEntry
->
dataLen
);
pSyncNode
->
pFsm
->
FpCommitCb
(
pSyncNode
->
pFsm
,
&
rpcMsg
,
cbMeta
);
syncEntryDestory
(
pEntry
);
}
}
}
pStub
=
(
SRespStub
*
)
taosHashIterate
(
pObj
->
pRespHash
,
pStub
);
}
int32_t
arraySize
=
taosArrayGetSize
(
delIndexArray
);
sDebug
(
"vgId:%d, resp clean by ttl, cnt:%d, array-size:%d"
,
pSyncNode
->
vgId
,
cnt
,
arraySize
);
for
(
int32_t
i
=
0
;
i
<
arraySize
;
++
i
)
{
SyncIndex
*
pIndex
=
taosArrayGet
(
delIndexArray
,
i
);
taosHashRemove
(
pObj
->
pRespHash
,
pIndex
,
sizeof
(
SyncIndex
));
}
taosArrayDestroy
(
delIndexArray
);
}
source/libs/sync/src/syncSnapshot.c
浏览文件 @
93962a8c
...
...
@@ -22,9 +22,11 @@
#include "wal.h"
//----------------------------------
static
void
snapshotReceiverDoStart
(
SSyncSnapshotReceiver
*
pReceiver
,
SyncTerm
privateTerm
,
SyncSnapshotSend
*
pBeginMsg
);
static
void
snapshotReceiverGotData
(
SSyncSnapshotReceiver
*
pReceiver
,
SyncSnapshotSend
*
pMsg
);
static
void
snapshotSenderUpdateProgress
(
SSyncSnapshotSender
*
pSender
,
SyncSnapshotRsp
*
pMsg
);
static
void
snapshotReceiverDoStart
(
SSyncSnapshotReceiver
*
pReceiver
,
SyncSnapshotSend
*
pBeginMsg
);
static
void
snapshotReceiverForceStop
(
SSyncSnapshotReceiver
*
pReceiver
);
static
void
snapshotReceiverGotData
(
SSyncSnapshotReceiver
*
pReceiver
,
SyncSnapshotSend
*
pMsg
);
static
int32_t
snapshotReceiverFinish
(
SSyncSnapshotReceiver
*
pReceiver
,
SyncSnapshotSend
*
pMsg
);
//----------------------------------
SSyncSnapshotSender
*
snapshotSenderCreate
(
SSyncNode
*
pSyncNode
,
int32_t
replicaIndex
)
{
...
...
@@ -68,7 +70,9 @@ void snapshotSenderDestroy(SSyncSnapshotSender *pSender) {
// close reader
if
(
pSender
->
pReader
!=
NULL
)
{
int32_t
ret
=
pSender
->
pSyncNode
->
pFsm
->
FpSnapshotStopRead
(
pSender
->
pSyncNode
->
pFsm
,
pSender
->
pReader
);
ASSERT
(
ret
==
0
);
if
(
ret
!=
0
)
{
syncNodeErrorLog
(
pSender
->
pSyncNode
,
"stop reader error"
);
}
pSender
->
pReader
=
NULL
;
}
...
...
@@ -79,7 +83,12 @@ void snapshotSenderDestroy(SSyncSnapshotSender *pSender) {
bool
snapshotSenderIsStart
(
SSyncSnapshotSender
*
pSender
)
{
return
pSender
->
start
;
}
// begin send snapshot by snapshot, pReader
// begin send snapshot by param, snapshot, pReader
//
// action:
// 1. assert reader not start
// 2. update state
// 3. send first snapshot block
int32_t
snapshotSenderStart
(
SSyncSnapshotSender
*
pSender
,
SSnapshotParam
snapshotParam
,
SSnapshot
snapshot
,
void
*
pReader
)
{
ASSERT
(
!
snapshotSenderIsStart
(
pSender
));
...
...
@@ -98,7 +107,7 @@ int32_t snapshotSenderStart(SSyncSnapshotSender *pSender, SSnapshotParam snapsho
// update term
pSender
->
term
=
pSender
->
pSyncNode
->
pRaftStore
->
currentTerm
;
++
(
pSender
->
privateTerm
);
++
(
pSender
->
privateTerm
);
// increase private term
// update state
pSender
->
finish
=
false
;
...
...
@@ -114,9 +123,7 @@ int32_t snapshotSenderStart(SSyncSnapshotSender *pSender, SSnapshotParam snapsho
code
=
pSender
->
pSyncNode
->
pLogStore
->
syncLogGetEntry
(
pSender
->
pSyncNode
->
pLogStore
,
pSender
->
snapshot
.
lastConfigIndex
,
&
pEntry
);
if
(
code
==
0
)
{
ASSERT
(
pEntry
!=
NULL
);
if
(
code
==
0
&&
pEntry
!=
NULL
)
{
SRpcMsg
rpcMsg
;
syncEntry2OriginalRpc
(
pEntry
,
&
rpcMsg
);
...
...
@@ -207,6 +214,8 @@ int32_t snapshotSenderStop(SSyncSnapshotSender *pSender, bool finish) {
pSender
->
start
=
false
;
pSender
->
finish
=
finish
;
// do not update term, maybe print
// event log
do
{
char
*
eventLog
=
snapshotSender2SimpleStr
(
pSender
,
"snapshot sender stop"
);
...
...
@@ -243,6 +252,7 @@ int32_t snapshotSend(SSyncSnapshotSender *pSender) {
pMsg
->
srcId
=
pSender
->
pSyncNode
->
myRaftId
;
pMsg
->
destId
=
(
pSender
->
pSyncNode
->
replicasId
)[
pSender
->
replicaIndex
];
pMsg
->
term
=
pSender
->
pSyncNode
->
pRaftStore
->
currentTerm
;
pMsg
->
beginIndex
=
pSender
->
snapshotParam
.
start
;
pMsg
->
lastIndex
=
pSender
->
snapshot
.
lastApplyIndex
;
pMsg
->
lastTerm
=
pSender
->
snapshot
.
lastApplyTerm
;
pMsg
->
lastConfigIndex
=
pSender
->
snapshot
.
lastConfigIndex
;
...
...
@@ -281,11 +291,13 @@ int32_t snapshotReSend(SSyncSnapshotSender *pSender) {
pMsg
->
srcId
=
pSender
->
pSyncNode
->
myRaftId
;
pMsg
->
destId
=
(
pSender
->
pSyncNode
->
replicasId
)[
pSender
->
replicaIndex
];
pMsg
->
term
=
pSender
->
pSyncNode
->
pRaftStore
->
currentTerm
;
pMsg
->
beginIndex
=
pSender
->
snapshotParam
.
start
;
pMsg
->
lastIndex
=
pSender
->
snapshot
.
lastApplyIndex
;
pMsg
->
lastTerm
=
pSender
->
snapshot
.
lastApplyTerm
;
pMsg
->
lastConfigIndex
=
pSender
->
snapshot
.
lastConfigIndex
;
pMsg
->
lastConfig
=
pSender
->
lastConfig
;
pMsg
->
seq
=
pSender
->
seq
;
pMsg
->
privateTerm
=
pSender
->
privateTerm
;
memcpy
(
pMsg
->
data
,
pSender
->
pCurrentBlock
,
pSender
->
blockLen
);
// send msg
...
...
@@ -305,6 +317,12 @@ int32_t snapshotReSend(SSyncSnapshotSender *pSender) {
return
0
;
}
static
void
snapshotSenderUpdateProgress
(
SSyncSnapshotSender
*
pSender
,
SyncSnapshotRsp
*
pMsg
)
{
ASSERT
(
pMsg
->
ack
==
pSender
->
seq
);
pSender
->
ack
=
pMsg
->
ack
;
++
(
pSender
->
seq
);
}
cJSON
*
snapshotSender2Json
(
SSyncSnapshotSender
*
pSender
)
{
char
u64buf
[
128
];
cJSON
*
pRoot
=
cJSON_CreateObject
();
...
...
@@ -371,10 +389,11 @@ char *snapshotSender2SimpleStr(SSyncSnapshotSender *pSender, char *event) {
syncUtilU642Addr
(
destId
.
addr
,
host
,
sizeof
(
host
),
&
port
);
snprintf
(
s
,
len
,
"%s {%p laindex:%ld laterm:%lu lcindex:%ld seq:%d ack:%d finish:%d pterm:%lu replica-index:%d %s:%d}"
,
event
,
pSender
,
pSender
->
snapshot
.
lastApplyIndex
,
pSender
->
snapshot
.
lastApplyTerm
,
pSender
->
snapshot
.
lastConfigIndex
,
pSender
->
seq
,
pSender
->
ack
,
pSender
->
finish
,
pSender
->
privateTerm
,
pSender
->
replicaIndex
,
host
,
port
);
"%s {%p s-param:%ld e-param:%ld laindex:%ld laterm:%lu lcindex:%ld seq:%d ack:%d finish:%d pterm:%lu "
"replica-index:%d %s:%d}"
,
event
,
pSender
,
pSender
->
snapshotParam
.
start
,
pSender
->
snapshotParam
.
end
,
pSender
->
snapshot
.
lastApplyIndex
,
pSender
->
snapshot
.
lastApplyTerm
,
pSender
->
snapshot
.
lastConfigIndex
,
pSender
->
seq
,
pSender
->
ack
,
pSender
->
finish
,
pSender
->
privateTerm
,
pSender
->
replicaIndex
,
host
,
port
);
return
s
;
}
...
...
@@ -429,11 +448,10 @@ bool snapshotReceiverIsStart(SSyncSnapshotReceiver *pReceiver) { return pReceive
// static do start by privateTerm, pBeginMsg
// receive first snapshot data
// write first block data
static
void
snapshotReceiverDoStart
(
SSyncSnapshotReceiver
*
pReceiver
,
SyncTerm
privateTerm
,
SyncSnapshotSend
*
pBeginMsg
)
{
static
void
snapshotReceiverDoStart
(
SSyncSnapshotReceiver
*
pReceiver
,
SyncSnapshotSend
*
pBeginMsg
)
{
// update state
pReceiver
->
term
=
pReceiver
->
pSyncNode
->
pRaftStore
->
currentTerm
;
pReceiver
->
privateTerm
=
privateTerm
;
pReceiver
->
privateTerm
=
p
BeginMsg
->
p
rivateTerm
;
pReceiver
->
ack
=
SYNC_SNAPSHOT_SEQ_BEGIN
;
pReceiver
->
fromId
=
pBeginMsg
->
srcId
;
pReceiver
->
start
=
true
;
...
...
@@ -445,7 +463,7 @@ static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver, SyncTerm p
pReceiver
->
snapshotParam
.
start
=
pBeginMsg
->
beginIndex
;
pReceiver
->
snapshotParam
.
end
=
pBeginMsg
->
lastIndex
;
//
write data
//
start writer
ASSERT
(
pReceiver
->
pWriter
==
NULL
);
int32_t
ret
=
pReceiver
->
pSyncNode
->
pFsm
->
FpSnapshotStartWrite
(
pReceiver
->
pSyncNode
->
pFsm
,
&
(
pReceiver
->
snapshotParam
),
&
(
pReceiver
->
pWriter
));
...
...
@@ -481,10 +499,10 @@ static void snapshotReceiverForceStop(SSyncSnapshotReceiver *pReceiver) {
// if receiver receive msg from seq = SYNC_SNAPSHOT_SEQ_BEGIN, start receiver
// if already start, force close, start again
int32_t
snapshotReceiverStart
(
SSyncSnapshotReceiver
*
pReceiver
,
Sync
Term
privateTerm
,
Sync
SnapshotSend
*
pBeginMsg
)
{
int32_t
snapshotReceiverStart
(
SSyncSnapshotReceiver
*
pReceiver
,
SyncSnapshotSend
*
pBeginMsg
)
{
if
(
!
snapshotReceiverIsStart
(
pReceiver
))
{
// first start
snapshotReceiverDoStart
(
pReceiver
,
p
rivateTerm
,
p
BeginMsg
);
snapshotReceiverDoStart
(
pReceiver
,
pBeginMsg
);
}
else
{
// already start
...
...
@@ -494,12 +512,14 @@ int32_t snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncTerm private
snapshotReceiverForceStop
(
pReceiver
);
// start again
snapshotReceiverDoStart
(
pReceiver
,
p
rivateTerm
,
p
BeginMsg
);
snapshotReceiverDoStart
(
pReceiver
,
pBeginMsg
);
}
return
0
;
}
// just set start = false
// FpSnapshotStopWrite should not be called, assert writer == NULL
int32_t
snapshotReceiverStop
(
SSyncSnapshotReceiver
*
pReceiver
)
{
if
(
pReceiver
->
pWriter
!=
NULL
)
{
int32_t
ret
=
...
...
@@ -522,6 +542,7 @@ int32_t snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver) {
return
0
;
}
// when recv last snapshot block, apply data into snapshot
static
int32_t
snapshotReceiverFinish
(
SSyncSnapshotReceiver
*
pReceiver
,
SyncSnapshotSend
*
pMsg
)
{
ASSERT
(
pMsg
->
seq
==
SYNC_SNAPSHOT_SEQ_END
);
...
...
@@ -550,7 +571,7 @@ static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnap
pReceiver
->
pSyncNode
->
commitIndex
=
pReceiver
->
snapshot
.
lastApplyIndex
;
}
// stop writer
// stop writer
, apply data
code
=
pReceiver
->
pSyncNode
->
pFsm
->
FpSnapshotStopWrite
(
pReceiver
->
pSyncNode
->
pFsm
,
pReceiver
->
pWriter
,
true
);
if
(
code
!=
0
)
{
syncNodeErrorLog
(
pReceiver
->
pSyncNode
,
"snapshot stop writer true error"
);
...
...
@@ -579,15 +600,20 @@ static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnap
return
0
;
}
// apply data block
// update progress
static
void
snapshotReceiverGotData
(
SSyncSnapshotReceiver
*
pReceiver
,
SyncSnapshotSend
*
pMsg
)
{
ASSERT
(
pMsg
->
seq
==
pReceiver
->
ack
+
1
);
if
(
pReceiver
->
pWriter
!=
NULL
)
{
if
(
pMsg
->
dataLen
>
0
)
{
// apply data block
int32_t
code
=
pReceiver
->
pSyncNode
->
pFsm
->
FpSnapshotDoWrite
(
pReceiver
->
pSyncNode
->
pFsm
,
pReceiver
->
pWriter
,
pMsg
->
data
,
pMsg
->
dataLen
);
ASSERT
(
code
==
0
);
}
// update progress
pReceiver
->
ack
=
pMsg
->
seq
;
// event log
...
...
@@ -665,14 +691,23 @@ char *snapshotReceiver2SimpleStr(SSyncSnapshotReceiver *pReceiver, char *event)
uint16_t
port
;
syncUtilU642Addr
(
fromId
.
addr
,
host
,
sizeof
(
host
),
&
port
);
snprintf
(
s
,
len
,
"%s {%p start:%d ack:%d term:%lu pterm:%lu from:%s:%d laindex:%ld laterm:%lu lcindex:%ld}"
,
event
,
pReceiver
,
pReceiver
->
start
,
pReceiver
->
ack
,
pReceiver
->
term
,
pReceiver
->
privateTerm
,
host
,
port
,
pReceiver
->
snapshot
.
lastApplyIndex
,
pReceiver
->
snapshot
.
lastApplyTerm
,
pReceiver
->
snapshot
.
lastConfigIndex
);
snprintf
(
s
,
len
,
"%s {%p start:%d ack:%d term:%lu pterm:%lu from:%s:%d s-param:%ld e-param:%ld laindex:%ld laterm:%lu "
"lcindex:%ld}"
,
event
,
pReceiver
,
pReceiver
->
start
,
pReceiver
->
ack
,
pReceiver
->
term
,
pReceiver
->
privateTerm
,
host
,
port
,
pReceiver
->
snapshotParam
.
start
,
pReceiver
->
snapshotParam
.
end
,
pReceiver
->
snapshot
.
lastApplyIndex
,
pReceiver
->
snapshot
.
lastApplyTerm
,
pReceiver
->
snapshot
.
lastConfigIndex
);
return
s
;
}
// receiver do something
// receiver on message
//
// condition 1, recv SYNC_SNAPSHOT_SEQ_BEGIN, start receiver, update privateTerm
// condition 2, recv SYNC_SNAPSHOT_SEQ_END, finish receiver(apply snapshot data, update commit index, maybe reconfig)
// condition 3, recv SYNC_SNAPSHOT_SEQ_FORCE_CLOSE, force close
// condition 4, got data, update ack
//
int32_t
syncNodeOnSnapshotSendCb
(
SSyncNode
*
pSyncNode
,
SyncSnapshotSend
*
pMsg
)
{
// get receiver
SSyncSnapshotReceiver
*
pReceiver
=
pSyncNode
->
pNewNodeReceiver
;
...
...
@@ -683,11 +718,13 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
if
(
pSyncNode
->
state
==
TAOS_SYNC_STATE_FOLLOWER
)
{
if
(
pMsg
->
term
==
pSyncNode
->
pRaftStore
->
currentTerm
)
{
if
(
pMsg
->
seq
==
SYNC_SNAPSHOT_SEQ_BEGIN
)
{
// condition 1
// begin, no data
snapshotReceiverStart
(
pReceiver
,
pMsg
->
privateTerm
,
pMsg
);
snapshotReceiverStart
(
pReceiver
,
pMsg
);
needRsp
=
true
;
}
else
if
(
pMsg
->
seq
==
SYNC_SNAPSHOT_SEQ_END
)
{
// condition 2
// end, finish FSM
code
=
snapshotReceiverFinish
(
pReceiver
,
pMsg
);
if
(
code
==
0
)
{
...
...
@@ -697,7 +734,6 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
// maybe update lastconfig
if
(
pMsg
->
lastConfigIndex
>=
SYNC_INDEX_BEGIN
)
{
// int32_t oldReplicaNum = pSyncNode->replicaNum;
SSyncCfg
oldSyncCfg
=
pSyncNode
->
pRaftCfg
->
cfg
;
// update new config myIndex
...
...
@@ -709,11 +745,13 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
}
}
else
if
(
pMsg
->
seq
==
SYNC_SNAPSHOT_SEQ_FORCE_CLOSE
)
{
// condition 3
// force close
snapshotReceiverForceStop
(
pReceiver
);
needRsp
=
false
;
}
else
if
(
pMsg
->
seq
>
SYNC_SNAPSHOT_SEQ_BEGIN
&&
pMsg
->
seq
<
SYNC_SNAPSHOT_SEQ_END
)
{
// condition 4
// transfering
if
(
pMsg
->
seq
==
pReceiver
->
ack
+
1
)
{
snapshotReceiverGotData
(
pReceiver
,
pMsg
);
...
...
@@ -752,6 +790,7 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
syncNodeSendMsgById
(
&
(
pRspMsg
->
destId
),
pSyncNode
,
&
rpcMsg
);
syncSnapshotRspDestroy
(
pRspMsg
);
}
}
else
{
// error log
do
{
...
...
@@ -759,6 +798,8 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
syncNodeErrorLog
(
pSyncNode
,
eventLog
);
taosMemoryFree
(
eventLog
);
}
while
(
0
);
return
-
1
;
}
}
else
{
// error log
...
...
@@ -767,19 +808,19 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
syncNodeErrorLog
(
pSyncNode
,
eventLog
);
taosMemoryFree
(
eventLog
);
}
while
(
0
);
return
-
1
;
}
return
0
;
}
static
void
snapshotSenderUpdateProgress
(
SSyncSnapshotSender
*
pSender
,
SyncSnapshotRsp
*
pMsg
)
{
ASSERT
(
pMsg
->
ack
==
pSender
->
seq
);
pSender
->
ack
=
pMsg
->
ack
;
++
(
pSender
->
seq
);
}
// sender receives ack, set seq = ack + 1, send msg from seq
// if ack == SYNC_SNAPSHOT_SEQ_END, stop sender
// sender on message
//
// condition 1 sender receives SYNC_SNAPSHOT_SEQ_END, close sender
// condition 2 sender receives ack, set seq = ack + 1, send msg from seq
// condition 3 sender receives error msg, just print error log
//
int32_t
syncNodeOnSnapshotRspCb
(
SSyncNode
*
pSyncNode
,
SyncSnapshotRsp
*
pMsg
)
{
// if already drop replica, do not process
if
(
!
syncNodeInRaftGroup
(
pSyncNode
,
&
(
pMsg
->
srcId
))
&&
pSyncNode
->
state
==
TAOS_SYNC_STATE_LEADER
)
{
...
...
@@ -794,12 +835,14 @@ int32_t syncNodeOnSnapshotRspCb(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) {
// state, term, seq/ack
if
(
pSyncNode
->
state
==
TAOS_SYNC_STATE_LEADER
)
{
if
(
pMsg
->
term
==
pSyncNode
->
pRaftStore
->
currentTerm
)
{
// receiver ack is finish, close sender
// condition 1
// receive ack is finish, close sender
if
(
pMsg
->
ack
==
SYNC_SNAPSHOT_SEQ_END
)
{
snapshotSenderStop
(
pSender
,
true
);
return
0
;
}
// condition 2
// send next msg
if
(
pMsg
->
ack
==
pSender
->
seq
)
{
// update sender ack
...
...
@@ -807,6 +850,7 @@ int32_t syncNodeOnSnapshotRspCb(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) {
snapshotSend
(
pSender
);
}
else
if
(
pMsg
->
ack
==
pSender
->
seq
-
1
)
{
// maybe resend
snapshotReSend
(
pSender
);
}
else
{
...
...
source/libs/sync/test/syncIndexTest.cpp
浏览文件 @
93962a8c
...
...
@@ -8,7 +8,13 @@ void print(SHashObj *pNextIndex) {
printf
(
"----------------
\n
"
);
uint64_t
*
p
=
(
uint64_t
*
)
taosHashIterate
(
pNextIndex
,
NULL
);
while
(
p
)
{
printf
(
"%lu
\n
"
,
*
p
);
size_t
len
;
void
*
key
=
taosHashGetKey
(
p
,
&
len
);
SRaftId
*
pRaftId
=
(
SRaftId
*
)
key
;
printf
(
"key:<%lu, %d>, value:%lu
\n
"
,
pRaftId
->
addr
,
pRaftId
->
vgId
,
*
p
);
p
=
(
uint64_t
*
)
taosHashIterate
(
pNextIndex
,
p
);
}
}
...
...
source/libs/sync/test/syncRespMgrTest.cpp
浏览文件 @
93962a8c
...
...
@@ -73,9 +73,15 @@ void syncRespMgrGetAndDelTest(uint64_t i) {
}
}
SSyncNode
*
createSyncNode
()
{
SSyncNode
*
pSyncNode
=
(
SSyncNode
*
)
taosMemoryMalloc
(
sizeof
(
SSyncNode
));
memset
(
pSyncNode
,
0
,
sizeof
(
SSyncNode
));
return
pSyncNode
;
}
void
test1
()
{
printf
(
"------- test1 ---------
\n
"
);
pMgr
=
syncRespMgrCreate
(
NULL
,
0
);
pMgr
=
syncRespMgrCreate
(
createSyncNode
()
,
0
);
assert
(
pMgr
!=
NULL
);
syncRespMgrInsert
(
10
);
...
...
@@ -100,7 +106,7 @@ void test1() {
void
test2
()
{
printf
(
"------- test2 ---------
\n
"
);
pMgr
=
syncRespMgrCreate
(
NULL
,
0
);
pMgr
=
syncRespMgrCreate
(
createSyncNode
()
,
0
);
assert
(
pMgr
!=
NULL
);
syncRespMgrInsert
(
10
);
...
...
@@ -117,7 +123,7 @@ void test2() {
void
test3
()
{
printf
(
"------- test3 ---------
\n
"
);
pMgr
=
syncRespMgrCreate
(
NULL
,
0
);
pMgr
=
syncRespMgrCreate
(
createSyncNode
()
,
0
);
assert
(
pMgr
!=
NULL
);
syncRespMgrInsert
(
10
);
...
...
@@ -132,13 +138,34 @@ void test3() {
syncRespMgrDestroy
(
pMgr
);
}
void
test4
()
{
printf
(
"------- test4 ---------
\n
"
);
pMgr
=
syncRespMgrCreate
(
createSyncNode
(),
2
);
assert
(
pMgr
!=
NULL
);
syncRespMgrInsert
(
5
);
syncRespMgrPrint
();
taosMsleep
(
3000
);
syncRespMgrInsert
(
3
);
syncRespMgrPrint
();
printf
(
"====== after clean ttl
\n
"
);
syncRespClean
(
pMgr
);
syncRespMgrPrint
();
syncRespMgrDestroy
(
pMgr
);
}
int
main
()
{
tsAsyncLog
=
0
;
sDebugFlag
=
DEBUG_TRACE
+
DEBUG_SCREEN
+
DEBUG_FILE
;
sDebugFlag
=
DEBUG_
DEBUG
+
DEBUG_
TRACE
+
DEBUG_SCREEN
+
DEBUG_FILE
;
logTest
();
test1
();
test2
();
test3
();
test4
();
return
0
;
}
tests/script/tsim/stable/tag_modify.sim
浏览文件 @
93962a8c
...
...
@@ -14,7 +14,7 @@ sql_error alter table db.stb MODIFY tag ts int
sql_error alter table db.stb MODIFY tag t2 binary(3)
sql_error alter table db.stb MODIFY tag t2 int
sql_error alter table db.stb MODIFY tag t1 int
sql create table db.ctb using db.stb tags(101, "123
45
")
sql create table db.ctb using db.stb tags(101, "123")
sql insert into db.ctb values(now, 1, "1234")
sql select * from db.stb
...
...
@@ -32,7 +32,7 @@ endi
if $data[0][3] != 101 then
return -1
endi
if $data[0][4] != 123
4
then
if $data[0][4] != 123 then
return -1
endi
...
...
tests/script/tsim/stable/tag_rename.sim
浏览文件 @
93962a8c
...
...
@@ -14,7 +14,7 @@ sql_error alter table db.stb rename tag ts c3
sql_error alter table db.stb rename tag t2 t1
sql_error alter table db.stb rename tag t2 t2
sql_error alter table db.stb rename tag t1 t2
sql create table db.ctb using db.stb tags(101, "123
45
")
sql create table db.ctb using db.stb tags(101, "123")
sql insert into db.ctb values(now, 1, "1234")
sql select * from db.stb
...
...
@@ -32,7 +32,7 @@ endi
if $data[0][3] != 101 then
return -1
endi
if $data[0][4] != 123
4
then
if $data[0][4] != 123 then
return -1
endi
...
...
@@ -56,7 +56,7 @@ endi
if $data[0][3] != 101 then
return -1
endi
if $data[0][4] != 123
4
then
if $data[0][4] != 123 then
return -1
endi
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录