Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
5663550d
T
TDengine
项目概览
taosdata
/
TDengine
1 年多 前同步成功
通知
1185
Star
22016
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看板
提交
5663550d
编写于
4月 24, 2023
作者:
H
Haojun Liao
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
enh(query): opt query perf by allocate the SLDataIter when opening reader.
上级
7e943260
变更
6
隐藏空白更改
内联
并排
Showing
6 changed file
with
69 addition
and
81 deletion
+69
-81
source/dnode/vnode/inc/vnode.h
source/dnode/vnode/inc/vnode.h
+0
-1
source/dnode/vnode/src/inc/tsdb.h
source/dnode/vnode/src/inc/tsdb.h
+18
-2
source/dnode/vnode/src/tsdb/tsdbCache.c
source/dnode/vnode/src/tsdb/tsdbCache.c
+6
-4
source/dnode/vnode/src/tsdb/tsdbCacheRead.c
source/dnode/vnode/src/tsdb/tsdbCacheRead.c
+10
-1
source/dnode/vnode/src/tsdb/tsdbMergeTree.c
source/dnode/vnode/src/tsdb/tsdbMergeTree.c
+32
-71
source/dnode/vnode/src/tsdb/tsdbRead.c
source/dnode/vnode/src/tsdb/tsdbRead.c
+3
-2
未找到文件。
source/dnode/vnode/inc/vnode.h
浏览文件 @
5663550d
...
@@ -162,7 +162,6 @@ int32_t metaTbCursorPrev(SMTbCursor *pTbCur, ETableType jumpTableType);
...
@@ -162,7 +162,6 @@ int32_t metaTbCursorPrev(SMTbCursor *pTbCur, ETableType jumpTableType);
#endif
#endif
// tsdb
// tsdb
// typedef struct STsdb STsdb;
typedef
struct
STsdbReader
STsdbReader
;
typedef
struct
STsdbReader
STsdbReader
;
#define TSDB_DEFAULT_STT_FILE 8
#define TSDB_DEFAULT_STT_FILE 8
...
...
source/dnode/vnode/src/inc/tsdb.h
浏览文件 @
5663550d
...
@@ -705,7 +705,6 @@ typedef struct SSttBlockLoadInfo {
...
@@ -705,7 +705,6 @@ typedef struct SSttBlockLoadInfo {
typedef
struct
SMergeTree
{
typedef
struct
SMergeTree
{
int8_t
backward
;
int8_t
backward
;
SRBTree
rbt
;
SRBTree
rbt
;
SArray
*
pIterList
;
SLDataIter
*
pIter
;
SLDataIter
*
pIter
;
bool
destroyLoadInfo
;
bool
destroyLoadInfo
;
SSttBlockLoadInfo
*
pLoadInfo
;
SSttBlockLoadInfo
*
pLoadInfo
;
...
@@ -751,9 +750,25 @@ struct SDiskDataBuilder {
...
@@ -751,9 +750,25 @@ struct SDiskDataBuilder {
SBlkInfo
bi
;
SBlkInfo
bi
;
};
};
typedef
struct
SLDataIter
{
SRBTreeNode
node
;
SSttBlk
*
pSttBlk
;
SDataFReader
*
pReader
;
int32_t
iStt
;
int8_t
backward
;
int32_t
iSttBlk
;
int32_t
iRow
;
SRowInfo
rInfo
;
uint64_t
uid
;
STimeWindow
timeWindow
;
SVersionRange
verRange
;
SSttBlockLoadInfo
*
pBlockLoadInfo
;
bool
ignoreEarlierTs
;
}
SLDataIter
;
int32_t
tMergeTreeOpen
(
SMergeTree
*
pMTree
,
int8_t
backward
,
SDataFReader
*
pFReader
,
uint64_t
suid
,
uint64_t
uid
,
int32_t
tMergeTreeOpen
(
SMergeTree
*
pMTree
,
int8_t
backward
,
SDataFReader
*
pFReader
,
uint64_t
suid
,
uint64_t
uid
,
STimeWindow
*
pTimeWindow
,
SVersionRange
*
pVerRange
,
SSttBlockLoadInfo
*
pBlockLoadInfo
,
STimeWindow
*
pTimeWindow
,
SVersionRange
*
pVerRange
,
SSttBlockLoadInfo
*
pBlockLoadInfo
,
bool
destroyLoadInfo
,
const
char
*
idStr
,
bool
strictTimeRange
);
bool
destroyLoadInfo
,
const
char
*
idStr
,
bool
strictTimeRange
,
SLDataIter
*
pLDataIter
);
void
tMergeTreeAddIter
(
SMergeTree
*
pMTree
,
SLDataIter
*
pIter
);
void
tMergeTreeAddIter
(
SMergeTree
*
pMTree
,
SLDataIter
*
pIter
);
bool
tMergeTreeNext
(
SMergeTree
*
pMTree
);
bool
tMergeTreeNext
(
SMergeTree
*
pMTree
);
bool
tMergeTreeIgnoreEarlierTs
(
SMergeTree
*
pMTree
);
bool
tMergeTreeIgnoreEarlierTs
(
SMergeTree
*
pMTree
);
...
@@ -782,6 +797,7 @@ typedef struct SCacheRowsReader {
...
@@ -782,6 +797,7 @@ typedef struct SCacheRowsReader {
STableKeyInfo
*
pTableList
;
// table id list
STableKeyInfo
*
pTableList
;
// table id list
int32_t
numOfTables
;
int32_t
numOfTables
;
SSttBlockLoadInfo
*
pLoadInfo
;
SSttBlockLoadInfo
*
pLoadInfo
;
SLDataIter
*
pDataIter
;
STsdbReadSnap
*
pReadSnap
;
STsdbReadSnap
*
pReadSnap
;
SDataFReader
*
pDataFReader
;
SDataFReader
*
pDataFReader
;
SDataFReader
*
pDataFReaderLast
;
SDataFReader
*
pDataFReaderLast
;
...
...
source/dnode/vnode/src/tsdb/tsdbCache.c
浏览文件 @
5663550d
...
@@ -598,6 +598,7 @@ typedef struct {
...
@@ -598,6 +598,7 @@ typedef struct {
SMergeTree
mergeTree
;
SMergeTree
mergeTree
;
SMergeTree
*
pMergeTree
;
SMergeTree
*
pMergeTree
;
SSttBlockLoadInfo
*
pLoadInfo
;
SSttBlockLoadInfo
*
pLoadInfo
;
SLDataIter
*
pDataIter
;
int64_t
lastTs
;
int64_t
lastTs
;
}
SFSLastNextRowIter
;
}
SFSLastNextRowIter
;
...
@@ -645,7 +646,7 @@ static int32_t getNextRowFromFSLast(void *iter, TSDBROW **ppRow, bool *pIgnoreEa
...
@@ -645,7 +646,7 @@ static int32_t getNextRowFromFSLast(void *iter, TSDBROW **ppRow, bool *pIgnoreEa
}
}
tMergeTreeOpen
(
&
state
->
mergeTree
,
1
,
*
state
->
pDataFReader
,
state
->
suid
,
state
->
uid
,
tMergeTreeOpen
(
&
state
->
mergeTree
,
1
,
*
state
->
pDataFReader
,
state
->
suid
,
state
->
uid
,
&
(
STimeWindow
){.
skey
=
state
->
lastTs
,
.
ekey
=
TSKEY_MAX
},
&
(
STimeWindow
){.
skey
=
state
->
lastTs
,
.
ekey
=
TSKEY_MAX
},
&
(
SVersionRange
){.
minVer
=
0
,
.
maxVer
=
UINT64_MAX
},
state
->
pLoadInfo
,
false
,
NULL
,
true
);
&
(
SVersionRange
){.
minVer
=
0
,
.
maxVer
=
UINT64_MAX
},
state
->
pLoadInfo
,
false
,
NULL
,
true
,
state
->
pDataIter
);
state
->
pMergeTree
=
&
state
->
mergeTree
;
state
->
pMergeTree
=
&
state
->
mergeTree
;
state
->
state
=
SFSLASTNEXTROW_BLOCKROW
;
state
->
state
=
SFSLASTNEXTROW_BLOCKROW
;
}
}
...
@@ -1211,7 +1212,7 @@ typedef struct {
...
@@ -1211,7 +1212,7 @@ typedef struct {
}
CacheNextRowIter
;
}
CacheNextRowIter
;
static
int32_t
nextRowIterOpen
(
CacheNextRowIter
*
pIter
,
tb_uid_t
uid
,
STsdb
*
pTsdb
,
STSchema
*
pTSchema
,
tb_uid_t
suid
,
static
int32_t
nextRowIterOpen
(
CacheNextRowIter
*
pIter
,
tb_uid_t
uid
,
STsdb
*
pTsdb
,
STSchema
*
pTSchema
,
tb_uid_t
suid
,
SSttBlockLoadInfo
*
pLoadInfo
,
STsdbReadSnap
*
pReadSnap
,
SDataFReader
**
pDataFReader
,
SSttBlockLoadInfo
*
pLoadInfo
,
S
LDataIter
*
pLDataIter
,
S
TsdbReadSnap
*
pReadSnap
,
SDataFReader
**
pDataFReader
,
SDataFReader
**
pDataFReaderLast
,
int64_t
lastTs
)
{
SDataFReader
**
pDataFReaderLast
,
int64_t
lastTs
)
{
int
code
=
0
;
int
code
=
0
;
...
@@ -1274,6 +1275,7 @@ static int32_t nextRowIterOpen(CacheNextRowIter *pIter, tb_uid_t uid, STsdb *pTs
...
@@ -1274,6 +1275,7 @@ static int32_t nextRowIterOpen(CacheNextRowIter *pIter, tb_uid_t uid, STsdb *pTs
pIter
->
fsLastState
.
pLoadInfo
=
pLoadInfo
;
pIter
->
fsLastState
.
pLoadInfo
=
pLoadInfo
;
pIter
->
fsLastState
.
pDataFReader
=
pDataFReaderLast
;
pIter
->
fsLastState
.
pDataFReader
=
pDataFReaderLast
;
pIter
->
fsLastState
.
lastTs
=
lastTs
;
pIter
->
fsLastState
.
lastTs
=
lastTs
;
pIter
->
fsLastState
.
pDataIter
=
pLDataIter
;
pIter
->
fsState
.
state
=
SFSNEXTROW_FS
;
pIter
->
fsState
.
state
=
SFSNEXTROW_FS
;
pIter
->
fsState
.
pTsdb
=
pTsdb
;
pIter
->
fsState
.
pTsdb
=
pTsdb
;
...
@@ -1465,7 +1467,7 @@ static int32_t mergeLastRow(tb_uid_t uid, STsdb *pTsdb, bool *dup, SArray **ppCo
...
@@ -1465,7 +1467,7 @@ static int32_t mergeLastRow(tb_uid_t uid, STsdb *pTsdb, bool *dup, SArray **ppCo
TSKEY
lastRowTs
=
TSKEY_MAX
;
TSKEY
lastRowTs
=
TSKEY_MAX
;
CacheNextRowIter
iter
=
{
0
};
CacheNextRowIter
iter
=
{
0
};
nextRowIterOpen
(
&
iter
,
uid
,
pTsdb
,
pTSchema
,
pr
->
suid
,
pr
->
pLoadInfo
,
pr
->
pReadSnap
,
&
pr
->
pDataFReader
,
nextRowIterOpen
(
&
iter
,
uid
,
pTsdb
,
pTSchema
,
pr
->
suid
,
pr
->
pLoadInfo
,
pr
->
p
DataIter
,
pr
->
p
ReadSnap
,
&
pr
->
pDataFReader
,
&
pr
->
pDataFReaderLast
,
pr
->
lastTs
);
&
pr
->
pDataFReaderLast
,
pr
->
lastTs
);
do
{
do
{
...
@@ -1622,7 +1624,7 @@ static int32_t mergeLast(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray, SCach
...
@@ -1622,7 +1624,7 @@ static int32_t mergeLast(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray, SCach
TSKEY
lastRowTs
=
TSKEY_MAX
;
TSKEY
lastRowTs
=
TSKEY_MAX
;
CacheNextRowIter
iter
=
{
0
};
CacheNextRowIter
iter
=
{
0
};
nextRowIterOpen
(
&
iter
,
uid
,
pTsdb
,
pTSchema
,
pr
->
suid
,
pr
->
pLoadInfo
,
pr
->
pReadSnap
,
&
pr
->
pDataFReader
,
nextRowIterOpen
(
&
iter
,
uid
,
pTsdb
,
pTSchema
,
pr
->
suid
,
pr
->
pLoadInfo
,
pr
->
p
DataIter
,
pr
->
p
ReadSnap
,
&
pr
->
pDataFReader
,
&
pr
->
pDataFReaderLast
,
pr
->
lastTs
);
&
pr
->
pDataFReaderLast
,
pr
->
lastTs
);
do
{
do
{
...
...
source/dnode/vnode/src/tsdb/tsdbCacheRead.c
浏览文件 @
5663550d
...
@@ -187,13 +187,21 @@ int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, void* pTableIdList,
...
@@ -187,13 +187,21 @@ int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, void* pTableIdList,
}
}
}
}
int32_t
numOfStt
=
((
SVnode
*
)
pVnode
)
->
config
.
sttTrigger
;
SVnodeCfg
*
pCfg
=
&
((
SVnode
*
)
pVnode
)
->
config
;
int32_t
numOfStt
=
pCfg
->
sttTrigger
;
p
->
pLoadInfo
=
tCreateLastBlockLoadInfo
(
p
->
pSchema
,
NULL
,
0
,
numOfStt
);
p
->
pLoadInfo
=
tCreateLastBlockLoadInfo
(
p
->
pSchema
,
NULL
,
0
,
numOfStt
);
if
(
p
->
pLoadInfo
==
NULL
)
{
if
(
p
->
pLoadInfo
==
NULL
)
{
tsdbCacherowsReaderClose
(
p
);
tsdbCacherowsReaderClose
(
p
);
return
TSDB_CODE_OUT_OF_MEMORY
;
return
TSDB_CODE_OUT_OF_MEMORY
;
}
}
p
->
pDataIter
=
taosMemoryCalloc
(
pCfg
->
sttTrigger
,
sizeof
(
SLDataIter
));
if
(
p
->
pDataIter
==
NULL
)
{
tsdbCacherowsReaderClose
(
p
);
return
TSDB_CODE_OUT_OF_MEMORY
;
}
p
->
idstr
=
taosStrdup
(
idstr
);
p
->
idstr
=
taosStrdup
(
idstr
);
taosThreadMutexInit
(
&
p
->
readerMutex
,
NULL
);
taosThreadMutexInit
(
&
p
->
readerMutex
,
NULL
);
...
@@ -215,6 +223,7 @@ void* tsdbCacherowsReaderClose(void* pReader) {
...
@@ -215,6 +223,7 @@ void* tsdbCacherowsReaderClose(void* pReader) {
taosMemoryFree
(
p
->
pSchema
);
taosMemoryFree
(
p
->
pSchema
);
}
}
taosMemoryFreeClear
(
p
->
pDataIter
);
taosMemoryFree
(
p
->
pCurrSchema
);
taosMemoryFree
(
p
->
pCurrSchema
);
destroyLastBlockLoadInfo
(
p
->
pLoadInfo
);
destroyLastBlockLoadInfo
(
p
->
pLoadInfo
);
...
...
source/dnode/vnode/src/tsdb/tsdbMergeTree.c
浏览文件 @
5663550d
...
@@ -16,22 +16,6 @@
...
@@ -16,22 +16,6 @@
#include "tsdb.h"
#include "tsdb.h"
// SLDataIter =================================================
// SLDataIter =================================================
struct
SLDataIter
{
SRBTreeNode
node
;
SSttBlk
*
pSttBlk
;
SDataFReader
*
pReader
;
int32_t
iStt
;
int8_t
backward
;
int32_t
iSttBlk
;
int32_t
iRow
;
SRowInfo
rInfo
;
uint64_t
uid
;
STimeWindow
timeWindow
;
SVersionRange
verRange
;
SSttBlockLoadInfo
*
pBlockLoadInfo
;
bool
ignoreEarlierTs
;
};
SSttBlockLoadInfo
*
tCreateLastBlockLoadInfo
(
STSchema
*
pSchema
,
int16_t
*
colList
,
int32_t
numOfCols
,
SSttBlockLoadInfo
*
tCreateLastBlockLoadInfo
(
STSchema
*
pSchema
,
int16_t
*
colList
,
int32_t
numOfCols
,
int32_t
numOfSttTrigger
)
{
int32_t
numOfSttTrigger
)
{
SSttBlockLoadInfo
*
pLoadInfo
=
taosMemoryCalloc
(
numOfSttTrigger
,
sizeof
(
SSttBlockLoadInfo
));
SSttBlockLoadInfo
*
pLoadInfo
=
taosMemoryCalloc
(
numOfSttTrigger
,
sizeof
(
SSttBlockLoadInfo
));
...
@@ -268,25 +252,19 @@ static int32_t binarySearchForStartRowIndex(uint64_t *uidList, int32_t num, uint
...
@@ -268,25 +252,19 @@ static int32_t binarySearchForStartRowIndex(uint64_t *uidList, int32_t num, uint
}
}
}
}
int32_t
tLDataIterOpen
(
struct
SLDataIter
*
*
pIter
,
SDataFReader
*
pReader
,
int32_t
iStt
,
int8_t
backward
,
uint64_t
suid
,
int32_t
tLDataIterOpen
(
struct
SLDataIter
*
pIter
,
SDataFReader
*
pReader
,
int32_t
iStt
,
int8_t
backward
,
uint64_t
suid
,
uint64_t
uid
,
STimeWindow
*
pTimeWindow
,
SVersionRange
*
pRange
,
SSttBlockLoadInfo
*
pBlockLoadInfo
,
uint64_t
uid
,
STimeWindow
*
pTimeWindow
,
SVersionRange
*
pRange
,
SSttBlockLoadInfo
*
pBlockLoadInfo
,
const
char
*
idStr
,
bool
strictTimeRange
)
{
const
char
*
idStr
,
bool
strictTimeRange
)
{
int32_t
code
=
TSDB_CODE_SUCCESS
;
int32_t
code
=
TSDB_CODE_SUCCESS
;
*
pIter
=
taosMemoryCalloc
(
1
,
sizeof
(
SLDataIter
));
pIter
->
uid
=
uid
;
if
(
*
pIter
==
NULL
)
{
pIter
->
pReader
=
pReader
;
code
=
TSDB_CODE_OUT_OF_MEMORY
;
pIter
->
iStt
=
iStt
;
goto
_exit
;
pIter
->
backward
=
backward
;
}
pIter
->
verRange
=
*
pRange
;
pIter
->
timeWindow
=
*
pTimeWindow
;
(
*
pIter
)
->
uid
=
uid
;
pIter
->
pBlockLoadInfo
=
pBlockLoadInfo
;
(
*
pIter
)
->
pReader
=
pReader
;
(
*
pIter
)
->
iStt
=
iStt
;
(
*
pIter
)
->
backward
=
backward
;
(
*
pIter
)
->
verRange
=
*
pRange
;
(
*
pIter
)
->
timeWindow
=
*
pTimeWindow
;
(
*
pIter
)
->
pBlockLoadInfo
=
pBlockLoadInfo
;
if
(
!
pBlockLoadInfo
->
sttBlockLoaded
)
{
if
(
!
pBlockLoadInfo
->
sttBlockLoaded
)
{
int64_t
st
=
taosGetTimestampUs
();
int64_t
st
=
taosGetTimestampUs
();
...
@@ -294,7 +272,7 @@ int32_t tLDataIterOpen(struct SLDataIter **pIter, SDataFReader *pReader, int32_t
...
@@ -294,7 +272,7 @@ int32_t tLDataIterOpen(struct SLDataIter **pIter, SDataFReader *pReader, int32_t
code
=
tsdbReadSttBlk
(
pReader
,
iStt
,
pBlockLoadInfo
->
aSttBlk
);
code
=
tsdbReadSttBlk
(
pReader
,
iStt
,
pBlockLoadInfo
->
aSttBlk
);
if
(
code
)
{
if
(
code
)
{
goto
_exit
;
return
code
;
}
}
// only apply to the child tables, ordinary tables will not incur this filter procedure.
// only apply to the child tables, ordinary tables will not incur this filter procedure.
...
@@ -310,7 +288,7 @@ int32_t tLDataIterOpen(struct SLDataIter **pIter, SDataFReader *pReader, int32_t
...
@@ -310,7 +288,7 @@ int32_t tLDataIterOpen(struct SLDataIter **pIter, SDataFReader *pReader, int32_t
// no qualified stt block existed
// no qualified stt block existed
taosArrayClear
(
pBlockLoadInfo
->
aSttBlk
);
taosArrayClear
(
pBlockLoadInfo
->
aSttBlk
);
(
*
pIter
)
->
iSttBlk
=
-
1
;
pIter
->
iSttBlk
=
-
1
;
double
el
=
(
taosGetTimestampUs
()
-
st
)
/
1000
.
0
;
double
el
=
(
taosGetTimestampUs
()
-
st
)
/
1000
.
0
;
tsdbDebug
(
"load the last file info completed, elapsed time:%.2fms, %s"
,
el
,
idStr
);
tsdbDebug
(
"load the last file info completed, elapsed time:%.2fms, %s"
,
el
,
idStr
);
return
code
;
return
code
;
...
@@ -343,31 +321,27 @@ int32_t tLDataIterOpen(struct SLDataIter **pIter, SDataFReader *pReader, int32_t
...
@@ -343,31 +321,27 @@ int32_t tLDataIterOpen(struct SLDataIter **pIter, SDataFReader *pReader, int32_t
size_t
size
=
taosArrayGetSize
(
pBlockLoadInfo
->
aSttBlk
);
size_t
size
=
taosArrayGetSize
(
pBlockLoadInfo
->
aSttBlk
);
// find the start block
// find the start block
(
*
pIter
)
->
iSttBlk
=
binarySearchForStartBlock
(
pBlockLoadInfo
->
aSttBlk
->
pData
,
size
,
uid
,
backward
);
pIter
->
iSttBlk
=
binarySearchForStartBlock
(
pBlockLoadInfo
->
aSttBlk
->
pData
,
size
,
uid
,
backward
);
if
(
(
*
pIter
)
->
iSttBlk
!=
-
1
)
{
if
(
pIter
->
iSttBlk
!=
-
1
)
{
(
*
pIter
)
->
pSttBlk
=
taosArrayGet
(
pBlockLoadInfo
->
aSttBlk
,
(
*
pIter
)
->
iSttBlk
);
pIter
->
pSttBlk
=
taosArrayGet
(
pBlockLoadInfo
->
aSttBlk
,
pIter
->
iSttBlk
);
(
*
pIter
)
->
iRow
=
((
*
pIter
)
->
backward
)
?
(
*
pIter
)
->
pSttBlk
->
nRow
:
-
1
;
pIter
->
iRow
=
(
pIter
->
backward
)
?
pIter
->
pSttBlk
->
nRow
:
-
1
;
if
((
!
backward
)
&&
((
strictTimeRange
&&
(
*
pIter
)
->
pSttBlk
->
minKey
>=
(
*
pIter
)
->
timeWindow
.
ekey
)
||
if
((
!
backward
)
&&
((
strictTimeRange
&&
pIter
->
pSttBlk
->
minKey
>=
pIter
->
timeWindow
.
ekey
)
||
(
!
strictTimeRange
&&
(
*
pIter
)
->
pSttBlk
->
minKey
>
(
*
pIter
)
->
timeWindow
.
ekey
)))
{
(
!
strictTimeRange
&&
pIter
->
pSttBlk
->
minKey
>
pIter
->
timeWindow
.
ekey
)))
{
(
*
pIter
)
->
pSttBlk
=
NULL
;
pIter
->
pSttBlk
=
NULL
;
}
}
if
(
backward
&&
((
strictTimeRange
&&
(
*
pIter
)
->
pSttBlk
->
maxKey
<=
(
*
pIter
)
->
timeWindow
.
skey
)
||
if
(
backward
&&
((
strictTimeRange
&&
pIter
->
pSttBlk
->
maxKey
<=
pIter
->
timeWindow
.
skey
)
||
(
!
strictTimeRange
&&
(
*
pIter
)
->
pSttBlk
->
maxKey
<
(
*
pIter
)
->
timeWindow
.
skey
)))
{
(
!
strictTimeRange
&&
pIter
->
pSttBlk
->
maxKey
<
pIter
->
timeWindow
.
skey
)))
{
(
*
pIter
)
->
pSttBlk
=
NULL
;
pIter
->
pSttBlk
=
NULL
;
(
*
pIter
)
->
ignoreEarlierTs
=
true
;
pIter
->
ignoreEarlierTs
=
true
;
}
}
}
}
return
code
;
return
code
;
_exit:
taosMemoryFree
(
*
pIter
);
return
code
;
}
}
void
tLDataIterClose
(
SLDataIter
*
pIter
)
{
taosMemoryFree
(
pIter
);
}
void
tLDataIterClose
(
SLDataIter
*
pIter
)
{
/*taosMemoryFree(pIter); */
}
void
tLDataIterNextBlock
(
SLDataIter
*
pIter
,
const
char
*
idStr
)
{
void
tLDataIterNextBlock
(
SLDataIter
*
pIter
,
const
char
*
idStr
)
{
int32_t
step
=
pIter
->
backward
?
-
1
:
1
;
int32_t
step
=
pIter
->
backward
?
-
1
:
1
;
...
@@ -594,43 +568,38 @@ static FORCE_INLINE int32_t tLDataIterDescCmprFn(const SRBTreeNode *p1, const SR
...
@@ -594,43 +568,38 @@ static FORCE_INLINE int32_t tLDataIterDescCmprFn(const SRBTreeNode *p1, const SR
int32_t
tMergeTreeOpen
(
SMergeTree
*
pMTree
,
int8_t
backward
,
SDataFReader
*
pFReader
,
uint64_t
suid
,
uint64_t
uid
,
int32_t
tMergeTreeOpen
(
SMergeTree
*
pMTree
,
int8_t
backward
,
SDataFReader
*
pFReader
,
uint64_t
suid
,
uint64_t
uid
,
STimeWindow
*
pTimeWindow
,
SVersionRange
*
pVerRange
,
SSttBlockLoadInfo
*
pBlockLoadInfo
,
STimeWindow
*
pTimeWindow
,
SVersionRange
*
pVerRange
,
SSttBlockLoadInfo
*
pBlockLoadInfo
,
bool
destroyLoadInfo
,
const
char
*
idStr
,
bool
strictTimeRange
)
{
bool
destroyLoadInfo
,
const
char
*
idStr
,
bool
strictTimeRange
,
SLDataIter
*
pLDataIter
)
{
int32_t
code
=
TSDB_CODE_SUCCESS
;
pMTree
->
backward
=
backward
;
pMTree
->
backward
=
backward
;
pMTree
->
pIter
=
NULL
;
pMTree
->
pIter
=
NULL
;
pMTree
->
pIterList
=
taosArrayInit
(
4
,
POINTER_BYTES
);
if
(
pMTree
->
pIterList
==
NULL
)
{
return
TSDB_CODE_OUT_OF_MEMORY
;
}
pMTree
->
idStr
=
idStr
;
pMTree
->
idStr
=
idStr
;
if
(
!
pMTree
->
backward
)
{
// asc
if
(
!
pMTree
->
backward
)
{
// asc
tRBTreeCreate
(
&
pMTree
->
rbt
,
tLDataIterCmprFn
);
tRBTreeCreate
(
&
pMTree
->
rbt
,
tLDataIterCmprFn
);
}
else
{
// desc
}
else
{
// desc
tRBTreeCreate
(
&
pMTree
->
rbt
,
tLDataIterDescCmprFn
);
tRBTreeCreate
(
&
pMTree
->
rbt
,
tLDataIterDescCmprFn
);
}
}
int32_t
code
=
TSDB_CODE_SUCCESS
;
pMTree
->
pLoadInfo
=
pBlockLoadInfo
;
pMTree
->
pLoadInfo
=
pBlockLoadInfo
;
pMTree
->
destroyLoadInfo
=
destroyLoadInfo
;
pMTree
->
destroyLoadInfo
=
destroyLoadInfo
;
pMTree
->
ignoreEarlierTs
=
false
;
pMTree
->
ignoreEarlierTs
=
false
;
for
(
int32_t
i
=
0
;
i
<
pFReader
->
pSet
->
nSttF
;
++
i
)
{
// open all last file
for
(
int32_t
i
=
0
;
i
<
pFReader
->
pSet
->
nSttF
;
++
i
)
{
// open all last file
struct
SLDataIter
*
pIter
=
NULL
;
memset
(
&
pLDataIter
[
i
],
0
,
sizeof
(
SLDataIter
))
;
code
=
tLDataIterOpen
(
&
p
Iter
,
pFReader
,
i
,
pMTree
->
backward
,
suid
,
uid
,
pTimeWindow
,
pVerRange
,
code
=
tLDataIterOpen
(
&
p
LDataIter
[
i
]
,
pFReader
,
i
,
pMTree
->
backward
,
suid
,
uid
,
pTimeWindow
,
pVerRange
,
&
pMTree
->
pLoadInfo
[
i
],
pMTree
->
idStr
,
strictTimeRange
);
&
pMTree
->
pLoadInfo
[
i
],
pMTree
->
idStr
,
strictTimeRange
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
goto
_end
;
goto
_end
;
}
}
bool
hasVal
=
tLDataIterNextRow
(
pIter
,
pMTree
->
idStr
);
bool
hasVal
=
tLDataIterNextRow
(
&
pLDataIter
[
i
]
,
pMTree
->
idStr
);
if
(
hasVal
)
{
if
(
hasVal
)
{
taosArrayPush
(
pMTree
->
pIterList
,
&
pIter
);
tMergeTreeAddIter
(
pMTree
,
&
pLDataIter
[
i
]);
tMergeTreeAddIter
(
pMTree
,
pIter
);
}
else
{
}
else
{
if
(
!
pMTree
->
ignoreEarlierTs
)
{
if
(
!
pMTree
->
ignoreEarlierTs
)
{
pMTree
->
ignoreEarlierTs
=
p
Iter
->
ignoreEarlierTs
;
pMTree
->
ignoreEarlierTs
=
p
LDataIter
[
i
].
ignoreEarlierTs
;
}
}
tLDataIterClose
(
pIter
);
}
}
}
}
...
@@ -681,15 +650,7 @@ bool tMergeTreeNext(SMergeTree *pMTree) {
...
@@ -681,15 +650,7 @@ bool tMergeTreeNext(SMergeTree *pMTree) {
TSDBROW
tMergeTreeGetRow
(
SMergeTree
*
pMTree
)
{
return
pMTree
->
pIter
->
rInfo
.
row
;
}
TSDBROW
tMergeTreeGetRow
(
SMergeTree
*
pMTree
)
{
return
pMTree
->
pIter
->
rInfo
.
row
;
}
void
tMergeTreeClose
(
SMergeTree
*
pMTree
)
{
void
tMergeTreeClose
(
SMergeTree
*
pMTree
)
{
size_t
size
=
taosArrayGetSize
(
pMTree
->
pIterList
);
for
(
int32_t
i
=
0
;
i
<
size
;
++
i
)
{
SLDataIter
*
pIter
=
taosArrayGetP
(
pMTree
->
pIterList
,
i
);
tLDataIterClose
(
pIter
);
}
pMTree
->
pIterList
=
taosArrayDestroy
(
pMTree
->
pIterList
);
pMTree
->
pIter
=
NULL
;
pMTree
->
pIter
=
NULL
;
if
(
pMTree
->
destroyLoadInfo
)
{
if
(
pMTree
->
destroyLoadInfo
)
{
pMTree
->
pLoadInfo
=
destroyLastBlockLoadInfo
(
pMTree
->
pLoadInfo
);
pMTree
->
pLoadInfo
=
destroyLastBlockLoadInfo
(
pMTree
->
pLoadInfo
);
pMTree
->
destroyLoadInfo
=
false
;
pMTree
->
destroyLoadInfo
=
false
;
...
...
source/dnode/vnode/src/tsdb/tsdbRead.c
浏览文件 @
5663550d
...
@@ -156,6 +156,7 @@ typedef struct SReaderStatus {
...
@@ -156,6 +156,7 @@ typedef struct SReaderStatus {
SBlockData
fileBlockData
;
SBlockData
fileBlockData
;
SFilesetIter
fileIter
;
SFilesetIter
fileIter
;
SDataBlockIter
blockIter
;
SDataBlockIter
blockIter
;
SLDataIter
*
pLDataIter
;
}
SReaderStatus
;
}
SReaderStatus
;
typedef
struct
SBlockInfoBuf
{
typedef
struct
SBlockInfoBuf
{
...
@@ -185,7 +186,6 @@ struct STsdbReader {
...
@@ -185,7 +186,6 @@ struct STsdbReader {
STsdbReadSnap
*
pReadSnap
;
STsdbReadSnap
*
pReadSnap
;
SIOCostSummary
cost
;
SIOCostSummary
cost
;
STSchema
*
pSchema
;
// the newest version schema
STSchema
*
pSchema
;
// the newest version schema
// STSchema* pMemSchema; // the previous schema for in-memory data, to avoid load schema too many times
SSHashObj
*
pSchemaMap
;
// keep the retrieved schema info, to avoid the overhead by repeatly load schema
SSHashObj
*
pSchemaMap
;
// keep the retrieved schema info, to avoid the overhead by repeatly load schema
SDataFReader
*
pFileReader
;
// the file reader
SDataFReader
*
pFileReader
;
// the file reader
SDelFReader
*
pDelFReader
;
// the del file reader
SDelFReader
*
pDelFReader
;
// the del file reader
...
@@ -741,6 +741,7 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, STsd
...
@@ -741,6 +741,7 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, STsd
pReader
->
type
=
pCond
->
type
;
pReader
->
type
=
pCond
->
type
;
pReader
->
window
=
updateQueryTimeWindow
(
pReader
->
pTsdb
,
&
pCond
->
twindows
);
pReader
->
window
=
updateQueryTimeWindow
(
pReader
->
pTsdb
,
&
pCond
->
twindows
);
pReader
->
blockInfoBuf
.
numPerBucket
=
1000
;
// 1000 tables per bucket
pReader
->
blockInfoBuf
.
numPerBucket
=
1000
;
// 1000 tables per bucket
pReader
->
status
.
pLDataIter
=
taosMemoryCalloc
(
pVnode
->
config
.
sttTrigger
,
sizeof
(
SLDataIter
));
if
(
pReader
->
pResBlock
==
NULL
)
{
if
(
pReader
->
pResBlock
==
NULL
)
{
pReader
->
freeBlock
=
true
;
pReader
->
freeBlock
=
true
;
...
@@ -2547,7 +2548,7 @@ static bool initLastBlockReader(SLastBlockReader* pLBlockReader, STableBlockScan
...
@@ -2547,7 +2548,7 @@ static bool initLastBlockReader(SLastBlockReader* pLBlockReader, STableBlockScan
pScanInfo
->
uid
,
pReader
->
idStr
);
pScanInfo
->
uid
,
pReader
->
idStr
);
int32_t
code
=
tMergeTreeOpen
(
&
pLBlockReader
->
mergeTree
,
(
pLBlockReader
->
order
==
TSDB_ORDER_DESC
),
int32_t
code
=
tMergeTreeOpen
(
&
pLBlockReader
->
mergeTree
,
(
pLBlockReader
->
order
==
TSDB_ORDER_DESC
),
pReader
->
pFileReader
,
pReader
->
suid
,
pScanInfo
->
uid
,
&
w
,
&
pLBlockReader
->
verRange
,
pReader
->
pFileReader
,
pReader
->
suid
,
pScanInfo
->
uid
,
&
w
,
&
pLBlockReader
->
verRange
,
pLBlockReader
->
pInfo
,
false
,
pReader
->
idStr
,
false
);
pLBlockReader
->
pInfo
,
false
,
pReader
->
idStr
,
false
,
pReader
->
status
.
pLDataIter
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
return
false
;
return
false
;
}
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录