Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
a0c5cece
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1181
Star
22014
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
a0c5cece
编写于
10月 27, 2022
作者:
S
Shengliang Guan
提交者:
GitHub
10月 27, 2022
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #17697 from taosdata/feature/stream_ly
fix(stream): read preversion data for delete
上级
52d8f7bd
442f8978
变更
1
隐藏空白更改
内联
并排
Showing
1 changed file
with
33 addition
and
29 deletion
+33
-29
source/dnode/vnode/src/tsdb/tsdbRead.c
source/dnode/vnode/src/tsdb/tsdbRead.c
+33
-29
未找到文件。
source/dnode/vnode/src/tsdb/tsdbRead.c
浏览文件 @
a0c5cece
...
...
@@ -111,7 +111,7 @@ typedef struct SDataBlockIter {
int32_t
index
;
SArray
*
blockList
;
// SArray<SFileDataBlockInfo>
int32_t
order
;
SDataBlk
block
;
// current SDataBlk data
SDataBlk
block
;
// current SDataBlk data
SHashObj
*
pTableMap
;
}
SDataBlockIter
;
...
...
@@ -169,14 +169,14 @@ static TSDBROW* getValidMemRow(SIterInfo* pIter, const SArray* pDelList, STsdbRe
static
int32_t
doMergeRowsInFileBlocks
(
SBlockData
*
pBlockData
,
STableBlockScanInfo
*
pScanInfo
,
STsdbReader
*
pReader
,
SRowMerger
*
pMerger
);
static
int32_t
doMergeRowsInLastBlock
(
SLastBlockReader
*
pLastBlockReader
,
STableBlockScanInfo
*
pScanInfo
,
int64_t
ts
,
SRowMerger
*
pMerger
);
SRowMerger
*
pMerger
,
SVersionRange
*
pVerRange
);
static
int32_t
doMergeRowsInBuf
(
SIterInfo
*
pIter
,
uint64_t
uid
,
int64_t
ts
,
SArray
*
pDelList
,
SRowMerger
*
pMerger
,
STsdbReader
*
pReader
);
static
int32_t
doAppendRowFromTSRow
(
SSDataBlock
*
pBlock
,
STsdbReader
*
pReader
,
STSRow
*
pTSRow
,
uint64_t
uid
);
static
int32_t
doAppendRowFromFileBlock
(
SSDataBlock
*
pResBlock
,
STsdbReader
*
pReader
,
SBlockData
*
pBlockData
,
int32_t
rowIndex
);
static
void
setComposedBlockFlag
(
STsdbReader
*
pReader
,
bool
composed
);
static
bool
hasBeenDropped
(
const
SArray
*
pDelList
,
int32_t
*
index
,
TSDBKEY
*
pKey
,
int32_t
order
);
static
bool
hasBeenDropped
(
const
SArray
*
pDelList
,
int32_t
*
index
,
TSDBKEY
*
pKey
,
int32_t
order
,
SVersionRange
*
pRange
);
static
int32_t
doMergeMemTableMultiRows
(
TSDBROW
*
pRow
,
uint64_t
uid
,
SIterInfo
*
pIter
,
SArray
*
pDelList
,
STSRow
**
pTSRow
,
STsdbReader
*
pReader
,
bool
*
freeTSRow
);
...
...
@@ -1052,7 +1052,7 @@ static int32_t doSetCurrentBlock(SDataBlockIter* pBlockIter, const char* idStr)
if
(
pBlockInfo
!=
NULL
)
{
STableBlockScanInfo
*
pScanInfo
=
taosHashGet
(
pBlockIter
->
pTableMap
,
&
pBlockInfo
->
uid
,
sizeof
(
pBlockInfo
->
uid
));
if
(
pScanInfo
==
NULL
)
{
tsdbError
(
"failed to locate the uid:%"
PRIu64
" in query table uid list, %s"
,
pBlockInfo
->
uid
,
idStr
);
tsdbError
(
"failed to locate the uid:%"
PRIu64
" in query table uid list, %s"
,
pBlockInfo
->
uid
,
idStr
);
return
TSDB_CODE_INVALID_PARA
;
}
...
...
@@ -1466,7 +1466,8 @@ static bool tryCopyDistinctRowFromFileBlock(STsdbReader* pReader, SBlockData* pB
return
false
;
}
static
bool
nextRowFromLastBlocks
(
SLastBlockReader
*
pLastBlockReader
,
STableBlockScanInfo
*
pBlockScanInfo
)
{
static
bool
nextRowFromLastBlocks
(
SLastBlockReader
*
pLastBlockReader
,
STableBlockScanInfo
*
pBlockScanInfo
,
SVersionRange
*
pVerRange
)
{
while
(
1
)
{
bool
hasVal
=
tMergeTreeNext
(
&
pLastBlockReader
->
mergeTree
);
if
(
!
hasVal
)
{
...
...
@@ -1475,7 +1476,8 @@ static bool nextRowFromLastBlocks(SLastBlockReader* pLastBlockReader, STableBloc
TSDBROW
row
=
tMergeTreeGetRow
(
&
pLastBlockReader
->
mergeTree
);
TSDBKEY
k
=
TSDBROW_KEY
(
&
row
);
if
(
!
hasBeenDropped
(
pBlockScanInfo
->
delSkyline
,
&
pBlockScanInfo
->
lastBlockDelIndex
,
&
k
,
pLastBlockReader
->
order
))
{
if
(
!
hasBeenDropped
(
pBlockScanInfo
->
delSkyline
,
&
pBlockScanInfo
->
lastBlockDelIndex
,
&
k
,
pLastBlockReader
->
order
,
pVerRange
))
{
return
true
;
}
}
...
...
@@ -1483,7 +1485,7 @@ static bool nextRowFromLastBlocks(SLastBlockReader* pLastBlockReader, STableBloc
static
bool
tryCopyDistinctRowFromSttBlock
(
TSDBROW
*
fRow
,
SLastBlockReader
*
pLastBlockReader
,
STableBlockScanInfo
*
pScanInfo
,
int64_t
ts
,
STsdbReader
*
pReader
)
{
bool
hasVal
=
nextRowFromLastBlocks
(
pLastBlockReader
,
pScanInfo
);
bool
hasVal
=
nextRowFromLastBlocks
(
pLastBlockReader
,
pScanInfo
,
&
pReader
->
verRange
);
if
(
hasVal
)
{
int64_t
next1
=
getCurrentKeyInLastBlock
(
pLastBlockReader
);
if
(
next1
!=
ts
)
{
...
...
@@ -1602,7 +1604,7 @@ static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo*
return
code
;
}
}
doMergeRowsInLastBlock
(
pLastBlockReader
,
pBlockScanInfo
,
tsLast
,
&
merge
);
doMergeRowsInLastBlock
(
pLastBlockReader
,
pBlockScanInfo
,
tsLast
,
&
merge
,
&
pReader
->
verRange
);
}
if
(
minKey
==
k
.
ts
)
{
...
...
@@ -1647,7 +1649,7 @@ static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo*
return
code
;
}
}
doMergeRowsInLastBlock
(
pLastBlockReader
,
pBlockScanInfo
,
tsLast
,
&
merge
);
doMergeRowsInLastBlock
(
pLastBlockReader
,
pBlockScanInfo
,
tsLast
,
&
merge
,
&
pReader
->
verRange
);
}
if
(
minKey
==
key
)
{
...
...
@@ -1699,7 +1701,7 @@ static int32_t doMergeFileBlockAndLastBlock(SLastBlockReader* pLastBlockReader,
TSDBROW
fRow1
=
tMergeTreeGetRow
(
&
pLastBlockReader
->
mergeTree
);
tRowMerge
(
&
merge
,
&
fRow1
);
doMergeRowsInLastBlock
(
pLastBlockReader
,
pBlockScanInfo
,
tsLastBlock
,
&
merge
);
doMergeRowsInLastBlock
(
pLastBlockReader
,
pBlockScanInfo
,
tsLastBlock
,
&
merge
,
&
pReader
->
verRange
);
code
=
tRowMergerGetRow
(
&
merge
,
&
pTSRow
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
...
...
@@ -1717,7 +1719,7 @@ static int32_t doMergeFileBlockAndLastBlock(SLastBlockReader* pLastBlockReader,
return
code
;
}
doMergeRowsInLastBlock
(
pLastBlockReader
,
pBlockScanInfo
,
tsLastBlock
,
&
merge
);
doMergeRowsInLastBlock
(
pLastBlockReader
,
pBlockScanInfo
,
tsLastBlock
,
&
merge
,
&
pReader
->
verRange
);
ASSERT
(
mergeBlockData
);
// merge with block data if ts == key
...
...
@@ -1771,7 +1773,7 @@ static int32_t mergeFileBlockAndLastBlock(STsdbReader* pReader, SLastBlockReader
TSDBROW
fRow1
=
tMergeTreeGetRow
(
&
pLastBlockReader
->
mergeTree
);
tRowMerge
(
&
merge
,
&
fRow1
);
doMergeRowsInLastBlock
(
pLastBlockReader
,
pBlockScanInfo
,
ts
,
&
merge
);
doMergeRowsInLastBlock
(
pLastBlockReader
,
pBlockScanInfo
,
ts
,
&
merge
,
&
pReader
->
verRange
);
code
=
tRowMergerGetRow
(
&
merge
,
&
pTSRow
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
...
...
@@ -1882,7 +1884,7 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo*
}
}
doMergeRowsInLastBlock
(
pLastBlockReader
,
pBlockScanInfo
,
tsLast
,
&
merge
);
doMergeRowsInLastBlock
(
pLastBlockReader
,
pBlockScanInfo
,
tsLast
,
&
merge
,
&
pReader
->
verRange
);
}
if
(
minKey
==
ik
.
ts
)
{
...
...
@@ -1901,8 +1903,8 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo*
}
}
code
=
doMergeRowsInBuf
(
&
pBlockScanInfo
->
iiter
,
pBlockScanInfo
->
uid
,
ik
.
ts
,
pBlockScanInfo
->
delSkyline
,
&
merge
,
pReader
);
code
=
doMergeRowsInBuf
(
&
pBlockScanInfo
->
iiter
,
pBlockScanInfo
->
uid
,
ik
.
ts
,
pBlockScanInfo
->
delSkyline
,
&
merge
,
pReader
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
return
code
;
}
...
...
@@ -1973,7 +1975,7 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo*
return
code
;
}
}
doMergeRowsInLastBlock
(
pLastBlockReader
,
pBlockScanInfo
,
tsLast
,
&
merge
);
doMergeRowsInLastBlock
(
pLastBlockReader
,
pBlockScanInfo
,
tsLast
,
&
merge
,
&
pReader
->
verRange
);
}
if
(
minKey
==
key
)
{
...
...
@@ -1993,7 +1995,7 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo*
}
}
if
(
merge
.
pTSchema
==
NULL
)
{
if
(
merge
.
pTSchema
==
NULL
)
{
return
code
;
}
...
...
@@ -2095,7 +2097,8 @@ static bool isValidFileBlockRow(SBlockData* pBlockData, SFileBlockDumpInfo* pDum
}
TSDBKEY
k
=
{.
ts
=
ts
,
.
version
=
ver
};
if
(
hasBeenDropped
(
pBlockScanInfo
->
delSkyline
,
&
pBlockScanInfo
->
fileDelIndex
,
&
k
,
pReader
->
order
))
{
if
(
hasBeenDropped
(
pBlockScanInfo
->
delSkyline
,
&
pBlockScanInfo
->
fileDelIndex
,
&
k
,
pReader
->
order
,
&
pReader
->
verRange
))
{
return
false
;
}
...
...
@@ -2130,7 +2133,7 @@ static bool initLastBlockReader(SLastBlockReader* pLBlockReader, STableBlockScan
return
false
;
}
return
nextRowFromLastBlocks
(
pLBlockReader
,
pScanInfo
);
return
nextRowFromLastBlocks
(
pLBlockReader
,
pScanInfo
,
&
pReader
->
verRange
);
}
static
int64_t
getCurrentKeyInLastBlock
(
SLastBlockReader
*
pLastBlockReader
)
{
...
...
@@ -2225,8 +2228,8 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
pBlockScanInfo
=
taosHashGet
(
pReader
->
status
.
pTableMap
,
&
pBlockInfo
->
uid
,
sizeof
(
pBlockInfo
->
uid
));
if
(
pBlockScanInfo
==
NULL
)
{
code
=
TSDB_CODE_INVALID_PARA
;
tsdbError
(
"failed to locate the uid:%"
PRIu64
" in query table uid list, total tables:%d, %s"
,
pBlockInfo
->
uid
,
taosHashGetSize
(
pReader
->
status
.
pTableMap
),
pReader
->
idStr
);
tsdbError
(
"failed to locate the uid:%"
PRIu64
" in query table uid list, total tables:%d, %s"
,
pBlockInfo
->
uid
,
taosHashGetSize
(
pReader
->
status
.
pTableMap
),
pReader
->
idStr
);
goto
_end
;
}
...
...
@@ -2290,7 +2293,7 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
}
_end:
pResBlock
->
info
.
uid
=
(
pBlockScanInfo
!=
NULL
)
?
pBlockScanInfo
->
uid
:
0
;
pResBlock
->
info
.
uid
=
(
pBlockScanInfo
!=
NULL
)
?
pBlockScanInfo
->
uid
:
0
;
blockDataUpdateTsWindow
(
pResBlock
,
0
);
setComposedBlockFlag
(
pReader
,
true
);
...
...
@@ -2859,7 +2862,7 @@ SVersionRange getQueryVerRange(SVnode* pVnode, SQueryTableDataCond* pCond, int8_
return
(
SVersionRange
){.
minVer
=
startVer
,
.
maxVer
=
endVer
};
}
bool
hasBeenDropped
(
const
SArray
*
pDelList
,
int32_t
*
index
,
TSDBKEY
*
pKey
,
int32_t
order
)
{
bool
hasBeenDropped
(
const
SArray
*
pDelList
,
int32_t
*
index
,
TSDBKEY
*
pKey
,
int32_t
order
,
SVersionRange
*
pVerRange
)
{
ASSERT
(
pKey
!=
NULL
);
if
(
pDelList
==
NULL
)
{
return
false
;
...
...
@@ -2887,7 +2890,8 @@ bool hasBeenDropped(const SArray* pDelList, int32_t* index, TSDBKEY* pKey, int32
return
false
;
}
if
(
pCurrent
->
ts
<=
pKey
->
ts
&&
pNext
->
ts
>=
pKey
->
ts
&&
pCurrent
->
version
>=
pKey
->
version
)
{
if
(
pCurrent
->
ts
<=
pKey
->
ts
&&
pNext
->
ts
>=
pKey
->
ts
&&
pCurrent
->
version
>=
pKey
->
version
&&
pVerRange
->
maxVer
>=
pCurrent
->
version
)
{
return
true
;
}
...
...
@@ -2973,7 +2977,7 @@ TSDBROW* getValidMemRow(SIterInfo* pIter, const SArray* pDelList, STsdbReader* p
// it is a valid data version
if
((
key
.
version
<=
pReader
->
verRange
.
maxVer
&&
key
.
version
>=
pReader
->
verRange
.
minVer
)
&&
(
!
hasBeenDropped
(
pDelList
,
&
pIter
->
index
,
&
key
,
pReader
->
order
)))
{
(
!
hasBeenDropped
(
pDelList
,
&
pIter
->
index
,
&
key
,
pReader
->
order
,
&
pReader
->
verRange
)))
{
return
pRow
;
}
...
...
@@ -2992,7 +2996,7 @@ TSDBROW* getValidMemRow(SIterInfo* pIter, const SArray* pDelList, STsdbReader* p
}
if
(
key
.
version
<=
pReader
->
verRange
.
maxVer
&&
key
.
version
>=
pReader
->
verRange
.
minVer
&&
(
!
hasBeenDropped
(
pDelList
,
&
pIter
->
index
,
&
key
,
pReader
->
order
)))
{
(
!
hasBeenDropped
(
pDelList
,
&
pIter
->
index
,
&
key
,
pReader
->
order
,
&
pReader
->
verRange
)))
{
return
pRow
;
}
}
...
...
@@ -3130,9 +3134,9 @@ int32_t doMergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pSc
}
int32_t
doMergeRowsInLastBlock
(
SLastBlockReader
*
pLastBlockReader
,
STableBlockScanInfo
*
pScanInfo
,
int64_t
ts
,
SRowMerger
*
pMerger
)
{
SRowMerger
*
pMerger
,
SVersionRange
*
pVerRange
)
{
pScanInfo
->
lastKey
=
ts
;
while
(
nextRowFromLastBlocks
(
pLastBlockReader
,
pScanInfo
))
{
while
(
nextRowFromLastBlocks
(
pLastBlockReader
,
pScanInfo
,
pVerRange
))
{
int64_t
next1
=
getCurrentKeyInLastBlock
(
pLastBlockReader
);
if
(
next1
==
ts
)
{
TSDBROW
fRow1
=
tMergeTreeGetRow
(
&
pLastBlockReader
->
mergeTree
);
...
...
@@ -3772,7 +3776,7 @@ bool tsdbNextDataBlock(STsdbReader* pReader) {
bool
tsdbTableNextDataBlock
(
STsdbReader
*
pReader
,
uint64_t
uid
)
{
STableBlockScanInfo
*
pBlockScanInfo
=
taosHashGet
(
pReader
->
status
.
pTableMap
,
&
uid
,
sizeof
(
uid
));
if
(
pBlockScanInfo
==
NULL
)
{
// no data block for the table of given uid
if
(
pBlockScanInfo
==
NULL
)
{
// no data block for the table of given uid
return
false
;
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录