Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
89495a32
T
TDengine
项目概览
taosdata
/
TDengine
12 个月 前同步成功
通知
1180
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,发现更多精彩内容 >>
未验证
提交
89495a32
编写于
4月 29, 2022
作者:
C
Cary Xu
提交者:
GitHub
4月 29, 2022
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #12027 from taosdata/feature/TD-14481-3.0
feat: migrate tsdb read interface to vnode
上级
210e9cdb
da4f86f3
变更
8
隐藏空白更改
内联
并排
Showing
8 changed file
with
131 addition
and
21 deletion
+131
-21
include/common/taosdef.h
include/common/taosdef.h
+7
-1
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
+3
-1
source/dnode/vnode/inc/vnode.h
source/dnode/vnode/inc/vnode.h
+12
-2
source/dnode/vnode/src/inc/vnodeInt.h
source/dnode/vnode/src/inc/vnodeInt.h
+20
-6
source/dnode/vnode/src/tsdb/tsdbOpen.c
source/dnode/vnode/src/tsdb/tsdbOpen.c
+26
-4
source/dnode/vnode/src/tsdb/tsdbSma.c
source/dnode/vnode/src/tsdb/tsdbSma.c
+2
-3
source/dnode/vnode/src/vnd/vnodeCfg.c
source/dnode/vnode/src/vnd/vnodeCfg.c
+36
-0
source/dnode/vnode/src/vnd/vnodeOpen.c
source/dnode/vnode/src/vnd/vnodeOpen.c
+25
-4
未找到文件。
include/common/taosdef.h
浏览文件 @
89495a32
...
...
@@ -71,13 +71,19 @@ typedef enum {
TSDB_SMA_STAT_DROPPED
=
2
,
// sma dropped
}
ETsdbSmaStat
;
// bit operation
typedef
enum
{
TSDB_SMA_TYPE_BLOCK
=
0
,
// Block-wise SMA
TSDB_SMA_TYPE_TIME_RANGE
=
1
,
// Time-range-wise SMA
TSDB_SMA_TYPE_ROLLUP
=
2
,
// Rollup SMA
}
ETsdbSmaType
;
typedef
enum
{
TSDB_RSMA_RETENTION_0
=
0
,
TSDB_RSMA_RETENTION_1
=
1
,
TSDB_RSMA_RETENTION_2
=
2
,
TSDB_RSMA_RETENTION_MAX
=
3
}
ERSmaRetention
;
extern
char
*
qtypeStr
[];
#define TSDB_PORT_HTTP 11
...
...
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
浏览文件 @
89495a32
...
...
@@ -112,7 +112,9 @@ static void vmGenerateVnodeCfg(SCreateVnodeReq *pCreate, SVnodeCfg *pCfg) {
pCfg
->
tsdbCfg
.
keep2
=
3650
;
pCfg
->
tsdbCfg
.
keep0
=
3650
;
pCfg
->
tsdbCfg
.
keep1
=
3650
;
pCfg
->
tsdbCfg
.
retentions
=
pCreate
->
pRetensions
;
for
(
size_t
i
=
0
;
i
<
taosArrayGetSize
(
pCreate
->
pRetensions
);
++
i
)
{
memcpy
(
&
pCfg
->
tsdbCfg
.
retentions
[
i
],
taosArrayGet
(
pCreate
->
pRetensions
,
i
),
sizeof
(
SRetention
));
}
pCfg
->
walCfg
.
vgId
=
pCreate
->
vgId
;
pCfg
->
hashBegin
=
pCreate
->
hashBegin
;
pCfg
->
hashEnd
=
pCreate
->
hashEnd
;
...
...
source/dnode/vnode/inc/vnode.h
浏览文件 @
89495a32
...
...
@@ -37,7 +37,7 @@
#ifdef __cplusplus
extern
"C"
{
#endif
#define TSDB_VNODE_SMA_DEBUG // TODO: evaluate to remove the macro and the relative codes
// vnode
typedef
struct
SVnode
SVnode
;
typedef
struct
STsdbCfg
STsdbCfg
;
// todo: remove
...
...
@@ -143,9 +143,19 @@ struct STsdbCfg {
int32_t
keep0
;
int32_t
keep1
;
int32_t
keep2
;
SArray
*
retentions
;
// TODO: save to tsdb cfg file
int8_t
type
;
// ETsdbType
SRetention
retentions
[
TSDB_RSMA_RETENTION_MAX
];
};
typedef
enum
{
TSDB_TYPE_TSDB
=
0
,
// TSDB
TSDB_TYPE_TSMA
=
1
,
// TSMA
TSDB_TYPE_RSMA_L0
=
2
,
// RSMA Level 0
TSDB_TYPE_RSMA_L1
=
3
,
// RSMA Level 1
TSDB_TYPE_RSMA_L2
=
4
,
// RSMA Level 2
}
ETsdbType
;
struct
SVnodeCfg
{
int32_t
vgId
;
char
dbname
[
TSDB_DB_NAME_LEN
];
...
...
source/dnode/vnode/src/inc/vnodeInt.h
浏览文件 @
89495a32
...
...
@@ -54,10 +54,13 @@ typedef struct SVState SVState;
typedef
struct
SVBufPool
SVBufPool
;
typedef
struct
SQWorker
SQHandle
;
#define VNODE_META_DIR "meta"
#define VNODE_TSDB_DIR "tsdb"
#define VNODE_TQ_DIR "tq"
#define VNODE_WAL_DIR "wal"
#define VNODE_META_DIR "meta"
#define VNODE_TSDB_DIR "tsdb"
#define VNODE_TQ_DIR "tq"
#define VNODE_WAL_DIR "wal"
#define VNODE_TSMA_DIR "tsma"
#define VNODE_RSMA1_DIR "rsma1"
#define VNODE_RSMA2_DIR "rsma2"
// vnd.h
void
*
vnodeBufPoolMalloc
(
SVBufPool
*
pPool
,
int
size
);
...
...
@@ -88,7 +91,7 @@ int32_t metaCreateTSma(SMeta* pMeta, SSmaCfg* pCfg);
int32_t
metaDropTSma
(
SMeta
*
pMeta
,
int64_t
indexUid
);
// tsdb
int
tsdbOpen
(
SVnode
*
pVnode
,
STsdb
**
ppTsdb
);
int
tsdbOpen
(
SVnode
*
pVnode
,
int8_t
type
);
int
tsdbClose
(
STsdb
*
pTsdb
);
int
tsdbBegin
(
STsdb
*
pTsdb
);
int
tsdbCommit
(
STsdb
*
pTsdb
);
...
...
@@ -161,6 +164,8 @@ struct SVnode {
SVBufPool
*
onRecycle
;
SMeta
*
pMeta
;
STsdb
*
pTsdb
;
STsdb
*
pRSma1
;
STsdb
*
pRSma2
;
SWal
*
pWal
;
STQ
*
pTq
;
SSink
*
pSink
;
...
...
@@ -169,6 +174,11 @@ struct SVnode {
SQHandle
*
pQuery
;
};
#define VND_TSDB(vnd) ((vnd)->pTsdb)
#define VND_RSMA0(vnd) ((vnd)->pTsdb)
#define VND_RSMA1(vnd) ((vnd)->pRSma1)
#define VND_RSMA2(vnd) ((vnd)->pRSma2)
struct
STbUidStore
{
tb_uid_t
suid
;
SArray
*
tbUids
;
...
...
@@ -177,7 +187,11 @@ struct STbUidStore {
#define TD_VID(PVNODE) (PVNODE)->config.vgId
// typedef struct STbDdlH STbDdlH;
static
FORCE_INLINE
bool
tsdbIsRollup
(
SVnode
*
pVnode
)
{
SRetention
*
pRetention
=
&
(
pVnode
->
config
.
tsdbCfg
.
retentions
[
0
]);
return
(
pRetention
->
freq
>
0
&&
pRetention
->
keep
>
0
);
}
// sma
void
smaHandleRes
(
void
*
pVnode
,
int64_t
smaId
,
const
SArray
*
data
);
...
...
source/dnode/vnode/src/tsdb/tsdbOpen.c
浏览文件 @
89495a32
...
...
@@ -15,12 +15,34 @@
#include "tsdb.h"
int
tsdbOpen
(
SVnode
*
pVnode
,
STsdb
**
ppTsdb
)
{
static
int
tsdbOpenImpl
(
SVnode
*
pVnode
,
int8_t
type
,
STsdb
**
ppTsdb
,
const
char
*
dir
);
int
tsdbOpen
(
SVnode
*
pVnode
,
int8_t
type
)
{
switch
(
type
)
{
case
TSDB_TYPE_TSDB
:
return
tsdbOpenImpl
(
pVnode
,
type
,
&
VND_TSDB
(
pVnode
),
VNODE_TSDB_DIR
);
case
TSDB_TYPE_TSMA
:
ASSERT
(
0
);
break
;
case
TSDB_TYPE_RSMA_L0
:
return
tsdbOpenImpl
(
pVnode
,
type
,
&
VND_RSMA0
(
pVnode
),
VNODE_TSDB_DIR
);
case
TSDB_TYPE_RSMA_L1
:
return
tsdbOpenImpl
(
pVnode
,
type
,
&
VND_RSMA1
(
pVnode
),
VNODE_RSMA1_DIR
);
case
TSDB_TYPE_RSMA_L2
:
return
tsdbOpenImpl
(
pVnode
,
type
,
&
VND_RSMA2
(
pVnode
),
VNODE_RSMA2_DIR
);
default:
ASSERT
(
0
);
break
;
}
return
0
;
}
int
tsdbOpenImpl
(
SVnode
*
pVnode
,
int8_t
type
,
STsdb
**
ppTsdb
,
const
char
*
dir
)
{
STsdb
*
pTsdb
=
NULL
;
int
slen
=
0
;
*
ppTsdb
=
NULL
;
slen
=
strlen
(
tfsGetPrimaryPath
(
pVnode
->
pTfs
))
+
strlen
(
pVnode
->
path
)
+
strlen
(
VNODE_TSDB_DIR
)
+
3
;
slen
=
strlen
(
tfsGetPrimaryPath
(
pVnode
->
pTfs
))
+
strlen
(
pVnode
->
path
)
+
strlen
(
dir
)
+
3
;
// create handle
pTsdb
=
(
STsdb
*
)
taosMemoryCalloc
(
1
,
sizeof
(
*
pTsdb
)
+
slen
);
...
...
@@ -31,7 +53,7 @@ int tsdbOpen(SVnode *pVnode, STsdb **ppTsdb) {
pTsdb
->
path
=
(
char
*
)
&
pTsdb
[
1
];
sprintf
(
pTsdb
->
path
,
"%s%s%s%s%s"
,
tfsGetPrimaryPath
(
pVnode
->
pTfs
),
TD_DIRSEP
,
pVnode
->
path
,
TD_DIRSEP
,
VNODE_TSDB_DIR
);
dir
);
pTsdb
->
pVnode
=
pVnode
;
pTsdb
->
repoLocked
=
false
;
taosThreadMutexInit
(
&
pTsdb
->
mutex
,
NULL
);
...
...
@@ -45,7 +67,7 @@ int tsdbOpen(SVnode *pVnode, STsdb **ppTsdb) {
goto
_err
;
}
tsdbDebug
(
"vgId:
%d tsdb is opened"
,
TD_VID
(
pVnode
)
);
tsdbDebug
(
"vgId:
%d tsdb is opened for %s"
,
TD_VID
(
pVnode
),
pTsdb
->
path
);
*
ppTsdb
=
pTsdb
;
return
0
;
...
...
source/dnode/vnode/src/tsdb/tsdbSma.c
浏览文件 @
89495a32
...
...
@@ -103,10 +103,9 @@ typedef struct {
STSma
*
pSma
;
// cache schema
}
SSmaStatItem
;
#define RSMA_MAX_LEVEL 2
#define RSMA_TASK_INFO_HASH_SLOT 8
struct
SRSmaInfo
{
void
*
taskInfo
[
RSMA_MAX_LEVEL
];
// qTaskInfo_t
void
*
taskInfo
[
TSDB_RSMA_RETENTION_2
];
// qTaskInfo_t
};
struct
SSmaStat
{
...
...
@@ -128,7 +127,7 @@ static FORCE_INLINE void tsdbFreeTaskHandle(qTaskInfo_t *taskHandle) {
}
static
FORCE_INLINE
void
*
tsdbFreeRSmaInfo
(
SRSmaInfo
*
pInfo
)
{
for
(
int32_t
i
=
0
;
i
<
RSMA_MAX_LEVEL
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
TSDB_RSMA_RETENTION_MAX
;
++
i
)
{
if
(
pInfo
->
taskInfo
[
i
])
{
tsdbFreeTaskHandle
(
pInfo
->
taskInfo
[
i
]);
}
...
...
source/dnode/vnode/src/vnd/vnodeCfg.c
浏览文件 @
89495a32
...
...
@@ -66,6 +66,28 @@ int vnodeEncodeConfig(const void *pObj, SJson *pJson) {
if
(
tjsonAddIntegerToObject
(
pJson
,
"keep0"
,
pCfg
->
tsdbCfg
.
keep0
)
<
0
)
return
-
1
;
if
(
tjsonAddIntegerToObject
(
pJson
,
"keep1"
,
pCfg
->
tsdbCfg
.
keep1
)
<
0
)
return
-
1
;
if
(
tjsonAddIntegerToObject
(
pJson
,
"keep2"
,
pCfg
->
tsdbCfg
.
keep2
)
<
0
)
return
-
1
;
#ifdef TSDB_VNODE_SMA_DEBUG
if
(
pCfg
->
tsdbCfg
.
retentions
[
0
].
freq
>
0
)
{
int32_t
nRetention
=
1
;
if
(
pCfg
->
tsdbCfg
.
retentions
[
1
].
freq
>
0
)
{
++
nRetention
;
if
(
pCfg
->
tsdbCfg
.
retentions
[
2
].
freq
>
0
)
{
++
nRetention
;
}
}
SJson
*
pNodeRetentions
=
tjsonCreateArray
();
tjsonAddItemToObject
(
pJson
,
"retentions"
,
pNodeRetentions
);
for
(
int32_t
i
=
0
;
i
<
nRetention
;
++
i
)
{
SJson
*
pNodeRetention
=
tjsonCreateObject
();
const
SRetention
*
pRetention
=
pCfg
->
tsdbCfg
.
retentions
+
i
;
tjsonAddIntegerToObject
(
pNodeRetention
,
"freq"
,
pRetention
->
freq
);
tjsonAddIntegerToObject
(
pNodeRetention
,
"freqUnit"
,
pRetention
->
freqUnit
);
tjsonAddIntegerToObject
(
pNodeRetention
,
"keep"
,
pRetention
->
keep
);
tjsonAddIntegerToObject
(
pNodeRetention
,
"keepUnit"
,
pRetention
->
keepUnit
);
tjsonAddItemToArray
(
pNodeRetentions
,
pNodeRetention
);
}
}
#endif
if
(
tjsonAddIntegerToObject
(
pJson
,
"wal.vgId"
,
pCfg
->
walCfg
.
vgId
)
<
0
)
return
-
1
;
if
(
tjsonAddIntegerToObject
(
pJson
,
"wal.fsyncPeriod"
,
pCfg
->
walCfg
.
fsyncPeriod
)
<
0
)
return
-
1
;
if
(
tjsonAddIntegerToObject
(
pJson
,
"wal.retentionPeriod"
,
pCfg
->
walCfg
.
retentionPeriod
)
<
0
)
return
-
1
;
...
...
@@ -113,6 +135,20 @@ int vnodeDecodeConfig(const SJson *pJson, void *pObj) {
if
(
tjsonGetNumberValue
(
pJson
,
"keep0"
,
pCfg
->
tsdbCfg
.
keep0
)
<
0
)
return
-
1
;
if
(
tjsonGetNumberValue
(
pJson
,
"keep1"
,
pCfg
->
tsdbCfg
.
keep1
)
<
0
)
return
-
1
;
if
(
tjsonGetNumberValue
(
pJson
,
"keep2"
,
pCfg
->
tsdbCfg
.
keep2
)
<
0
)
return
-
1
;
#ifdef TSDB_VNODE_SMA_DEBUG
SJson
*
pNodeRetentions
=
tjsonGetObjectItem
(
pJson
,
"retentions"
);
int
nRetention
=
tjsonGetArraySize
(
pNodeRetentions
);
ASSERT
(
nRetention
<=
TSDB_RSMA_RETENTION_MAX
);
for
(
int32_t
i
=
0
;
i
<
nRetention
;
++
i
)
{
SJson
*
pNodeRetention
=
tjsonGetArrayItem
(
pNodeRetentions
,
i
);
ASSERT
(
pNodeRetention
!=
NULL
);
tjsonGetNumberValue
(
pNodeRetention
,
"freq"
,
(
pCfg
->
tsdbCfg
.
retentions
)[
i
].
freq
);
tjsonGetNumberValue
(
pNodeRetention
,
"freqUnit"
,
(
pCfg
->
tsdbCfg
.
retentions
)[
i
].
freqUnit
);
tjsonGetNumberValue
(
pNodeRetention
,
"keep"
,
(
pCfg
->
tsdbCfg
.
retentions
)[
i
].
keep
);
tjsonGetNumberValue
(
pNodeRetention
,
"keepUnit"
,
(
pCfg
->
tsdbCfg
.
retentions
)[
i
].
keepUnit
);
}
#endif
if
(
tjsonGetNumberValue
(
pJson
,
"wal.vgId"
,
pCfg
->
walCfg
.
vgId
)
<
0
)
return
-
1
;
if
(
tjsonGetNumberValue
(
pJson
,
"wal.fsyncPeriod"
,
pCfg
->
walCfg
.
fsyncPeriod
)
<
0
)
return
-
1
;
if
(
tjsonGetNumberValue
(
pJson
,
"wal.retentionPeriod"
,
pCfg
->
walCfg
.
retentionPeriod
)
<
0
)
return
-
1
;
...
...
source/dnode/vnode/src/vnd/vnodeOpen.c
浏览文件 @
89495a32
...
...
@@ -96,9 +96,26 @@ SVnode *vnodeOpen(const char *path, STfs *pTfs, SMsgCb msgCb) {
}
// open tsdb
if
(
tsdbOpen
(
pVnode
,
&
pVnode
->
pTsdb
)
<
0
)
{
vError
(
"vgId:%d failed to open vnode tsdb since %s"
,
TD_VID
(
pVnode
),
tstrerror
(
terrno
));
goto
_err
;
if
(
tsdbIsRollup
(
pVnode
))
{
if
(
tsdbOpen
(
pVnode
,
TSDB_TYPE_RSMA_L0
)
<
0
)
{
vError
(
"vgId: %d failed to open vnode rsma0 since %s"
,
TD_VID
(
pVnode
),
tstrerror
(
terrno
));
goto
_err
;
}
if
(
tsdbOpen
(
pVnode
,
TSDB_TYPE_RSMA_L1
)
<
0
)
{
vError
(
"vgId: %d failed to open vnode rsma1 since %s"
,
TD_VID
(
pVnode
),
tstrerror
(
terrno
));
goto
_err
;
}
if
(
tsdbOpen
(
pVnode
,
TSDB_TYPE_RSMA_L2
)
<
0
)
{
vError
(
"vgId: %d failed to open vnode rsma2 since %s"
,
TD_VID
(
pVnode
),
tstrerror
(
terrno
));
goto
_err
;
}
}
else
{
if
(
tsdbOpen
(
pVnode
,
TSDB_TYPE_TSDB
)
<
0
)
{
vError
(
"vgId: %d failed to open vnode tsdb since %s"
,
TD_VID
(
pVnode
),
tstrerror
(
terrno
));
goto
_err
;
}
}
// open wal
...
...
@@ -143,6 +160,8 @@ _err:
if
(
pVnode
->
pWal
)
walClose
(
pVnode
->
pWal
);
if
(
pVnode
->
pTsdb
)
tsdbClose
(
pVnode
->
pTsdb
);
if
(
pVnode
->
pMeta
)
metaClose
(
pVnode
->
pMeta
);
tsdbClose
(
VND_RSMA1
(
pVnode
));
tsdbClose
(
VND_RSMA2
(
pVnode
));
tsem_destroy
(
&
(
pVnode
->
canCommit
));
taosMemoryFree
(
pVnode
);
return
NULL
;
...
...
@@ -155,7 +174,9 @@ void vnodeClose(SVnode *pVnode) {
vnodeQueryClose
(
pVnode
);
walClose
(
pVnode
->
pWal
);
tqClose
(
pVnode
->
pTq
);
tsdbClose
(
pVnode
->
pTsdb
);
tsdbClose
(
VND_TSDB
(
pVnode
));
tsdbClose
(
VND_RSMA1
(
pVnode
));
tsdbClose
(
VND_RSMA2
(
pVnode
));
metaClose
(
pVnode
->
pMeta
);
vnodeCloseBufPool
(
pVnode
);
// destroy handle
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录