Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
d35ca7c2
T
TDengine
项目概览
taosdata
/
TDengine
大约 2 年 前同步成功
通知
1192
Star
22018
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看板
提交
d35ca7c2
编写于
11月 18, 2021
作者:
C
Cary Xu
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
code optimization
上级
44ad2279
变更
3
隐藏空白更改
内联
并排
Showing
3 changed file
with
59 addition
and
49 deletion
+59
-49
src/tsdb/src/tsdbMain.c
src/tsdb/src/tsdbMain.c
+1
-0
src/tsdb/src/tsdbReadImpl.c
src/tsdb/src/tsdbReadImpl.c
+0
-1
src/tsdb/src/tsdbTruncate.c
src/tsdb/src/tsdbTruncate.c
+58
-48
未找到文件。
src/tsdb/src/tsdbMain.c
浏览文件 @
d35ca7c2
...
...
@@ -557,6 +557,7 @@ static STsdbRepo *tsdbNewRepo(STsdbCfg *pCfg, STsdbAppH *pAppH) {
pRepo
->
state
=
TSDB_STATE_OK
;
pRepo
->
code
=
TSDB_CODE_SUCCESS
;
pRepo
->
compactState
=
0
;
pRepo
->
truncateState
=
0
;
pRepo
->
config
=
*
pCfg
;
if
(
pAppH
)
{
pRepo
->
appH
=
*
pAppH
;
...
...
src/tsdb/src/tsdbReadImpl.c
浏览文件 @
d35ca7c2
...
...
@@ -682,7 +682,6 @@ static int tsdbLoadBlockDataImpl(SReadH *pReadh, SBlock *pBlock, SDataCols *pDat
uint32_t
toffset
=
TSDB_KEY_COL_OFFSET
;
int32_t
tlen
=
pBlock
->
keyLen
;
if
(
dcol
!=
0
)
{
tsdbGetSBlockCol
(
pBlock
,
&
pBlockCol
,
pBlockData
->
cols
,
ccol
);
tcolId
=
pBlockCol
->
colId
;
...
...
src/tsdb/src/tsdbTruncate.c
浏览文件 @
d35ca7c2
...
...
@@ -24,7 +24,7 @@ typedef struct {
typedef
struct
{
SRtn
rtn
;
SFSIter
fsIter
;
SArray
*
tbArray
;
// STableTruncateH, table array to cache table obj and block indexes
SArray
*
tb
l
Array
;
// STableTruncateH, table array to cache table obj and block indexes
SReadH
readh
;
SDFileSet
wSet
;
SArray
*
aBlkIdx
;
...
...
@@ -51,8 +51,8 @@ static int tsdbTruncateTSData(STsdbRepo *pRepo, void *param);
static
int
tsdbTruncateFSet
(
STruncateH
*
pTruncateH
,
SDFileSet
*
pSet
);
static
int
tsdbInitTruncateH
(
STruncateH
*
pTruncateH
,
STsdbRepo
*
pRepo
);
static
void
tsdbDestroyTruncateH
(
STruncateH
*
pTruncateH
);
static
int
tsdbInit
CompTb
Array
(
STruncateH
*
pTruncateH
);
static
void
tsdbDestroy
CompTb
Array
(
STruncateH
*
pTruncateH
);
static
int
tsdbInit
TruncateTbl
Array
(
STruncateH
*
pTruncateH
);
static
void
tsdbDestroy
TruncateTbl
Array
(
STruncateH
*
pTruncateH
);
static
int
tsdbCacheFSetIndex
(
STruncateH
*
pTruncateH
);
static
int
tsdbTruncateCache
(
STsdbRepo
*
pRepo
,
void
*
param
);
static
int
tsdbTruncateFSetInit
(
STruncateH
*
pTruncateH
,
SDFileSet
*
pSet
);
...
...
@@ -250,6 +250,7 @@ static int tsdbTruncateFSet(STruncateH *pTruncateH, SDFileSet *pSet) {
tsdbInitDFileSet
(
TSDB_TRUNCATE_WSET
(
pTruncateH
),
did
,
REPO_ID
(
pRepo
),
TSDB_FSET_FID
(
pSet
),
FS_TXN_VERSION
(
REPO_FS
(
pRepo
)),
TSDB_LATEST_FSET_VER
);
if
(
tsdbCreateDFileSet
(
TSDB_TRUNCATE_WSET
(
pTruncateH
),
true
)
<
0
)
{
tsdbError
(
"vgId:%d failed to truncate FSET %d since %s"
,
REPO_ID
(
pRepo
),
pSet
->
fid
,
tstrerror
(
terrno
));
tsdbTruncateFSetEnd
(
pTruncateH
);
...
...
@@ -285,7 +286,7 @@ static int tsdbInitTruncateH(STruncateH *pTruncateH, STsdbRepo *pRepo) {
return
-
1
;
}
if
(
tsdbInit
CompTb
Array
(
pTruncateH
)
<
0
)
{
if
(
tsdbInit
TruncateTbl
Array
(
pTruncateH
)
<
0
)
{
tsdbDestroyTruncateH
(
pTruncateH
);
return
-
1
;
}
...
...
@@ -318,33 +319,33 @@ static void tsdbDestroyTruncateH(STruncateH *pTruncateH) {
pTruncateH
->
pDataCols
=
tdFreeDataCols
(
pTruncateH
->
pDataCols
);
pTruncateH
->
aSupBlk
=
taosArrayDestroy
(
pTruncateH
->
aSupBlk
);
pTruncateH
->
aBlkIdx
=
taosArrayDestroy
(
pTruncateH
->
aBlkIdx
);
tsdbDestroy
CompTb
Array
(
pTruncateH
);
tsdbDestroy
TruncateTbl
Array
(
pTruncateH
);
tsdbDestroyReadH
(
&
(
pTruncateH
->
readh
));
tsdbCloseDFileSet
(
TSDB_TRUNCATE_WSET
(
pTruncateH
));
}
static
int
tsdbInit
CompTbArray
(
STruncateH
*
pTruncateH
)
{
// Init pComp->tbArray
static
int
tsdbInit
TruncateTblArray
(
STruncateH
*
pTruncateH
)
{
STsdbRepo
*
pRepo
=
TSDB_TRUNCATE_REPO
(
pTruncateH
);
STsdbMeta
*
pMeta
=
pRepo
->
tsdbMeta
;
if
(
tsdbRLockRepoMeta
(
pRepo
)
<
0
)
return
-
1
;
pTruncateH
->
tbArray
=
taosArrayInit
(
pMeta
->
maxTables
,
sizeof
(
STableTruncateH
));
if
(
pTruncateH
->
tbArray
==
NULL
)
{
pTruncateH
->
tb
l
Array
=
taosArrayInit
(
pMeta
->
maxTables
,
sizeof
(
STableTruncateH
));
if
(
pTruncateH
->
tb
l
Array
==
NULL
)
{
terrno
=
TSDB_CODE_TDB_OUT_OF_MEMORY
;
tsdbUnlockRepoMeta
(
pRepo
);
return
-
1
;
}
// Note here must start from 0
for
(
int
i
=
0
;
i
<
pMeta
->
maxTables
;
i
++
)
{
for
(
int
i
=
0
;
i
<
pMeta
->
maxTables
;
++
i
)
{
STableTruncateH
ch
=
{
0
};
if
(
pMeta
->
tables
[
i
]
!=
NULL
)
{
tsdbRefTable
(
pMeta
->
tables
[
i
]);
ch
.
pTable
=
pMeta
->
tables
[
i
];
}
if
(
taosArrayPush
(
pTruncateH
->
tbArray
,
&
ch
)
==
NULL
)
{
if
(
taosArrayPush
(
pTruncateH
->
tb
l
Array
,
&
ch
)
==
NULL
)
{
terrno
=
TSDB_CODE_TDB_OUT_OF_MEMORY
;
tsdbUnlockRepoMeta
(
pRepo
);
return
-
1
;
...
...
@@ -355,22 +356,21 @@ static int tsdbInitCompTbArray(STruncateH *pTruncateH) { // Init pComp->tbArray
return
0
;
}
static
void
tsdbDestroy
CompTb
Array
(
STruncateH
*
pTruncateH
)
{
STableTruncateH
*
pT
h
;
static
void
tsdbDestroy
TruncateTbl
Array
(
STruncateH
*
pTruncateH
)
{
STableTruncateH
*
pT
blHandle
=
NULL
;
if
(
pTruncateH
->
tbArray
==
NULL
)
return
;
if
(
pTruncateH
->
tb
l
Array
==
NULL
)
return
;
for
(
size_t
i
=
0
;
i
<
taosArrayGetSize
(
pTruncateH
->
tb
Array
);
i
++
)
{
pT
h
=
(
STableTruncateH
*
)
taosArrayGet
(
pTruncateH
->
tb
Array
,
i
);
if
(
pT
h
->
pTable
)
{
tsdbUnRefTable
(
pT
h
->
pTable
);
for
(
size_t
i
=
0
;
i
<
taosArrayGetSize
(
pTruncateH
->
tb
lArray
);
++
i
)
{
pT
blHandle
=
(
STableTruncateH
*
)
taosArrayGet
(
pTruncateH
->
tbl
Array
,
i
);
if
(
pT
blHandle
->
pTable
)
{
tsdbUnRefTable
(
pT
blHandle
->
pTable
);
}
// pTh->pInfo = taosTZfree(pTh->pInfo);
tfree
(
pTh
->
pInfo
);
tfree
(
pTblHandle
->
pInfo
);
}
pTruncateH
->
tb
Array
=
taosArrayDestroy
(
pTruncateH
->
tb
Array
);
pTruncateH
->
tb
lArray
=
taosArrayDestroy
(
pTruncateH
->
tbl
Array
);
}
static
int
tsdbCacheFSetIndex
(
STruncateH
*
pTruncateH
)
{
...
...
@@ -380,21 +380,22 @@ static int tsdbCacheFSetIndex(STruncateH *pTruncateH) {
return
-
1
;
}
for
(
int
tid
=
1
;
tid
<
taosArrayGetSize
(
pTruncateH
->
tbArray
);
tid
++
)
{
STableTruncateH
*
pTh
=
(
STableTruncateH
*
)
taosArrayGet
(
pTruncateH
->
tbArray
,
tid
);
pTh
->
pBlkIdx
=
NULL
;
size_t
tblArraySize
=
taosArrayGetSize
(
pTruncateH
->
tblArray
);
for
(
size_t
tid
=
1
;
tid
<
tblArraySize
;
++
tid
)
{
STableTruncateH
*
pTblHandle
=
(
STableTruncateH
*
)
taosArrayGet
(
pTruncateH
->
tblArray
,
tid
);
pTblHandle
->
pBlkIdx
=
NULL
;
if
(
pT
h
->
pTable
==
NULL
)
continue
;
if
(
tsdbSetReadTable
(
pReadH
,
pT
h
->
pTable
)
<
0
)
{
if
(
pT
blHandle
->
pTable
==
NULL
)
continue
;
if
(
tsdbSetReadTable
(
pReadH
,
pT
blHandle
->
pTable
)
<
0
)
{
return
-
1
;
}
if
(
pReadH
->
pBlkIdx
==
NULL
)
continue
;
pT
h
->
bIndex
=
*
(
pReadH
->
pBlkIdx
);
pT
h
->
pBlkIdx
=
&
(
pTh
->
bIndex
);
pT
blHandle
->
bIndex
=
*
(
pReadH
->
pBlkIdx
);
pT
blHandle
->
pBlkIdx
=
&
(
pTblHandle
->
bIndex
);
uint32_t
originLen
=
0
;
if
(
tsdbLoadBlockInfo
(
pReadH
,
(
void
**
)(
&
(
pT
h
->
pInfo
)),
&
originLen
)
<
0
)
{
if
(
tsdbLoadBlockInfo
(
pReadH
,
(
void
**
)(
&
(
pT
blHandle
->
pInfo
)),
&
originLen
)
<
0
)
{
return
-
1
;
}
}
...
...
@@ -424,7 +425,7 @@ static int tsdbTruncateFSetImpl(STruncateH *pTruncateH) {
STsdbRepo
*
pRepo
=
TSDB_TRUNCATE_REPO
(
pTruncateH
);
STsdbCfg
*
pCfg
=
REPO_CFG
(
pRepo
);
SReadH
*
pReadh
=
&
(
pTruncateH
->
readh
);
SBlockIdx
blkIdx
;
SBlockIdx
blkIdx
=
{
0
}
;
void
**
ppBuf
=
&
(
TSDB_TRUNCATE_BUF
(
pTruncateH
));
void
**
ppCBuf
=
&
(
TSDB_TRUNCATE_COMP_BUF
(
pTruncateH
));
void
**
ppExBuf
=
&
(
TSDB_TRUNCATE_EXBUF
(
pTruncateH
));
...
...
@@ -432,51 +433,60 @@ static int tsdbTruncateFSetImpl(STruncateH *pTruncateH) {
taosArrayClear
(
pTruncateH
->
aBlkIdx
);
for
(
int
tid
=
1
;
tid
<
taosArrayGetSize
(
pTruncateH
->
tbArray
);
tid
++
)
{
STableTruncateH
*
pTh
=
(
STableTruncateH
*
)
taosArrayGet
(
pTruncateH
->
tbArray
,
tid
);
STSchema
*
pSchema
;
for
(
size_t
tid
=
1
;
tid
<
taosArrayGetSize
(
pTruncateH
->
tblArray
);
++
tid
)
{
STableTruncateH
*
pTblHandle
=
(
STableTruncateH
*
)
taosArrayGet
(
pTruncateH
->
tblArray
,
tid
);
STSchema
*
pSchema
=
NULL
;
if
(
pTblHandle
->
pTable
==
NULL
||
pTblHandle
->
pBlkIdx
==
NULL
)
continue
;
if
(
pTh
->
pTable
==
NULL
||
pTh
->
pBlkIdx
==
NULL
)
continue
;
if
((
pSchema
=
tsdbGetTableSchemaImpl
(
pTblHandle
->
pTable
,
true
,
true
,
-
1
,
-
1
))
==
NULL
)
{
return
-
1
;
}
pSchema
=
tsdbGetTableSchemaImpl
(
pTh
->
pTable
,
true
,
true
,
-
1
,
-
1
);
taosArrayClear
(
pTruncateH
->
aSupBlk
);
if
((
tdInitDataCols
(
pTruncateH
->
pDataCols
,
pSchema
)
<
0
)
||
(
tdInitDataCols
(
pReadh
->
pDCols
[
0
],
pSchema
)
<
0
)
||
(
tdInitDataCols
(
pReadh
->
pDCols
[
1
],
pSchema
)
<
0
))
{
terrno
=
TSDB_CODE_TDB_OUT_OF_MEMORY
;
tdFreeSchema
(
pSchema
);
return
-
1
;
}
tdFreeSchema
(
pSchema
);
// Loop to truncate each block data
for
(
int
i
=
0
;
i
<
pT
h
->
pBlkIdx
->
numOfBlocks
;
i
++
)
{
SBlock
*
pBlock
=
pT
h
->
pInfo
->
blocks
+
i
;
for
(
int
i
=
0
;
i
<
pT
blHandle
->
pBlkIdx
->
numOfBlocks
;
++
i
)
{
SBlock
*
pBlock
=
pT
blHandle
->
pInfo
->
blocks
+
i
;
// Load the block data
if
(
tsdbLoadBlockData
(
pReadh
,
pBlock
,
pT
h
->
pInfo
)
<
0
)
{
if
(
tsdbLoadBlockData
(
pReadh
,
pBlock
,
pT
blHandle
->
pInfo
)
<
0
)
{
return
-
1
;
}
// Merge pTruncateH->pDataCols and pReadh->pDCols[0] and write data to file
if
(
pTruncateH
->
pDataCols
->
numOfRows
==
0
&&
pBlock
->
numOfRows
>=
defaultRows
)
{
if
(
tsdbWriteBlockToRightFile
(
pTruncateH
,
pTh
->
pTable
,
pReadh
->
pDCols
[
0
],
ppBuf
,
ppCBuf
,
ppExBuf
)
<
0
)
{
if
((
pTruncateH
->
pDataCols
->
numOfRows
==
0
)
&&
(
pBlock
->
numOfRows
>=
defaultRows
))
{
// if pTruncateH->pDataCols has no data and pBlock->numOfRows >= defaultRows, write data directly
if
(
tsdbWriteBlockToRightFile
(
pTruncateH
,
pTblHandle
->
pTable
,
pReadh
->
pDCols
[
0
],
ppBuf
,
ppCBuf
,
ppExBuf
)
<
0
)
{
return
-
1
;
}
}
else
{
int
ridx
=
0
;
while
(
true
)
{
if
(
pReadh
->
pDCols
[
0
]
->
numOfRows
-
ridx
==
0
)
break
;
// no data left to merge anymore
if
((
pReadh
->
pDCols
[
0
]
->
numOfRows
-
ridx
)
==
0
)
break
;
// min(data left, target space left)
int
rowsToMerge
=
MIN
(
pReadh
->
pDCols
[
0
]
->
numOfRows
-
ridx
,
defaultRows
-
pTruncateH
->
pDataCols
->
numOfRows
);
tdMergeDataCols
(
pTruncateH
->
pDataCols
,
pReadh
->
pDCols
[
0
],
rowsToMerge
,
&
ridx
,
pCfg
->
update
!=
TD_ROW_PARTIAL_UPDATE
);
if
(
pTruncateH
->
pDataCols
->
numOfRows
<
defaultRows
)
{
// continue to read more blocks
break
;
}
if
(
tsdbWriteBlockToRightFile
(
pTruncateH
,
pT
h
->
pTable
,
pTruncateH
->
pDataCols
,
ppBuf
,
ppCBuf
,
ppExBuf
)
<
0
)
{
if
(
tsdbWriteBlockToRightFile
(
pTruncateH
,
pT
blHandle
->
pTable
,
pTruncateH
->
pDataCols
,
ppBuf
,
ppCBuf
,
ppExBuf
)
<
0
)
{
return
-
1
;
}
tdResetDataCols
(
pTruncateH
->
pDataCols
);
...
...
@@ -485,16 +495,16 @@ static int tsdbTruncateFSetImpl(STruncateH *pTruncateH) {
}
if
(
pTruncateH
->
pDataCols
->
numOfRows
>
0
&&
tsdbWriteBlockToRightFile
(
pTruncateH
,
pT
h
->
pTable
,
pTruncateH
->
pDataCols
,
ppBuf
,
ppCBuf
,
ppExBuf
)
<
0
)
{
tsdbWriteBlockToRightFile
(
pTruncateH
,
pT
blHandle
->
pTable
,
pTruncateH
->
pDataCols
,
ppBuf
,
ppCBuf
,
ppExBuf
)
<
0
)
{
return
-
1
;
}
if
(
tsdbWriteBlockInfoImpl
(
TSDB_TRUNCATE_HEAD_FILE
(
pTruncateH
),
pT
h
->
pTable
,
pTruncateH
->
aSupBlk
,
NULL
,
ppBuf
,
if
(
tsdbWriteBlockInfoImpl
(
TSDB_TRUNCATE_HEAD_FILE
(
pTruncateH
),
pT
blHandle
->
pTable
,
pTruncateH
->
aSupBlk
,
NULL
,
ppBuf
,
&
blkIdx
)
<
0
)
{
return
-
1
;
}
if
((
blkIdx
.
numOfBlocks
>
0
)
&&
(
taosArrayPush
(
pTruncateH
->
aBlkIdx
,
(
void
*
)(
&
blkIdx
))
==
NULL
))
{
if
((
blkIdx
.
numOfBlocks
>
0
)
&&
(
taosArrayPush
(
pTruncateH
->
aBlkIdx
,
(
const
void
*
)(
&
blkIdx
))
==
NULL
))
{
terrno
=
TSDB_CODE_TDB_OUT_OF_MEMORY
;
return
-
1
;
}
...
...
@@ -511,9 +521,9 @@ static int tsdbWriteBlockToRightFile(STruncateH *pTruncateH, STable *pTable, SDa
void
**
ppCBuf
,
void
**
ppExBuf
)
{
STsdbRepo
*
pRepo
=
TSDB_TRUNCATE_REPO
(
pTruncateH
);
STsdbCfg
*
pCfg
=
REPO_CFG
(
pRepo
);
SDFile
*
pDFile
;
bool
isLast
;
SBlock
block
;
SDFile
*
pDFile
=
NULL
;
bool
isLast
=
false
;
SBlock
block
=
{
0
}
;
ASSERT
(
pDataCols
->
numOfRows
>
0
);
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录