Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
慢慢CG
TDengine
提交
291aaff1
T
TDengine
项目概览
慢慢CG
/
TDengine
与 Fork 源项目一致
Fork自
taosdata / TDengine
通知
1
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
提交
291aaff1
编写于
2月 01, 2021
作者:
H
Hongze Cheng
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
fix two review bugs
上级
89d9be4d
变更
1
隐藏空白更改
内联
并排
Showing
1 changed file
with
37 addition
and
8 deletion
+37
-8
src/tsdb/src/tsdbCommit.c
src/tsdb/src/tsdbCommit.c
+37
-8
未找到文件。
src/tsdb/src/tsdbCommit.c
浏览文件 @
291aaff1
...
...
@@ -312,11 +312,11 @@ static int tsdbCommitTSData(STsdbRepo *pRepo) {
// Skip expired memory data and expired FSET
tsdbSeekCommitIter
(
&
commith
,
commith
.
rtn
.
minKey
);
while
((
pSet
=
tsdbFSIterNext
(
&
(
commith
.
fsIter
))))
{
if
(
pSet
->
fid
>=
commith
.
rtn
.
minFid
)
{
break
;
}
else
{
if
(
pSet
->
fid
<
commith
.
rtn
.
minFid
)
{
tsdbInfo
(
"vgId:%d FSET %d on level %d disk id %d expires, remove it"
,
REPO_ID
(
pRepo
),
pSet
->
fid
,
TSDB_FSET_LEVEL
(
pSet
),
TSDB_FSET_ID
(
pSet
));
}
else
{
break
;
}
}
...
...
@@ -427,14 +427,18 @@ static int tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid) {
if
(
pIter
->
pTable
==
NULL
)
continue
;
if
(
tsdbCommitToTable
(
pCommith
,
tid
)
<
0
)
{
// TODO: revert the file change
tsdbCloseCommitFile
(
pCommith
,
true
);
// revert the file change
tsdbApplyDFileSetChange
(
TSDB_COMMIT_WRITE_FSET
(
pCommith
),
pSet
);
return
-
1
;
}
}
if
(
tsdbWriteBlockIdx
(
pCommith
)
<
0
)
{
tsdbError
(
"vgId:%d failed to write SBlockIdx part to FSET %d since %s"
,
REPO_ID
(
pRepo
),
fid
,
tstrerror
(
terrno
));
tsdbCloseCommitFile
(
pCommith
,
true
);
// revert the file change
tsdbApplyDFileSetChange
(
TSDB_COMMIT_WRITE_FSET
(
pCommith
),
pSet
);
return
-
1
;
}
...
...
@@ -688,7 +692,11 @@ static int tsdbCommitToTable(SCommitH *pCommith, int tid) {
TSDB_RUNLOCK_TABLE
(
pIter
->
pTable
);
if
(
tsdbWriteBlockInfo
(
pCommith
)
<
0
)
return
-
1
;
if
(
tsdbWriteBlockInfo
(
pCommith
)
<
0
)
{
tsdbError
(
"vgId:%d failed to write SBlockInfo part into file %s since %s"
,
REPO_ID
(
pRepo
),
TSDB_FILE_FULL_NAME
(
TSDB_COMMIT_HEAD_FILE
(
pCommith
)),
tstrerror
(
terrno
));
return
-
1
;
}
return
0
;
}
...
...
@@ -940,6 +948,8 @@ static int tsdbWriteBlockIdx(SCommitH *pCommih) {
if
(
nidx
<=
0
)
{
// All data are deleted
pHeadf
->
info
.
offset
=
0
;
pHeadf
->
info
.
len
=
0
;
return
0
;
}
...
...
@@ -1241,7 +1251,6 @@ static void tsdbResetCommitFile(SCommitH *pCommith) {
}
static
void
tsdbResetCommitTable
(
SCommitH
*
pCommith
)
{
tdResetDataCols
(
pCommith
->
pDataCols
);
taosArrayClear
(
pCommith
->
aSubBlk
);
taosArrayClear
(
pCommith
->
aSupBlk
);
pCommith
->
pTable
=
NULL
;
...
...
@@ -1270,6 +1279,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
tsdbCloseAndUnsetFSet
(
&
(
pCommith
->
readh
));
return
-
1
;
}
tsdbDebug
(
"vgId:%d FSET %d at level %d disk id %d is opened to read to commit"
,
REPO_ID
(
pRepo
),
TSDB_FSET_FID
(
pSet
),
TSDB_FSET_LEVEL
(
pSet
),
TSDB_FSET_ID
(
pSet
));
}
else
{
pCommith
->
isRFileSet
=
false
;
}
...
...
@@ -1280,6 +1292,8 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
tsdbInitDFileSet
(
pWSet
,
did
,
REPO_ID
(
pRepo
),
fid
,
FS_TXN_VERSION
(
REPO_FS
(
pRepo
)));
if
(
tsdbCreateDFileSet
(
pWSet
,
true
)
<
0
)
{
tsdbError
(
"vgId:%d failed to create FSET %d at level %d disk id %d since %s"
,
REPO_ID
(
pRepo
),
TSDB_FSET_FID
(
pWSet
),
TSDB_FSET_LEVEL
(
pWSet
),
TSDB_FSET_ID
(
pWSet
),
tstrerror
(
terrno
));
if
(
pCommith
->
isRFileSet
)
{
tsdbCloseAndUnsetFSet
(
&
(
pCommith
->
readh
));
}
...
...
@@ -1288,6 +1302,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
pCommith
->
isDFileSame
=
false
;
pCommith
->
isLFileSame
=
false
;
tsdbDebug
(
"vgId:%d FSET %d at level %d disk id %d is created to commit"
,
REPO_ID
(
pRepo
),
TSDB_FSET_FID
(
pWSet
),
TSDB_FSET_LEVEL
(
pWSet
),
TSDB_FSET_ID
(
pWSet
));
}
else
{
did
.
level
=
TSDB_FSET_LEVEL
(
pSet
);
did
.
id
=
TSDB_FSET_ID
(
pSet
);
...
...
@@ -1299,6 +1316,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
SDFile
*
pWHeadf
=
TSDB_COMMIT_HEAD_FILE
(
pCommith
);
tsdbInitDFile
(
pWHeadf
,
did
,
REPO_ID
(
pRepo
),
fid
,
FS_TXN_VERSION
(
REPO_FS
(
pRepo
)),
TSDB_FILE_HEAD
);
if
(
tsdbCreateDFile
(
pWHeadf
,
true
)
<
0
)
{
tsdbError
(
"vgId:%d failed to create file %s to commit since %s"
,
REPO_ID
(
pRepo
),
TSDB_FILE_FULL_NAME
(
pWHeadf
),
tstrerror
(
terrno
));
if
(
pCommith
->
isRFileSet
)
{
tsdbCloseAndUnsetFSet
(
&
(
pCommith
->
readh
));
return
-
1
;
...
...
@@ -1310,7 +1330,10 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
SDFile
*
pWDataf
=
TSDB_COMMIT_DATA_FILE
(
pCommith
);
tsdbInitDFileEx
(
pWDataf
,
pRDataf
);
if
(
tsdbOpenDFile
(
pWDataf
,
O_WRONLY
)
<
0
)
{
tsdbCloseDFile
(
pWHeadf
);
tsdbError
(
"vgId:%d failed to open file %s to commit since %s"
,
REPO_ID
(
pRepo
),
TSDB_FILE_FULL_NAME
(
pWDataf
),
tstrerror
(
terrno
));
tsdbCloseDFileSet
(
pWSet
);
tsdbRemoveDFile
(
pWHeadf
);
if
(
pCommith
->
isRFileSet
)
{
tsdbCloseAndUnsetFSet
(
&
(
pCommith
->
readh
));
...
...
@@ -1327,6 +1350,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
pCommith
->
isLFileSame
=
true
;
if
(
tsdbOpenDFile
(
pWLastf
,
O_WRONLY
)
<
0
)
{
tsdbError
(
"vgId:%d failed to open file %s to commit since %s"
,
REPO_ID
(
pRepo
),
TSDB_FILE_FULL_NAME
(
pWLastf
),
tstrerror
(
terrno
));
tsdbCloseDFileSet
(
pWSet
);
tsdbRemoveDFile
(
pWHeadf
);
if
(
pCommith
->
isRFileSet
)
{
...
...
@@ -1339,6 +1365,9 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
pCommith
->
isLFileSame
=
false
;
if
(
tsdbCreateDFile
(
pWLastf
,
true
)
<
0
)
{
tsdbError
(
"vgId:%d failed to create file %s to commit since %s"
,
REPO_ID
(
pRepo
),
TSDB_FILE_FULL_NAME
(
pWLastf
),
tstrerror
(
terrno
));
tsdbCloseDFileSet
(
pWSet
);
tsdbRemoveDFile
(
pWHeadf
);
if
(
pCommith
->
isRFileSet
)
{
...
...
@@ -1374,7 +1403,7 @@ static bool tsdbCanAddSubBlock(SCommitH *pCommith, SBlock *pBlock, SMergeInfo *p
if
(
pBlock
->
last
)
{
if
(
pCommith
->
isLFileSame
&&
mergeRows
<
pCfg
->
minRowsPerFileBlock
)
return
true
;
}
else
{
if
(
mergeRows
<
pCfg
->
maxRowsPerFileBlock
)
return
true
;
if
(
pCommith
->
isDFileSame
&&
mergeRows
<=
pCfg
->
maxRowsPerFileBlock
)
return
true
;
}
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录