Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
da31d92b
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看板
提交
da31d92b
编写于
6月 02, 2023
作者:
H
Hongze Cheng
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
more code
上级
884d2cc8
变更
2
显示空白变更内容
内联
并排
Showing
2 changed file
with
311 addition
and
219 deletion
+311
-219
source/dnode/vnode/src/tsdb/dev/inc/tsdbDataFileRW.h
source/dnode/vnode/src/tsdb/dev/inc/tsdbDataFileRW.h
+2
-0
source/dnode/vnode/src/tsdb/dev/tsdbDataFileRW.c
source/dnode/vnode/src/tsdb/dev/tsdbDataFileRW.c
+309
-219
未找到文件。
source/dnode/vnode/src/tsdb/dev/inc/tsdbDataFileRW.h
浏览文件 @
da31d92b
...
...
@@ -26,6 +26,7 @@ extern "C" {
typedef
TARRAY2
(
SBlockIdx
)
TBlockIdxArray
;
typedef
TARRAY2
(
SDataBlk
)
TDataBlkArray
;
typedef
TARRAY2
(
SColumnDataAgg
)
TColumnDataAggArray
;
// SDataFileReader =============================================
typedef
struct
SDataFileReader
SDataFileReader
;
...
...
@@ -50,6 +51,7 @@ typedef struct SDataFileWriterConfig {
STsdb
*
tsdb
;
bool
hasOldFile
;
STFile
of
[
TSDB_FTYPE_MAX
];
int8_t
cmprAlg
;
int32_t
maxRow
;
int32_t
szPage
;
int64_t
cid
;
...
...
source/dnode/vnode/src/tsdb/dev/tsdbDataFileRW.c
浏览文件 @
da31d92b
...
...
@@ -17,7 +17,6 @@
typedef
struct
{
SFDataPtr
blockIdxPtr
[
1
];
SFDataPtr
dataBlkPtr
[
1
];
SFDataPtr
rsrvd
[
2
];
}
SDataFooter
;
...
...
@@ -101,11 +100,12 @@ struct SDataFileWriter {
const
TBlockIdxArray
*
blockIdxArray
;
int32_t
blockIdxArrayIdx
;
bool
tbHasOldData
;
TABLEID
tbid
[
1
];
const
TDataBlkArray
*
dataBlkArray
;
int32_t
dataBlkArrayIdx
;
SBlockData
bData
[
1
];
int32_t
iRow
;
TABLEID
tbid
[
1
];
}
ctx
[
1
];
STFile
file
[
TSDB_FTYPE_MAX
];
...
...
@@ -119,68 +119,29 @@ struct SDataFileWriter {
STbStatisBlock
sData
[
1
];
};
int32_t
tsdbDataFileWriterOpen
(
const
SDataFileWriterConfig
*
config
,
SDataFileWriter
**
writer
)
{
writer
[
0
]
=
taosMemoryCalloc
(
1
,
sizeof
(
SDataFileWriter
));
if
(
!
writer
[
0
])
return
TSDB_CODE_OUT_OF_MEMORY
;
writer
[
0
]
->
config
[
0
]
=
config
[
0
];
writer
[
0
]
->
ctx
->
opened
=
false
;
return
0
;
}
static
int32_t
tsdbDataFileWriteRemainData
(
SDataFileWriter
*
writer
)
{
// TODO
return
0
;
}
static
int32_t
tsdbDataFileWriteBlockIdx
(
SDataFileWriter
*
writer
)
{
int32_t
code
=
0
;
int32_t
lino
=
0
;
int32_t
vid
=
TD_VID
(
writer
->
config
->
tsdb
->
pVnode
);
int64_t
offset
=
writer
->
file
[
TSDB_FTYPE_HEAD
].
size
;
int64_t
size
=
TARRAY2_DATA_LEN
(
writer
->
dataBlkArray
);
if
(
TARRAY2_SIZE
(
writer
->
blockIdxArray
)
>
0
)
{
code
=
tsdbWriteFile
(
writer
->
fd
[
TSDB_FTYPE_HEAD
],
offset
,
(
void
*
)
TARRAY2_DATA
(
writer
->
blockIdxArray
),
size
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
writer
->
footer
->
blockIdxPtr
->
offset
=
writer
->
file
[
TSDB_FTYPE_HEAD
].
size
;
writer
->
footer
->
blockIdxPtr
->
size
=
TARRAY2_DATA_LEN
(
writer
->
blockIdxArray
);
_exit:
if
(
code
)
{
TSDB_ERROR_LOG
(
vid
,
lino
,
code
);
}
return
code
;
}
static
int32_t
tsdbDataFileWriterCloseCommit
(
SDataFileWriter
*
writer
)
{
int32_t
code
=
0
;
int32_t
lino
=
0
;
int32_t
vid
=
TD_VID
(
writer
->
config
->
tsdb
->
pVnode
);
code
=
tsdbDataFileWriteRemainData
(
writer
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
code
=
tsdbDataFileWriteBlockIdx
(
writer
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
for
(
int32_t
i
=
0
;
i
<
TSDB_FTYPE_MAX
;
++
i
)
{
if
(
writer
->
fd
[
i
])
{
code
=
tsdbFsyncFile
(
writer
->
fd
[
i
]);
if
(
writer
->
footer
->
blockIdxPtr
->
size
)
{
code
=
tsdbWriteFile
(
writer
->
fd
[
TSDB_FTYPE_HEAD
],
writer
->
footer
->
blockIdxPtr
->
offset
,
(
void
*
)
TARRAY2_DATA
(
writer
->
blockIdxArray
),
writer
->
footer
->
blockIdxPtr
->
size
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
tsdbCloseFile
(
&
writer
->
fd
[
i
]);
}
writer
->
file
[
TSDB_FTYPE_HEAD
].
size
+=
writer
->
footer
->
blockIdxPtr
->
size
;
}
_exit:
if
(
code
)
{
TSDB_ERROR_LOG
(
vid
,
lino
,
code
);
TSDB_ERROR_LOG
(
TD_VID
(
writer
->
config
->
tsdb
->
pVnode
)
,
lino
,
code
);
}
return
code
;
}
static
int32_t
tsdbDataFileWriterCloseAbort
(
SDataFileWriter
*
writer
)
{
// TODO
ASSERT
(
0
);
return
0
;
}
...
...
@@ -189,35 +150,6 @@ static int32_t tsdbDataFileWriterDoClose(SDataFileWriter *writer) {
return
0
;
}
int32_t
tsdbDataFileWriterClose
(
SDataFileWriter
**
writer
,
bool
abort
,
STFileOp
op
[
/*TSDB_FTYPE_MAX*/
])
{
int32_t
code
=
0
;
int32_t
lino
=
0
;
int32_t
vid
=
TD_VID
(
writer
[
0
]
->
config
->
tsdb
->
pVnode
);
if
(
!
writer
[
0
]
->
ctx
->
opened
)
{
for
(
int32_t
i
=
0
;
i
<
TSDB_FTYPE_MAX
;
++
i
)
{
op
[
i
].
optype
=
TSDB_FOP_NONE
;
}
}
else
{
if
(
abort
)
{
code
=
tsdbDataFileWriterCloseAbort
(
writer
[
0
]);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
else
{
code
=
tsdbDataFileWriterCloseCommit
(
writer
[
0
]);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
tsdbDataFileWriterDoClose
(
writer
[
0
]);
}
taosMemoryFree
(
writer
);
writer
[
0
]
=
NULL
;
_exit:
if
(
code
)
{
TSDB_ERROR_LOG
(
vid
,
lino
,
code
);
}
return
code
;
}
static
int32_t
tsdbDataFileWriterDoOpen
(
SDataFileWriter
*
writer
)
{
int32_t
code
=
0
;
int32_t
lino
=
0
;
...
...
@@ -256,27 +188,101 @@ _exit:
return
code
;
}
static
int32_t
tsdbDataFileWrite
BlockData
(
SDataFileWriter
*
writer
,
SBlockData
*
bData
)
{
static
int32_t
tsdbDataFileWrite
DataBlock
(
SDataFileWriter
*
writer
,
SBlockData
*
bData
)
{
if
(
bData
->
nRow
==
0
)
return
0
;
ASSERT
(
bData
->
uid
);
int32_t
code
=
0
;
int32_t
lino
=
0
;
int32_t
vid
=
TD_VID
(
writer
->
config
->
tsdb
->
pVnode
);
SDataBlk
dataBlk
[
1
];
SDataBlk
dataBlk
[
1
]
=
{{
.
minKey
=
{
.
ts
=
bData
->
aTSKEY
[
0
],
.
version
=
bData
->
aVersion
[
0
],
},
.
maxKey
=
{
.
ts
=
bData
->
aTSKEY
[
bData
->
nRow
-
1
],
.
version
=
bData
->
aVersion
[
bData
->
nRow
-
1
],
},
.
minVer
=
bData
->
aVersion
[
0
],
.
maxVer
=
bData
->
aVersion
[
0
],
.
nRow
=
bData
->
nRow
,
.
hasDup
=
0
,
.
nSubBlock
=
1
,
}};
// TODO: fill dataBlk
for
(
int32_t
i
=
1
;
i
<
bData
->
nRow
;
++
i
)
{
if
(
bData
->
aTSKEY
[
i
]
==
bData
->
aTSKEY
[
i
-
1
])
{
dataBlk
->
hasDup
=
1
;
}
dataBlk
->
minVer
=
TMIN
(
dataBlk
->
minVer
,
bData
->
aVersion
[
i
]);
dataBlk
->
maxVer
=
TMAX
(
dataBlk
->
maxVer
,
bData
->
aVersion
[
i
]);
}
int32_t
sizeArr
[
5
]
=
{
0
};
// to .data
code
=
tCmprBlockData
(
bData
,
writer
->
config
->
cmprAlg
,
NULL
,
NULL
,
writer
->
config
->
bufArr
,
sizeArr
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
dataBlk
->
aSubBlock
->
offset
=
writer
->
file
[
TSDB_FTYPE_DATA
].
size
;
dataBlk
->
aSubBlock
->
szKey
=
sizeArr
[
3
]
+
sizeArr
[
2
];
dataBlk
->
aSubBlock
->
szBlock
=
dataBlk
->
aSubBlock
->
szKey
+
sizeArr
[
1
]
+
sizeArr
[
0
];
for
(
int32_t
i
=
3
;
i
>=
0
;
--
i
)
{
if
(
sizeArr
[
i
])
{
code
=
tsdbWriteFile
(
writer
->
fd
[
TSDB_FTYPE_DATA
],
writer
->
file
[
TSDB_FTYPE_DATA
].
size
,
writer
->
config
->
bufArr
[
i
],
sizeArr
[
i
]);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
writer
->
file
[
TSDB_FTYPE_DATA
].
size
+=
sizeArr
[
i
];
}
}
// to .sma
TColumnDataAggArray
smaArr
[
1
]
=
{
0
};
for
(
int32_t
i
=
0
;
i
<
bData
->
nColData
;
++
i
)
{
SColData
*
colData
=
bData
->
aColData
+
i
;
if
((
!
colData
->
smaOn
)
//
||
((
colData
->
flag
&
HAS_VALUE
)
==
0
)
//
)
{
continue
;
}
SColumnDataAgg
sma
[
1
]
=
{{.
colId
=
colData
->
cid
}};
tColDataCalcSMA
[
colData
->
type
](
colData
,
&
sma
->
sum
,
&
sma
->
max
,
&
sma
->
min
,
&
sma
->
numOfNull
);
code
=
TARRAY2_APPEND_PTR
(
smaArr
,
sma
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
// TODO: write data
dataBlk
->
smaInfo
.
offset
=
writer
->
file
[
TSDB_FTYPE_SMA
].
size
;
dataBlk
->
smaInfo
.
size
=
TARRAY2_DATA_LEN
(
smaArr
);
if
(
dataBlk
->
smaInfo
.
size
)
{
code
=
tsdbWriteFile
(
writer
->
fd
[
TSDB_FTYPE_SMA
],
dataBlk
->
smaInfo
.
offset
,
(
const
uint8_t
*
)
TARRAY2_DATA
(
smaArr
),
dataBlk
->
smaInfo
.
size
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
writer
->
file
[
TSDB_FTYPE_SMA
].
size
+=
dataBlk
->
smaInfo
.
size
;
}
TARRAY2_FREE
(
smaArr
);
// to dataBlkArray
code
=
TARRAY2_APPEND_PTR
(
writer
->
dataBlkArray
,
dataBlk
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
tBlockDataClear
(
bData
);
if
(
bData
==
writer
->
bData
)
{
tBlockDataClear
(
writer
->
bData
);
}
_exit:
if
(
code
)
{
TSDB_ERROR_LOG
(
vid
,
lino
,
code
);
TSDB_ERROR_LOG
(
TD_VID
(
writer
->
config
->
tsdb
->
pVnode
)
,
lino
,
code
);
}
return
code
;
}
...
...
@@ -286,7 +292,6 @@ static int32_t tsdbDataFileWriteDataBlk(SDataFileWriter *writer, const TDataBlkA
int32_t
code
=
0
;
int32_t
lino
=
0
;
int32_t
vid
=
TD_VID
(
writer
->
config
->
tsdb
->
pVnode
);
SBlockIdx
blockIdx
[
1
];
blockIdx
->
suid
=
writer
->
ctx
->
tbid
->
suid
;
...
...
@@ -304,7 +309,7 @@ static int32_t tsdbDataFileWriteDataBlk(SDataFileWriter *writer, const TDataBlkA
_exit:
if
(
code
)
{
TSDB_ERROR_LOG
(
vid
,
lino
,
code
);
TSDB_ERROR_LOG
(
TD_VID
(
writer
->
config
->
tsdb
->
pVnode
)
,
lino
,
code
);
}
return
code
;
}
...
...
@@ -312,20 +317,93 @@ _exit:
static
int32_t
tsdbDataFileDoWriteTSRow
(
SDataFileWriter
*
writer
,
TSDBROW
*
row
)
{
int32_t
code
=
0
;
int32_t
lino
=
0
;
int32_t
vid
=
TD_VID
(
writer
->
config
->
tsdb
->
pVnode
);
// update/append
if
(
row
->
type
==
TSDBROW_ROW_FMT
)
{
// TODO: udpate row schema
code
=
tsdbUpdateSkmRow
(
writer
->
config
->
tsdb
,
writer
->
ctx
->
tbid
,
TSDBROW_SVERSION
(
row
),
writer
->
config
->
skmRow
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
code
=
tBlockDataAppendRow
(
writer
->
ctx
->
bData
,
row
,
NULL
/* TODO */
,
writer
->
ctx
->
tbid
->
uid
);
TSDBKEY
key
[
1
]
=
{
TSDBROW_KEY
(
row
)};
if
(
key
->
version
<=
writer
->
config
->
compactVersion
//
&&
writer
->
bData
->
nRow
>
0
//
&&
writer
->
bData
->
aTSKEY
[
writer
->
bData
->
nRow
-
1
]
==
key
->
ts
//
)
{
code
=
tBlockDataUpdateRow
(
writer
->
bData
,
row
,
writer
->
config
->
skmRow
->
pTSchema
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
else
{
if
(
writer
->
bData
->
nRow
>=
writer
->
config
->
maxRow
)
{
code
=
tsdbDataFileWriteDataBlock
(
writer
,
writer
->
bData
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
if
(
writer
->
ctx
->
bData
->
nRow
>=
writer
->
config
->
maxRow
)
{
code
=
tsdbDataFileWriteBlockData
(
writer
,
writer
->
bData
);
code
=
tBlockDataAppendRow
(
writer
->
bData
,
row
,
writer
->
config
->
skmRow
->
pTSchema
,
writer
->
ctx
->
tbid
->
uid
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
_exit:
if
(
code
)
{
TSDB_ERROR_LOG
(
TD_VID
(
writer
->
config
->
tsdb
->
pVnode
),
lino
,
code
);
}
return
code
;
}
static
int32_t
tsdbDataFileDoWriteTSData
(
SDataFileWriter
*
writer
,
TSDBROW
*
row
)
{
int32_t
code
=
0
;
int32_t
lino
=
0
;
int32_t
vid
=
TD_VID
(
writer
->
config
->
tsdb
->
pVnode
);
while
(
writer
->
ctx
->
tbHasOldData
)
{
for
(;
writer
->
ctx
->
iRow
<
writer
->
ctx
->
bData
->
nRow
;
writer
->
ctx
->
iRow
++
)
{
TSDBROW
row1
[
1
]
=
{
tsdbRowFromBlockData
(
writer
->
ctx
->
bData
,
writer
->
ctx
->
iRow
)};
int32_t
c
=
tsdbRowCmprFn
(
row
,
row1
);
ASSERT
(
c
);
if
(
c
>
0
)
{
code
=
tsdbDataFileDoWriteTSRow
(
writer
,
row1
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
else
{
goto
_do_write
;
}
}
if
(
writer
->
ctx
->
dataBlkArrayIdx
>=
TARRAY2_SIZE
(
writer
->
ctx
->
dataBlkArray
))
{
writer
->
ctx
->
tbHasOldData
=
false
;
break
;
}
for
(;
writer
->
ctx
->
dataBlkArrayIdx
<
TARRAY2_SIZE
(
writer
->
ctx
->
dataBlkArray
);
writer
->
ctx
->
dataBlkArrayIdx
++
)
{
const
SDataBlk
*
dataBlk
=
TARRAY2_GET_PTR
(
writer
->
ctx
->
dataBlkArray
,
writer
->
ctx
->
dataBlkArrayIdx
);
TSDBKEY
key
=
TSDBROW_KEY
(
row
);
SDataBlk
dataBlk1
[
1
]
=
{{
.
minKey
=
key
,
.
maxKey
=
key
,
}};
int32_t
c
=
tDataBlkCmprFn
(
dataBlk
,
dataBlk1
);
if
(
c
<
0
)
{
code
=
tsdbDataFileWriteDataBlock
(
writer
,
writer
->
bData
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
code
=
TARRAY2_APPEND_PTR
(
writer
->
dataBlkArray
,
dataBlk
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
else
if
(
c
>
0
)
{
goto
_do_write
;
}
else
{
code
=
tsdbDataFileReadDataBlock
(
writer
->
ctx
->
reader
,
dataBlk
,
writer
->
ctx
->
bData
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
writer
->
ctx
->
iRow
=
0
;
writer
->
ctx
->
dataBlkArrayIdx
++
;
break
;
}
}
}
_do_write:
code
=
tsdbDataFileDoWriteTSRow
(
writer
,
row
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
_exit:
if
(
code
)
{
TSDB_ERROR_LOG
(
vid
,
lino
,
code
);
...
...
@@ -339,6 +417,7 @@ static int32_t tsdbDataFileWriteTableDataEnd(SDataFileWriter *writer) {
int32_t
code
=
0
;
int32_t
lino
=
0
;
// handle table remain data
if
(
writer
->
ctx
->
tbHasOldData
)
{
for
(;
writer
->
ctx
->
iRow
<
writer
->
ctx
->
bData
->
nRow
;
writer
->
ctx
->
iRow
++
)
{
TSDBROW
row
[
1
]
=
{
tsdbRowFromBlockData
(
writer
->
ctx
->
bData
,
writer
->
ctx
->
iRow
)};
...
...
@@ -347,7 +426,7 @@ static int32_t tsdbDataFileWriteTableDataEnd(SDataFileWriter *writer) {
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
code
=
tsdbDataFileWrite
BlockData
(
writer
,
writer
->
bData
);
code
=
tsdbDataFileWrite
DataBlock
(
writer
,
writer
->
bData
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
for
(;
writer
->
ctx
->
dataBlkArrayIdx
<
TARRAY2_SIZE
(
writer
->
ctx
->
dataBlkArray
);
writer
->
ctx
->
dataBlkArrayIdx
++
)
{
...
...
@@ -355,9 +434,11 @@ static int32_t tsdbDataFileWriteTableDataEnd(SDataFileWriter *writer) {
TARRAY2_GET_PTR
(
writer
->
ctx
->
dataBlkArray
,
writer
->
ctx
->
dataBlkArrayIdx
));
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
writer
->
ctx
->
tbHasOldData
=
false
;
}
code
=
tsdbDataFileWrite
BlockData
(
writer
,
writer
->
bData
);
code
=
tsdbDataFileWrite
DataBlock
(
writer
,
writer
->
bData
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
code
=
tsdbDataFileWriteDataBlk
(
writer
,
writer
->
dataBlkArray
);
...
...
@@ -373,17 +454,18 @@ _exit:
static
int32_t
tsdbDataFileWriteTableDataBegin
(
SDataFileWriter
*
writer
,
const
TABLEID
*
tbid
)
{
int32_t
code
=
0
;
int32_t
lino
=
0
;
int32_t
vid
=
TD_VID
(
writer
->
config
->
tsdb
->
pVnode
)
;
SMetaInfo
info
;
writer
->
ctx
->
tbHasOldData
=
false
;
// skip data of previous table
if
(
writer
->
ctx
->
blockIdxArray
)
{
for
(;
writer
->
ctx
->
blockIdxArrayIdx
<
TARRAY2_SIZE
(
writer
->
ctx
->
blockIdxArray
);
writer
->
ctx
->
blockIdxArrayIdx
++
)
{
const
SBlockIdx
*
blockIdx
=
TARRAY2_GET_PTR
(
writer
->
ctx
->
blockIdxArray
,
writer
->
ctx
->
blockIdxArrayIdx
);
int32_t
c
=
tTABLEIDCmprFn
(
blockIdx
,
tbid
);
if
(
c
<
0
)
{
SMetaInfo
info
;
if
(
metaGetInfo
(
writer
->
config
->
tsdb
->
pVnode
->
pMeta
,
blockIdx
->
suid
,
&
info
,
NULL
)
==
0
)
{
if
(
metaGetInfo
(
writer
->
config
->
tsdb
->
pVnode
->
pMeta
,
blockIdx
->
uid
,
&
info
,
NULL
)
==
0
)
{
code
=
tsdbDataFileReadDataBlk
(
writer
->
ctx
->
reader
,
blockIdx
,
&
writer
->
ctx
->
dataBlkArray
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
...
...
@@ -392,133 +474,160 @@ static int32_t tsdbDataFileWriteTableDataBegin(SDataFileWriter *writer, const TA
code
=
tsdbDataFileWriteDataBlk
(
writer
,
writer
->
ctx
->
dataBlkArray
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
else
{
continue
;
}
}
else
{
if
(
c
==
0
)
{
writer
->
ctx
->
tbHasOldData
=
true
;
code
=
tsdbDataFileReadDataBlk
(
writer
->
ctx
->
reader
,
blockIdx
,
&
writer
->
ctx
->
dataBlkArray
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
writer
->
ctx
->
dataBlkArrayIdx
=
0
;
tBlockDataReset
(
writer
->
ctx
->
bData
);
writer
->
ctx
->
iRow
=
0
;
writer
->
ctx
->
blockIdxArrayIdx
++
;
}
break
;
}
}
}
// make sure state is correct
writer
->
ctx
->
tbid
[
0
]
=
tbid
[
0
];
if
(
tbid
->
suid
==
INT64_MAX
&&
tbid
->
uid
==
INT64_MAX
)
goto
_exit
;
TARRAY2_CLEAR
(
writer
->
dataBlkArray
,
NULL
);
code
=
tsdbUpdateSkmTb
(
writer
->
config
->
tsdb
,
tbid
,
writer
->
config
->
skmTb
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
code
=
tBlockDataInit
(
writer
->
bData
,
writer
->
ctx
->
tbid
,
writer
->
config
->
skmTb
->
pTSchema
,
NULL
,
0
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
_exit:
if
(
code
)
{
TSDB_ERROR_LOG
(
vid
,
lino
,
code
);
TSDB_ERROR_LOG
(
TD_VID
(
writer
->
config
->
tsdb
->
pVnode
)
,
lino
,
code
);
}
return
code
;
}
static
int32_t
tsdbDataFile
DoWriteTableDataRow
(
SDataFileWriter
*
writer
,
TSDBROW
*
row
)
{
static
int32_t
tsdbDataFile
WriteFooter
(
SDataFileWriter
*
writer
)
{
int32_t
code
=
0
;
int32_t
lino
=
0
;
int32_t
vid
=
TD_VID
(
writer
->
config
->
tsdb
->
pVnode
);
while
(
writer
->
ctx
->
tbHasOldData
)
{
for
(;
writer
->
ctx
->
iRow
<
writer
->
ctx
->
bData
->
nRow
;
writer
->
ctx
->
iRow
++
)
{
TSDBROW
row1
[
1
]
=
{
tsdbRowFromBlockData
(
writer
->
ctx
->
bData
,
writer
->
ctx
->
iRow
)};
int32_t
c
=
tsdbRowCmprFn
(
row
,
row1
);
ASSERT
(
c
);
if
(
row
>
0
)
{
code
=
tsdbDataFileDoWriteTSRow
(
writer
,
row1
);
code
=
tsdbWriteFile
(
writer
->
fd
[
TSDB_FTYPE_HEAD
],
writer
->
file
[
TSDB_FTYPE_HEAD
].
size
,
(
const
uint8_t
*
)
writer
->
footer
,
sizeof
(
SDataFooter
));
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
else
{
goto
_write_row
;
}
writer
->
file
[
TSDB_FTYPE_HEAD
].
size
+=
sizeof
(
SDataFooter
);
_exit:
if
(
code
)
{
TSDB_ERROR_LOG
(
TD_VID
(
writer
->
config
->
tsdb
->
pVnode
),
lino
,
code
);
}
return
code
;
}
for
(;
writer
->
ctx
->
dataBlkArrayIdx
<
TARRAY2_SIZE
(
writer
->
ctx
->
dataBlkArray
);
writer
->
ctx
->
dataBlkArrayIdx
++
)
{
const
SDataBlk
*
dataBlk
=
TARRAY2_GET_PTR
(
writer
->
ctx
->
dataBlkArray
,
writer
->
ctx
->
dataBlkArrayIdx
);
SDataBlk
dataBlk1
[
1
]
=
{{.
minKey
=
{},
.
maxKey
=
{}}};
// TODO
static
int32_t
tsdbDataFileWriterCloseCommit
(
SDataFileWriter
*
writer
,
STFileOp
*
op
)
{
int32_t
code
=
0
;
int32_t
lino
=
0
;
TABLEID
tbid
[
1
]
=
{{
INT64_MAX
,
INT64_MAX
}};
int32_t
c
=
tDataBlkCmprFn
(
dataBlk
,
dataBlk1
);
if
(
c
<
0
)
{
code
=
tsdbDataFileWriteBlockData
(
writer
,
writer
->
bData
);
code
=
tsdbDataFileWriteTableDataEnd
(
writer
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
code
=
TARRAY2_APPEND_PTR
(
writer
->
dataBlkArray
,
dataBlk
);
code
=
tsdbDataFileWriteTableDataBegin
(
writer
,
tbid
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
else
if
(
c
>
0
)
{
goto
_write_row
;
}
else
{
code
=
tsdbDataFileReadDataBlock
(
writer
->
ctx
->
reader
,
dataBlk
,
writer
->
ctx
->
bData
);
code
=
tsdbDataFileWriteBlockIdx
(
writer
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
writer
->
ctx
->
iRow
=
0
;
writer
->
ctx
->
dataBlkArrayIdx
++
;
break
;
}
}
if
(
writer
->
ctx
->
dataBlkArrayIdx
>=
TARRAY2_SIZE
(
writer
->
ctx
->
dataBlkArray
)
//
&&
writer
->
ctx
->
iRow
>=
writer
->
ctx
->
bData
->
nRow
)
{
writer
->
ctx
->
tbHasOldData
=
false
;
code
=
tsdbDataFileWriteFooter
(
writer
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
for
(
int32_t
i
=
0
;
i
<
TSDB_FTYPE_MAX
;
++
i
)
{
if
(
writer
->
fd
[
i
])
{
code
=
tsdbFsyncFile
(
writer
->
fd
[
i
]);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
tsdbCloseFile
(
&
writer
->
fd
[
i
]);
}
}
_write_row:
code
=
tsdbDataFileDoWriteTSRow
(
writer
,
row
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
// .head
// .data
// .sma
// .tomb
_exit:
if
(
code
)
{
TSDB_ERROR_LOG
(
vid
,
lino
,
code
);
TSDB_ERROR_LOG
(
TD_VID
(
writer
->
config
->
tsdb
->
pVnode
)
,
lino
,
code
);
}
return
code
;
}
static
int32_t
tsdbDataFileDoWriteTableDataBlock
(
SDataFileWriter
*
writer
,
SBlockData
*
bData
)
{
int32_t
tsdbDataFileWriterOpen
(
const
SDataFileWriterConfig
*
config
,
SDataFileWriter
**
writer
)
{
writer
[
0
]
=
taosMemoryCalloc
(
1
,
sizeof
(
SDataFileWriter
));
if
(
!
writer
[
0
])
return
TSDB_CODE_OUT_OF_MEMORY
;
writer
[
0
]
->
config
[
0
]
=
config
[
0
];
writer
[
0
]
->
ctx
->
opened
=
false
;
return
0
;
}
int32_t
tsdbDataFileWriterClose
(
SDataFileWriter
**
writer
,
bool
abort
,
STFileOp
op
[
/*TSDB_FTYPE_MAX*/
])
{
int32_t
code
=
0
;
int32_t
lino
=
0
;
int32_t
vid
=
TD_VID
(
writer
->
config
->
tsdb
->
pVnode
);
if
(
!
writer
->
ctx
->
tbHasOldData
&&
writer
->
bData
->
nRow
==
0
)
{
code
=
tsdbDataFileWriteBlockData
(
writer
,
bData
);
if
(
!
writer
[
0
]
->
ctx
->
opened
)
{
for
(
int32_t
i
=
0
;
i
<
TSDB_FTYPE_MAX
;
++
i
)
{
op
[
i
].
optype
=
TSDB_FOP_NONE
;
}
}
else
{
if
(
abort
)
{
code
=
tsdbDataFileWriterCloseAbort
(
writer
[
0
]);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
else
{
for
(
int32_t
i
=
0
;
i
<
bData
->
nRow
;
i
++
)
{
TSDBROW
row
[
1
]
=
{
tsdbRowFromBlockData
(
bData
,
i
)};
code
=
tsdbDataFileDoWriteTableDataRow
(
writer
,
row
);
code
=
tsdbDataFileWriterCloseCommit
(
writer
[
0
],
op
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
tsdbDataFileWriterDoClose
(
writer
[
0
]);
}
taosMemoryFree
(
writer
);
writer
[
0
]
=
NULL
;
_exit:
if
(
code
)
{
TSDB_ERROR_LOG
(
vid
,
lino
,
code
);
TSDB_ERROR_LOG
(
TD_VID
(
writer
[
0
]
->
config
->
tsdb
->
pVnode
)
,
lino
,
code
);
}
return
code
;
}
int32_t
tsdbDataFileWriteTSDataBlock
(
SDataFileWriter
*
writer
,
SBlockData
*
bData
)
{
if
(
bData
->
nRow
==
0
)
return
0
;
int32_t
tsdbDataFileWriteTSData
(
SDataFileWriter
*
writer
,
SRowInfo
*
row
)
{
int32_t
code
=
0
;
int32_t
lino
=
0
;
ASSERT
(
bData
->
uid
);
if
(
!
writer
->
ctx
->
opened
)
{
code
=
tsdbDataFileWriterDoOpen
(
writer
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
if
(
bData
->
uid
!=
writer
->
ctx
->
tbid
->
uid
)
{
if
(
row
->
uid
!=
writer
->
ctx
->
tbid
->
uid
)
{
code
=
tsdbDataFileWriteTableDataEnd
(
writer
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
code
=
tsdbDataFileWriteTableDataBegin
(
writer
,
(
TABLEID
*
)
bData
);
code
=
tsdbDataFileWriteTableDataBegin
(
writer
,
(
TABLEID
*
)
row
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
code
=
tsdbDataFileDoWriteT
ableDataBlock
(
writer
,
bData
);
code
=
tsdbDataFileDoWriteT
SData
(
writer
,
&
row
->
row
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
_exit:
...
...
@@ -528,49 +637,29 @@ _exit:
return
code
;
}
int32_t
tsdbDataFileWriteTSData
(
SDataFileWriter
*
writer
,
SRowInfo
*
row
)
{
int32_t
tsdbDataFileWriteTSDataBlock
(
SDataFileWriter
*
writer
,
SBlockData
*
bData
)
{
if
(
bData
->
nRow
==
0
)
return
0
;
int32_t
code
=
0
;
int32_t
lino
=
0
;
if
(
!
writer
->
ctx
->
opened
)
{
code
=
tsdbDataFileWriterDoOpen
(
writer
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
if
(
row
->
uid
!=
writer
->
ctx
->
tbid
->
uid
)
{
code
=
tsdbDataFileWriteTableDataEnd
(
writer
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
writer
->
ctx
->
tbid
->
suid
=
row
->
uid
;
writer
->
ctx
->
tbid
->
uid
=
row
->
uid
;
// ASSERT(bData->uid);
code
=
tsdbDataFileWriteTableDataBegin
(
writer
,
(
TABLEID
*
)
row
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
// if (!writer->ctx->opened) {
// code = tsdbDataFileWriterDoOpen(writer);
// TSDB_CHECK_CODE(code, lino, _exit);
// }
if
(
row
->
row
.
type
==
TSDBROW_ROW_FMT
)
{
code
=
tsdbUpdateSkmRow
(
writer
->
config
->
tsdb
,
writer
->
ctx
->
tbid
,
//
TSDBROW_SVERSION
(
&
row
->
row
),
writer
->
config
->
skmRow
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
// if (bData->uid != writer->ctx->tbid->uid) {
// code = tsdbDataFileWriteTableDataEnd(writer);
// TSDB_CHECK_CODE(code, lino, _exit);
// update/append
TSDBKEY
key
[
1
]
=
{
TSDBROW_KEY
(
&
row
->
row
)};
if
(
key
->
version
<=
writer
->
config
->
compactVersion
//
&&
writer
->
bData
->
nRow
>
0
//
&&
writer
->
bData
->
aTSKEY
[
writer
->
bData
->
nRow
-
1
]
==
key
->
ts
//
)
{
code
=
tBlockDataUpdateRow
(
writer
->
bData
,
&
row
->
row
,
writer
->
config
->
skmRow
->
pTSchema
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
else
{
if
(
writer
->
bData
->
nRow
>=
writer
->
config
->
maxRow
)
{
code
=
tsdbDataFileDoWriteTableDataBlock
(
writer
,
writer
->
bData
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
// code = tsdbDataFileWriteTableDataBegin(writer, (TABLEID *)bData);
// TSDB_CHECK_CODE(code, lino, _exit);
// }
code
=
tBlockDataAppendRow
(
writer
->
bData
,
&
row
->
row
,
writer
->
config
->
skmRow
->
pTSchema
,
row
->
uid
);
TSDB_CHECK_CODE
(
code
,
lino
,
_exit
);
}
// code = tsdbDataFileDoWriteTableDataBlock(writer, bData);
// TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if
(
code
)
{
...
...
@@ -580,6 +669,7 @@ _exit:
}
int32_t
tsdbDataFileFLushTSDataBlock
(
SDataFileWriter
*
writer
)
{
if
(
writer
->
bData
->
nRow
==
0
)
return
0
;
return
tsdbDataFileDoWriteTableDataBlock
(
writer
,
writer
->
bData
);
// if (writer->bData->nRow == 0) return 0;
// return tsdbDataFileDoWriteTableDataBlock(writer, writer->bData);
return
0
;
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录