Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
916c2973
T
TDengine
项目概览
taosdata
/
TDengine
1 年多 前同步成功
通知
1185
Star
22016
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
提交
916c2973
编写于
12月 27, 2022
作者:
H
Hongze Cheng
浏览文件
操作
浏览文件
下载
差异文件
Merge branch 'feat/vnode_compact' of
https://github.com/taosdata/TDengine
into feat/vnode_compact
上级
1427f921
344696b9
变更
8
隐藏空白更改
内联
并排
Showing
8 changed file
with
152 addition
and
13 deletion
+152
-13
include/common/tmsg.h
include/common/tmsg.h
+1
-1
source/common/src/systable.c
source/common/src/systable.c
+1
-0
source/common/src/tmsg.c
source/common/src/tmsg.c
+2
-6
source/dnode/mnode/impl/inc/mndDef.h
source/dnode/mnode/impl/inc/mndDef.h
+1
-0
source/dnode/mnode/impl/inc/mndVgroup.h
source/dnode/mnode/impl/inc/mndVgroup.h
+1
-0
source/dnode/mnode/impl/src/mndDb.c
source/dnode/mnode/impl/src/mndDb.c
+64
-4
source/dnode/mnode/impl/src/mndVgroup.c
source/dnode/mnode/impl/src/mndVgroup.c
+63
-0
source/dnode/vnode/src/vnd/vnodeSvr.c
source/dnode/vnode/src/vnd/vnodeSvr.c
+19
-2
未找到文件。
include/common/tmsg.h
浏览文件 @
916c2973
...
@@ -1233,7 +1233,7 @@ int32_t tDeserializeSDropVnodeReq(void* buf, int32_t bufLen, SDropVnodeReq* pReq
...
@@ -1233,7 +1233,7 @@ int32_t tDeserializeSDropVnodeReq(void* buf, int32_t bufLen, SDropVnodeReq* pReq
typedef
struct
{
typedef
struct
{
int64_t
dbUid
;
int64_t
dbUid
;
char
db
[
TSDB_DB_FNAME_LEN
];
char
db
[
TSDB_DB_FNAME_LEN
];
int64_t
reserved
[
8
]
;
int64_t
compactStartTime
;
}
SCompactVnodeReq
;
}
SCompactVnodeReq
;
int32_t
tSerializeSCompactVnodeReq
(
void
*
buf
,
int32_t
bufLen
,
SCompactVnodeReq
*
pReq
);
int32_t
tSerializeSCompactVnodeReq
(
void
*
buf
,
int32_t
bufLen
,
SCompactVnodeReq
*
pReq
);
...
...
source/common/src/systable.c
浏览文件 @
916c2973
...
@@ -216,6 +216,7 @@ static const SSysDbTableSchema vgroupsSchema[] = {
...
@@ -216,6 +216,7 @@ static const SSysDbTableSchema vgroupsSchema[] = {
{.
name
=
"v4_status"
,
.
bytes
=
9
+
VARSTR_HEADER_SIZE
,
.
type
=
TSDB_DATA_TYPE_VARCHAR
,
.
sysInfo
=
true
},
{.
name
=
"v4_status"
,
.
bytes
=
9
+
VARSTR_HEADER_SIZE
,
.
type
=
TSDB_DATA_TYPE_VARCHAR
,
.
sysInfo
=
true
},
{.
name
=
"cacheload"
,
.
bytes
=
4
,
.
type
=
TSDB_DATA_TYPE_INT
,
.
sysInfo
=
true
},
{.
name
=
"cacheload"
,
.
bytes
=
4
,
.
type
=
TSDB_DATA_TYPE_INT
,
.
sysInfo
=
true
},
{.
name
=
"tsma"
,
.
bytes
=
1
,
.
type
=
TSDB_DATA_TYPE_TINYINT
,
.
sysInfo
=
true
},
{.
name
=
"tsma"
,
.
bytes
=
1
,
.
type
=
TSDB_DATA_TYPE_TINYINT
,
.
sysInfo
=
true
},
{.
name
=
"compact_start_time"
,
.
bytes
=
8
,
.
type
=
TSDB_DATA_TYPE_TIMESTAMP
,
.
sysInfo
=
false
},
};
};
static
const
SSysDbTableSchema
smaSchema
[]
=
{
static
const
SSysDbTableSchema
smaSchema
[]
=
{
...
...
source/common/src/tmsg.c
浏览文件 @
916c2973
...
@@ -3989,9 +3989,7 @@ int32_t tSerializeSCompactVnodeReq(void *buf, int32_t bufLen, SCompactVnodeReq *
...
@@ -3989,9 +3989,7 @@ int32_t tSerializeSCompactVnodeReq(void *buf, int32_t bufLen, SCompactVnodeReq *
if
(
tStartEncode
(
&
encoder
)
<
0
)
return
-
1
;
if
(
tStartEncode
(
&
encoder
)
<
0
)
return
-
1
;
if
(
tEncodeI64
(
&
encoder
,
pReq
->
dbUid
)
<
0
)
return
-
1
;
if
(
tEncodeI64
(
&
encoder
,
pReq
->
dbUid
)
<
0
)
return
-
1
;
if
(
tEncodeCStr
(
&
encoder
,
pReq
->
db
)
<
0
)
return
-
1
;
if
(
tEncodeCStr
(
&
encoder
,
pReq
->
db
)
<
0
)
return
-
1
;
for
(
int32_t
i
=
0
;
i
<
8
;
++
i
)
{
if
(
tEncodeI64
(
&
encoder
,
pReq
->
compactStartTime
)
<
0
)
return
-
1
;
if
(
tEncodeI64
(
&
encoder
,
pReq
->
reserved
[
i
])
<
0
)
return
-
1
;
}
tEndEncode
(
&
encoder
);
tEndEncode
(
&
encoder
);
int32_t
tlen
=
encoder
.
pos
;
int32_t
tlen
=
encoder
.
pos
;
...
@@ -4006,9 +4004,7 @@ int32_t tDeserializeSCompactVnodeReq(void *buf, int32_t bufLen, SCompactVnodeReq
...
@@ -4006,9 +4004,7 @@ int32_t tDeserializeSCompactVnodeReq(void *buf, int32_t bufLen, SCompactVnodeReq
if
(
tStartDecode
(
&
decoder
)
<
0
)
return
-
1
;
if
(
tStartDecode
(
&
decoder
)
<
0
)
return
-
1
;
if
(
tDecodeI64
(
&
decoder
,
&
pReq
->
dbUid
)
<
0
)
return
-
1
;
if
(
tDecodeI64
(
&
decoder
,
&
pReq
->
dbUid
)
<
0
)
return
-
1
;
if
(
tDecodeCStrTo
(
&
decoder
,
pReq
->
db
)
<
0
)
return
-
1
;
if
(
tDecodeCStrTo
(
&
decoder
,
pReq
->
db
)
<
0
)
return
-
1
;
for
(
int32_t
i
=
0
;
i
<
8
;
++
i
)
{
if
(
tDecodeI64
(
&
decoder
,
&
pReq
->
compactStartTime
)
<
0
)
return
-
1
;
if
(
tDecodeI64
(
&
decoder
,
&
pReq
->
reserved
[
i
])
<
0
)
return
-
1
;
}
tEndDecode
(
&
decoder
);
tEndDecode
(
&
decoder
);
tDecoderClear
(
&
decoder
);
tDecoderClear
(
&
decoder
);
...
...
source/dnode/mnode/impl/inc/mndDef.h
浏览文件 @
916c2973
...
@@ -327,6 +327,7 @@ typedef struct {
...
@@ -327,6 +327,7 @@ typedef struct {
SDbCfg
cfg
;
SDbCfg
cfg
;
SRWLatch
lock
;
SRWLatch
lock
;
int64_t
stateTs
;
int64_t
stateTs
;
int64_t
compactStartTime
;
}
SDbObj
;
}
SDbObj
;
typedef
struct
{
typedef
struct
{
...
...
source/dnode/mnode/impl/inc/mndVgroup.h
浏览文件 @
916c2973
...
@@ -43,6 +43,7 @@ int32_t mndAddDropVnodeAction(SMnode *, STrans *pTrans, SDbObj *pDb, SVgObj *pVg
...
@@ -43,6 +43,7 @@ int32_t mndAddDropVnodeAction(SMnode *, STrans *pTrans, SDbObj *pDb, SVgObj *pVg
int32_t
mndSetMoveVgroupsInfoToTrans
(
SMnode
*
,
STrans
*
pTrans
,
int32_t
dropDnodeId
,
bool
force
);
int32_t
mndSetMoveVgroupsInfoToTrans
(
SMnode
*
,
STrans
*
pTrans
,
int32_t
dropDnodeId
,
bool
force
);
int32_t
mndBuildAlterVgroupAction
(
SMnode
*
pMnode
,
STrans
*
pTrans
,
SDbObj
*
pOldDb
,
SDbObj
*
pNewDb
,
SVgObj
*
pVgroup
,
int32_t
mndBuildAlterVgroupAction
(
SMnode
*
pMnode
,
STrans
*
pTrans
,
SDbObj
*
pOldDb
,
SDbObj
*
pNewDb
,
SVgObj
*
pVgroup
,
SArray
*
pArray
);
SArray
*
pArray
);
int32_t
mndBuildCompactVgroupAction
(
SMnode
*
pMnode
,
STrans
*
pTrans
,
SDbObj
*
pDb
,
SVgObj
*
pVgroup
,
int64_t
compactTs
);
void
*
mndBuildCreateVnodeReq
(
SMnode
*
,
SDnodeObj
*
pDnode
,
SDbObj
*
pDb
,
SVgObj
*
pVgroup
,
int32_t
*
pContLen
);
void
*
mndBuildCreateVnodeReq
(
SMnode
*
,
SDnodeObj
*
pDnode
,
SDbObj
*
pDb
,
SVgObj
*
pVgroup
,
int32_t
*
pContLen
);
void
*
mndBuildDropVnodeReq
(
SMnode
*
,
SDnodeObj
*
pDnode
,
SDbObj
*
pDb
,
SVgObj
*
pVgroup
,
int32_t
*
pContLen
);
void
*
mndBuildDropVnodeReq
(
SMnode
*
,
SDnodeObj
*
pDnode
,
SDbObj
*
pDb
,
SVgObj
*
pVgroup
,
int32_t
*
pContLen
);
...
...
source/dnode/mnode/impl/src/mndDb.c
浏览文件 @
916c2973
...
@@ -30,7 +30,7 @@
...
@@ -30,7 +30,7 @@
#include "systable.h"
#include "systable.h"
#define DB_VER_NUMBER 1
#define DB_VER_NUMBER 1
#define DB_RESERVE_SIZE
54
#define DB_RESERVE_SIZE
46
static
SSdbRaw
*
mndDbActionEncode
(
SDbObj
*
pDb
);
static
SSdbRaw
*
mndDbActionEncode
(
SDbObj
*
pDb
);
static
SSdbRow
*
mndDbActionDecode
(
SSdbRaw
*
pRaw
);
static
SSdbRow
*
mndDbActionDecode
(
SSdbRaw
*
pRaw
);
...
@@ -128,6 +128,7 @@ static SSdbRaw *mndDbActionEncode(SDbObj *pDb) {
...
@@ -128,6 +128,7 @@ static SSdbRaw *mndDbActionEncode(SDbObj *pDb) {
SDB_SET_INT16
(
pRaw
,
dataPos
,
pDb
->
cfg
.
hashPrefix
,
_OVER
)
SDB_SET_INT16
(
pRaw
,
dataPos
,
pDb
->
cfg
.
hashPrefix
,
_OVER
)
SDB_SET_INT16
(
pRaw
,
dataPos
,
pDb
->
cfg
.
hashSuffix
,
_OVER
)
SDB_SET_INT16
(
pRaw
,
dataPos
,
pDb
->
cfg
.
hashSuffix
,
_OVER
)
SDB_SET_INT32
(
pRaw
,
dataPos
,
pDb
->
cfg
.
tsdbPageSize
,
_OVER
)
SDB_SET_INT32
(
pRaw
,
dataPos
,
pDb
->
cfg
.
tsdbPageSize
,
_OVER
)
SDB_SET_INT64
(
pRaw
,
dataPos
,
pDb
->
compactStartTime
,
_OVER
)
SDB_SET_RESERVE
(
pRaw
,
dataPos
,
DB_RESERVE_SIZE
,
_OVER
)
SDB_SET_RESERVE
(
pRaw
,
dataPos
,
DB_RESERVE_SIZE
,
_OVER
)
SDB_SET_DATALEN
(
pRaw
,
dataPos
,
_OVER
)
SDB_SET_DATALEN
(
pRaw
,
dataPos
,
_OVER
)
...
@@ -217,6 +218,7 @@ static SSdbRow *mndDbActionDecode(SSdbRaw *pRaw) {
...
@@ -217,6 +218,7 @@ static SSdbRow *mndDbActionDecode(SSdbRaw *pRaw) {
SDB_GET_INT16
(
pRaw
,
dataPos
,
&
pDb
->
cfg
.
hashPrefix
,
_OVER
)
SDB_GET_INT16
(
pRaw
,
dataPos
,
&
pDb
->
cfg
.
hashPrefix
,
_OVER
)
SDB_GET_INT16
(
pRaw
,
dataPos
,
&
pDb
->
cfg
.
hashSuffix
,
_OVER
)
SDB_GET_INT16
(
pRaw
,
dataPos
,
&
pDb
->
cfg
.
hashSuffix
,
_OVER
)
SDB_GET_INT32
(
pRaw
,
dataPos
,
&
pDb
->
cfg
.
tsdbPageSize
,
_OVER
)
SDB_GET_INT32
(
pRaw
,
dataPos
,
&
pDb
->
cfg
.
tsdbPageSize
,
_OVER
)
SDB_GET_INT64
(
pRaw
,
dataPos
,
&
pDb
->
compactStartTime
,
_OVER
)
SDB_GET_RESERVE
(
pRaw
,
dataPos
,
DB_RESERVE_SIZE
,
_OVER
)
SDB_GET_RESERVE
(
pRaw
,
dataPos
,
DB_RESERVE_SIZE
,
_OVER
)
taosInitRWLatch
(
&
pDb
->
lock
);
taosInitRWLatch
(
&
pDb
->
lock
);
...
@@ -275,6 +277,7 @@ static int32_t mndDbActionUpdate(SSdb *pSdb, SDbObj *pOld, SDbObj *pNew) {
...
@@ -275,6 +277,7 @@ static int32_t mndDbActionUpdate(SSdb *pSdb, SDbObj *pOld, SDbObj *pNew) {
pOld
->
cfg
.
replications
=
pNew
->
cfg
.
replications
;
pOld
->
cfg
.
replications
=
pNew
->
cfg
.
replications
;
pOld
->
cfg
.
sstTrigger
=
pNew
->
cfg
.
sstTrigger
;
pOld
->
cfg
.
sstTrigger
=
pNew
->
cfg
.
sstTrigger
;
pOld
->
cfg
.
tsdbPageSize
=
pNew
->
cfg
.
tsdbPageSize
;
pOld
->
cfg
.
tsdbPageSize
=
pNew
->
cfg
.
tsdbPageSize
;
pOld
->
compactStartTime
=
pNew
->
compactStartTime
;
taosWUnLockLatch
(
&
pOld
->
lock
);
taosWUnLockLatch
(
&
pOld
->
lock
);
return
0
;
return
0
;
}
}
...
@@ -1388,7 +1391,63 @@ int32_t mndValidateDbInfo(SMnode *pMnode, SDbVgVersion *pDbs, int32_t numOfDbs,
...
@@ -1388,7 +1391,63 @@ int32_t mndValidateDbInfo(SMnode *pMnode, SDbVgVersion *pDbs, int32_t numOfDbs,
return
0
;
return
0
;
}
}
static
int32_t
mndCompactDb
(
SMnode
*
pMnode
,
SDbObj
*
pDb
)
{
return
0
;
}
static
int32_t
mndSetCompactDbCommitLogs
(
SMnode
*
pMnode
,
STrans
*
pTrans
,
SDbObj
*
pDb
,
int64_t
compactTs
)
{
SDbObj
dbObj
=
{
0
};
memcpy
(
&
dbObj
,
pDb
,
sizeof
(
SDbObj
));
dbObj
.
compactStartTime
=
compactTs
;
SSdbRaw
*
pCommitRaw
=
mndDbActionEncode
(
&
dbObj
);
if
(
pCommitRaw
==
NULL
)
return
-
1
;
if
(
mndTransAppendCommitlog
(
pTrans
,
pCommitRaw
)
!=
0
)
{
sdbFreeRaw
(
pCommitRaw
);
return
-
1
;
}
(
void
)
sdbSetRawStatus
(
pCommitRaw
,
SDB_STATUS_READY
);
return
0
;
}
static
int32_t
mndSetCompactDbRedoActions
(
SMnode
*
pMnode
,
STrans
*
pTrans
,
SDbObj
*
pDb
,
int64_t
compactTs
)
{
SSdb
*
pSdb
=
pMnode
->
pSdb
;
void
*
pIter
=
NULL
;
while
(
1
)
{
SVgObj
*
pVgroup
=
NULL
;
pIter
=
sdbFetch
(
pSdb
,
SDB_VGROUP
,
pIter
,
(
void
**
)
&
pVgroup
);
if
(
pIter
==
NULL
)
break
;
if
(
mndVgroupInDb
(
pVgroup
,
pDb
->
uid
))
{
if
(
mndBuildCompactVgroupAction
(
pMnode
,
pTrans
,
pDb
,
pVgroup
,
compactTs
)
!=
0
)
{
sdbCancelFetch
(
pSdb
,
pIter
);
sdbRelease
(
pSdb
,
pVgroup
);
return
-
1
;
}
}
sdbRelease
(
pSdb
,
pVgroup
);
}
return
0
;
}
static
int32_t
mndCompactDb
(
SMnode
*
pMnode
,
SRpcMsg
*
pReq
,
SDbObj
*
pDb
)
{
int64_t
compactTs
=
taosGetTimestampMs
();
int32_t
code
=
-
1
;
STrans
*
pTrans
=
mndTransCreate
(
pMnode
,
TRN_POLICY_RETRY
,
TRN_CONFLICT_DB
,
pReq
,
"compact-db"
);
if
(
pTrans
==
NULL
)
goto
_OVER
;
mInfo
(
"trans:%d, used to compact db:%s"
,
pTrans
->
id
,
pDb
->
name
);
mndTransSetDbName
(
pTrans
,
pDb
->
name
,
NULL
);
if
(
mndTrancCheckConflict
(
pMnode
,
pTrans
)
!=
0
)
goto
_OVER
;
if
(
mndSetCompactDbCommitLogs
(
pMnode
,
pTrans
,
pDb
,
compactTs
)
!=
0
)
goto
_OVER
;
if
(
mndSetCompactDbRedoActions
(
pMnode
,
pTrans
,
pDb
,
compactTs
)
!=
0
)
goto
_OVER
;
if
(
mndTransPrepare
(
pMnode
,
pTrans
)
!=
0
)
goto
_OVER
;
code
=
0
;
_OVER:
mndTransDrop
(
pTrans
);
return
code
;
}
static
int32_t
mndProcessCompactDbReq
(
SRpcMsg
*
pReq
)
{
static
int32_t
mndProcessCompactDbReq
(
SRpcMsg
*
pReq
)
{
SMnode
*
pMnode
=
pReq
->
info
.
node
;
SMnode
*
pMnode
=
pReq
->
info
.
node
;
...
@@ -1412,10 +1471,11 @@ static int32_t mndProcessCompactDbReq(SRpcMsg *pReq) {
...
@@ -1412,10 +1471,11 @@ static int32_t mndProcessCompactDbReq(SRpcMsg *pReq) {
goto
_OVER
;
goto
_OVER
;
}
}
code
=
mndCompactDb
(
pMnode
,
pDb
);
code
=
mndCompactDb
(
pMnode
,
pReq
,
pDb
);
if
(
code
==
0
)
code
=
TSDB_CODE_ACTION_IN_PROGRESS
;
_OVER:
_OVER:
if
(
code
!=
0
)
{
if
(
code
!=
0
&&
code
!=
TSDB_CODE_ACTION_IN_PROGRESS
)
{
mError
(
"db:%s, failed to process compact db req since %s"
,
compactReq
.
db
,
terrstr
());
mError
(
"db:%s, failed to process compact db req since %s"
,
compactReq
.
db
,
terrstr
());
}
}
...
...
source/dnode/mnode/impl/src/mndVgroup.c
浏览文件 @
916c2973
...
@@ -729,6 +729,13 @@ static int32_t mndRetrieveVgroups(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *p
...
@@ -729,6 +729,13 @@ static int32_t mndRetrieveVgroups(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *p
pColInfo
=
taosArrayGet
(
pBlock
->
pDataBlock
,
cols
++
);
pColInfo
=
taosArrayGet
(
pBlock
->
pDataBlock
,
cols
++
);
colDataAppend
(
pColInfo
,
numOfRows
,
(
const
char
*
)
&
pVgroup
->
isTsma
,
false
);
colDataAppend
(
pColInfo
,
numOfRows
,
(
const
char
*
)
&
pVgroup
->
isTsma
,
false
);
pColInfo
=
taosArrayGet
(
pBlock
->
pDataBlock
,
cols
++
);
if
(
pDb
->
compactStartTime
<=
0
)
{
colDataAppendNULL
(
pColInfo
,
rows
);
}
else
{
colDataAppend
(
pColInfo
,
numOfRows
,
(
const
char
*
)
&
pDb
->
compactStartTime
,
false
);
}
numOfRows
++
;
numOfRows
++
;
sdbRelease
(
pSdb
,
pVgroup
);
sdbRelease
(
pSdb
,
pVgroup
);
}
}
...
@@ -2066,3 +2073,59 @@ _OVER:
...
@@ -2066,3 +2073,59 @@ _OVER:
}
}
bool
mndVgroupInDb
(
SVgObj
*
pVgroup
,
int64_t
dbUid
)
{
return
!
pVgroup
->
isTsma
&&
pVgroup
->
dbUid
==
dbUid
;
}
bool
mndVgroupInDb
(
SVgObj
*
pVgroup
,
int64_t
dbUid
)
{
return
!
pVgroup
->
isTsma
&&
pVgroup
->
dbUid
==
dbUid
;
}
static
void
*
mndBuildCompactVnodeReq
(
SMnode
*
pMnode
,
SDbObj
*
pDb
,
SVgObj
*
pVgroup
,
int32_t
*
pContLen
,
int64_t
compactTs
)
{
SCompactVnodeReq
compactReq
=
{
0
};
compactReq
.
dbUid
=
pDb
->
uid
;
compactReq
.
compactStartTime
=
compactTs
;
tstrncpy
(
compactReq
.
db
,
pDb
->
name
,
TSDB_DB_FNAME_LEN
);
mInfo
(
"vgId:%d, build compact vnode config req"
,
pVgroup
->
vgId
);
int32_t
contLen
=
tSerializeSCompactVnodeReq
(
NULL
,
0
,
&
compactReq
);
if
(
contLen
<
0
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
return
NULL
;
}
contLen
+=
sizeof
(
SMsgHead
);
void
*
pReq
=
taosMemoryMalloc
(
contLen
);
if
(
pReq
==
NULL
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
return
NULL
;
}
SMsgHead
*
pHead
=
pReq
;
pHead
->
contLen
=
htonl
(
contLen
);
pHead
->
vgId
=
htonl
(
pVgroup
->
vgId
);
tSerializeSCompactVnodeReq
((
char
*
)
pReq
+
sizeof
(
SMsgHead
),
contLen
,
&
compactReq
);
*
pContLen
=
contLen
;
return
pReq
;
}
static
int32_t
mndAddCompactVnodeAction
(
SMnode
*
pMnode
,
STrans
*
pTrans
,
SDbObj
*
pDb
,
SVgObj
*
pVgroup
,
int64_t
compactTs
)
{
STransAction
action
=
{
0
};
action
.
epSet
=
mndGetVgroupEpset
(
pMnode
,
pVgroup
);
int32_t
contLen
=
0
;
void
*
pReq
=
mndBuildCompactVnodeReq
(
pMnode
,
pDb
,
pVgroup
,
&
contLen
,
compactTs
);
if
(
pReq
==
NULL
)
return
-
1
;
action
.
pCont
=
pReq
;
action
.
contLen
=
contLen
;
action
.
msgType
=
TDMT_VND_COMPACT
;
if
(
mndTransAppendRedoAction
(
pTrans
,
&
action
)
!=
0
)
{
taosMemoryFree
(
pReq
);
return
-
1
;
}
return
0
;
}
int32_t
mndBuildCompactVgroupAction
(
SMnode
*
pMnode
,
STrans
*
pTrans
,
SDbObj
*
pDb
,
SVgObj
*
pVgroup
,
int64_t
compactTs
)
{
if
(
mndAddCompactVnodeAction
(
pMnode
,
pTrans
,
pDb
,
pVgroup
,
compactTs
)
!=
0
)
return
-
1
;
return
0
;
}
\ No newline at end of file
source/dnode/vnode/src/vnd/vnodeSvr.c
浏览文件 @
916c2973
...
@@ -30,6 +30,7 @@ static int32_t vnodeProcessDropTtlTbReq(SVnode *pVnode, int64_t version, void *p
...
@@ -30,6 +30,7 @@ static int32_t vnodeProcessDropTtlTbReq(SVnode *pVnode, int64_t version, void *p
static
int32_t
vnodeProcessTrimReq
(
SVnode
*
pVnode
,
int64_t
version
,
void
*
pReq
,
int32_t
len
,
SRpcMsg
*
pRsp
);
static
int32_t
vnodeProcessTrimReq
(
SVnode
*
pVnode
,
int64_t
version
,
void
*
pReq
,
int32_t
len
,
SRpcMsg
*
pRsp
);
static
int32_t
vnodeProcessDeleteReq
(
SVnode
*
pVnode
,
int64_t
version
,
void
*
pReq
,
int32_t
len
,
SRpcMsg
*
pRsp
);
static
int32_t
vnodeProcessDeleteReq
(
SVnode
*
pVnode
,
int64_t
version
,
void
*
pReq
,
int32_t
len
,
SRpcMsg
*
pRsp
);
static
int32_t
vnodeProcessBatchDeleteReq
(
SVnode
*
pVnode
,
int64_t
version
,
void
*
pReq
,
int32_t
len
,
SRpcMsg
*
pRsp
);
static
int32_t
vnodeProcessBatchDeleteReq
(
SVnode
*
pVnode
,
int64_t
version
,
void
*
pReq
,
int32_t
len
,
SRpcMsg
*
pRsp
);
static
int32_t
vnodeProcessCompactVnodeReq
(
SVnode
*
pVnode
,
int64_t
version
,
void
*
pReq
,
int32_t
len
,
SRpcMsg
*
pRsp
);
int32_t
vnodePreProcessWriteMsg
(
SVnode
*
pVnode
,
SRpcMsg
*
pMsg
)
{
int32_t
vnodePreProcessWriteMsg
(
SVnode
*
pVnode
,
SRpcMsg
*
pMsg
)
{
int32_t
code
=
0
;
int32_t
code
=
0
;
...
@@ -308,8 +309,7 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRp
...
@@ -308,8 +309,7 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRp
vnodeBegin
(
pVnode
);
vnodeBegin
(
pVnode
);
goto
_exit
;
goto
_exit
;
case
TDMT_VND_COMPACT
:
case
TDMT_VND_COMPACT
:
vnodeAsyncCompact
(
pVnode
);
vnodeProcessCompactVnodeReq
(
pVnode
,
version
,
pReq
,
len
,
pRsp
);
vnodeBegin
(
pVnode
);
goto
_exit
;
goto
_exit
;
default:
default:
vError
(
"vgId:%d, unprocessed msg, %d"
,
TD_VID
(
pVnode
),
pMsg
->
msgType
);
vError
(
"vgId:%d, unprocessed msg, %d"
,
TD_VID
(
pVnode
),
pMsg
->
msgType
);
...
@@ -1266,3 +1266,20 @@ static int32_t vnodeProcessDeleteReq(SVnode *pVnode, int64_t version, void *pReq
...
@@ -1266,3 +1266,20 @@ static int32_t vnodeProcessDeleteReq(SVnode *pVnode, int64_t version, void *pReq
_err:
_err:
return
code
;
return
code
;
}
}
static
int32_t
vnodeProcessCompactVnodeReq
(
SVnode
*
pVnode
,
int64_t
version
,
void
*
pReq
,
int32_t
len
,
SRpcMsg
*
pRsp
)
{
SCompactVnodeReq
req
=
{
0
};
if
(
tDeserializeSCompactVnodeReq
(
pReq
,
len
,
&
req
)
!=
0
)
{
terrno
=
TSDB_CODE_INVALID_MSG
;
return
TSDB_CODE_INVALID_MSG
;
}
vInfo
(
"vgId:%d, compact msg will be processed, db:%s dbUid:%"
PRId64
" compactStartTime:%"
PRId64
,
TD_VID
(
pVnode
),
req
.
db
,
req
.
dbUid
,
req
.
compactStartTime
);
#if 0
vnodeAsyncCompact(pVnode);
vnodeBegin(pVnode);
#endif
return
0
;
}
\ No newline at end of file
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录