Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
56ed56ca
TDengine
项目概览
taosdata
/
TDengine
1 年多 前同步成功
通知
1185
Star
22016
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看板
提交
56ed56ca
编写于
2月 04, 2020
作者:
H
hjxilinx
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
refactor some codes
上级
9c1db2ef
变更
1
显示空白变更内容
内联
并排
Showing
1 changed file
with
82 addition
and
90 deletion
+82
-90
src/system/detail/src/vnodeQueryImpl.c
src/system/detail/src/vnodeQueryImpl.c
+82
-90
未找到文件。
src/system/detail/src/vnodeQueryImpl.c
浏览文件 @
56ed56ca
...
@@ -49,8 +49,6 @@ enum {
...
@@ -49,8 +49,6 @@ enum {
#define IS_DISK_DATA_BLOCK(q) ((q)->fileId >= 0)
#define IS_DISK_DATA_BLOCK(q) ((q)->fileId >= 0)
// static int32_t copyDataFromMMapBuffer(int fd, SQInfo *pQInfo, SQueryFilesInfo *pQueryFile, char *buf, uint64_t
// offset, int32_t size);
static
int32_t
readDataFromDiskFile
(
int
fd
,
SQInfo
*
pQInfo
,
SQueryFilesInfo
*
pQueryFile
,
char
*
buf
,
uint64_t
offset
,
static
int32_t
readDataFromDiskFile
(
int
fd
,
SQInfo
*
pQInfo
,
SQueryFilesInfo
*
pQueryFile
,
char
*
buf
,
uint64_t
offset
,
int32_t
size
);
int32_t
size
);
...
@@ -70,12 +68,12 @@ static int32_t getNextDataFileCompInfo(SQueryRuntimeEnv *pRuntimeEnv, SMeterObj
...
@@ -70,12 +68,12 @@ static int32_t getNextDataFileCompInfo(SQueryRuntimeEnv *pRuntimeEnv, SMeterObj
static
void
setGroupOutputBuffer
(
SQueryRuntimeEnv
*
pRuntimeEnv
,
SOutputRes
*
pResult
);
static
void
setGroupOutputBuffer
(
SQueryRuntimeEnv
*
pRuntimeEnv
,
SOutputRes
*
pResult
);
static
void
getAlignedIntervalQueryRange
(
SQueryRuntimeEnv
*
pRuntimeEnv
,
TSKEY
keyInData
,
TSKEY
skey
,
TSKEY
ekey
);
static
void
getAlignedIntervalQueryRange
(
SQueryRuntimeEnv
*
pRuntimeEnv
,
TSKEY
keyInData
,
TSKEY
skey
,
TSKEY
ekey
);
static
void
doApplyIntervalQueryOnBlock
(
SMeterQuerySupportObj
*
pSupporter
,
SMeterQueryInfo
*
pInfo
,
static
void
doApplyIntervalQueryOnBlock
(
SMeterQuerySupportObj
*
pSupporter
,
SMeterQueryInfo
*
p
MeterQuery
Info
,
SBlockInfo
*
pBlockInfo
,
int64_t
*
pPrimaryCol
,
SField
*
pFields
,
SBlockInfo
*
pBlockInfo
,
int64_t
*
pPrimaryCol
,
SField
*
pFields
,
__block_search_fn_t
searchFn
);
__block_search_fn_t
searchFn
);
static
int32_t
saveResult
(
SMeterQuerySupportObj
*
pSupporter
,
SMeterQueryInfo
*
pMeterQueryInfo
,
int32_t
numOfResult
);
static
int32_t
saveResult
(
SMeterQuerySupportObj
*
pSupporter
,
SMeterQueryInfo
*
pMeterQueryInfo
,
int32_t
numOfResult
);
static
void
applyIntervalQueryOnBlock
(
SMeterQuerySupportObj
*
pSupporter
,
SMeterDataInfo
*
p
InfoEx
,
int64_t
*
pPrimaryData
,
static
void
applyIntervalQueryOnBlock
(
SMeterQuerySupportObj
*
pSupporter
,
SMeterDataInfo
*
p
MeterDataInfo
,
SBlockInfo
*
pBlockInfo
,
int32_t
blockStatus
,
SField
*
pFields
,
SBlockInfo
*
pBlockInfo
,
int32_t
blockStatus
,
SField
*
pFields
,
__block_search_fn_t
searchFn
);
__block_search_fn_t
searchFn
);
...
@@ -1728,6 +1726,8 @@ int32_t numOfClosedSlidingWindow(SSlidingWindowInfo *pSlidingWindowInfo) {
...
@@ -1728,6 +1726,8 @@ int32_t numOfClosedSlidingWindow(SSlidingWindowInfo *pSlidingWindowInfo) {
return
i
;
return
i
;
}
}
}
}
return
0
;
}
}
void
closeSlidingWindow
(
SSlidingWindowInfo
*
pSlidingWindowInfo
,
int32_t
slot
)
{
void
closeSlidingWindow
(
SSlidingWindowInfo
*
pSlidingWindowInfo
,
int32_t
slot
)
{
...
@@ -2112,7 +2112,7 @@ static int32_t rowwiseApplyAllFunctions(SQueryRuntimeEnv *pRuntimeEnv, int32_t *
...
@@ -2112,7 +2112,7 @@ static int32_t rowwiseApplyAllFunctions(SQueryRuntimeEnv *pRuntimeEnv, int32_t *
SSlidingWindowInfo
*
pSlidingWindowInfo
=
&
pRuntimeEnv
->
swindowResInfo
;
SSlidingWindowInfo
*
pSlidingWindowInfo
=
&
pRuntimeEnv
->
swindowResInfo
;
// query completed
// query completed
if
(
lastKey
>=
pQuery
->
ekey
&&
QUERY_IS_ASC_QUERY
(
pQuery
)
||
if
(
(
lastKey
>=
pQuery
->
ekey
&&
QUERY_IS_ASC_QUERY
(
pQuery
)
)
||
(
lastKey
<=
pQuery
->
ekey
&&
!
QUERY_IS_ASC_QUERY
(
pQuery
)))
{
(
lastKey
<=
pQuery
->
ekey
&&
!
QUERY_IS_ASC_QUERY
(
pQuery
)))
{
closeAllSlidingWindow
(
pSlidingWindowInfo
);
closeAllSlidingWindow
(
pSlidingWindowInfo
);
...
@@ -5027,7 +5027,7 @@ static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
...
@@ -5027,7 +5027,7 @@ static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
static
void
updatelastkey
(
SQuery
*
pQuery
,
SMeterQueryInfo
*
pMeterQInfo
)
{
pMeterQInfo
->
lastKey
=
pQuery
->
lastKey
;
}
static
void
updatelastkey
(
SQuery
*
pQuery
,
SMeterQueryInfo
*
pMeterQInfo
)
{
pMeterQInfo
->
lastKey
=
pQuery
->
lastKey
;
}
void
queryOnBlock
(
SMeterQuerySupportObj
*
pSupporter
,
int64_t
*
primaryKeys
,
int32_t
blockStatus
,
void
queryOnBlock
(
SMeterQuerySupportObj
*
pSupporter
,
int64_t
*
primaryKeys
,
int32_t
blockStatus
,
SBlockInfo
*
pBlockBasicInfo
,
SMeterDataInfo
*
p
DataHeadInfoEx
,
SField
*
pFields
,
SBlockInfo
*
pBlockBasicInfo
,
SMeterDataInfo
*
p
MeterDataInfo
,
SField
*
pFields
,
__block_search_fn_t
searchFn
)
{
__block_search_fn_t
searchFn
)
{
/* cache blocks may be assign to other meter, abort */
/* cache blocks may be assign to other meter, abort */
if
(
pBlockBasicInfo
->
size
<=
0
)
{
if
(
pBlockBasicInfo
->
size
<=
0
)
{
...
@@ -5043,18 +5043,17 @@ void queryOnBlock(SMeterQuerySupportObj *pSupporter, int64_t *primaryKeys, int32
...
@@ -5043,18 +5043,17 @@ void queryOnBlock(SMeterQuerySupportObj *pSupporter, int64_t *primaryKeys, int32
// note: only fixed number of output for each group by operation
// note: only fixed number of output for each group by operation
if
(
numOfRes
>
0
)
{
if
(
numOfRes
>
0
)
{
pSupporter
->
pResult
[
p
DataHeadInfoEx
->
groupIdx
].
numOfRows
=
numOfRes
;
pSupporter
->
pResult
[
p
MeterDataInfo
->
groupIdx
].
numOfRows
=
numOfRes
;
}
}
// used to decide the correct start position in cache after check all data in files
// used to decide the correct start position in cache after check all data in files
updatelastkey
(
pQuery
,
p
DataHeadInfoEx
->
pMeterQInfo
);
updatelastkey
(
pQuery
,
p
MeterDataInfo
->
pMeterQInfo
);
if
(
pRuntimeEnv
->
pTSBuf
!=
NULL
)
{
if
(
pRuntimeEnv
->
pTSBuf
!=
NULL
)
{
p
DataHeadInfoEx
->
pMeterQInfo
->
cur
=
tsBufGetCursor
(
pRuntimeEnv
->
pTSBuf
);
p
MeterDataInfo
->
pMeterQInfo
->
cur
=
tsBufGetCursor
(
pRuntimeEnv
->
pTSBuf
);
}
}
}
else
{
}
else
{
applyIntervalQueryOnBlock
(
pSupporter
,
pDataHeadInfoEx
,
primaryKeys
,
pBlockBasicInfo
,
blockStatus
,
pFields
,
applyIntervalQueryOnBlock
(
pSupporter
,
pMeterDataInfo
,
pBlockBasicInfo
,
blockStatus
,
pFields
,
searchFn
);
searchFn
);
}
}
}
}
...
@@ -5265,13 +5264,13 @@ static tFilePage *getFilePage(SMeterQuerySupportObj *pSupporter, int32_t pageId)
...
@@ -5265,13 +5264,13 @@ static tFilePage *getFilePage(SMeterQuerySupportObj *pSupporter, int32_t pageId)
return
(
tFilePage
*
)(
pSupporter
->
meterOutputMMapBuf
+
DEFAULT_INTERN_BUF_SIZE
*
pageId
);
return
(
tFilePage
*
)(
pSupporter
->
meterOutputMMapBuf
+
DEFAULT_INTERN_BUF_SIZE
*
pageId
);
}
}
static
tFilePage
*
getMeterDataPage
(
SMeterQuerySupportObj
*
pSupporter
,
SMeterDataInfo
*
p
InfoEx
,
int32_t
pageId
)
{
static
tFilePage
*
getMeterDataPage
(
SMeterQuerySupportObj
*
pSupporter
,
SMeterDataInfo
*
p
MeterDataInfo
,
int32_t
pageId
)
{
SMeterQueryInfo
*
p
Info
=
pInfoEx
->
pMeterQInfo
;
SMeterQueryInfo
*
p
MeterQueryInfo
=
pMeterDataInfo
->
pMeterQInfo
;
if
(
pageId
>=
pInfo
->
numOfPages
)
{
if
(
pageId
>=
p
MeterQuery
Info
->
numOfPages
)
{
return
NULL
;
return
NULL
;
}
}
int32_t
realId
=
pInfo
->
pageList
[
pageId
];
int32_t
realId
=
p
MeterQuery
Info
->
pageList
[
pageId
];
return
getFilePage
(
pSupporter
,
realId
);
return
getFilePage
(
pSupporter
,
realId
);
}
}
...
@@ -5281,14 +5280,14 @@ typedef struct Position {
...
@@ -5281,14 +5280,14 @@ typedef struct Position {
}
Position
;
}
Position
;
typedef
struct
SCompSupporter
{
typedef
struct
SCompSupporter
{
SMeterDataInfo
**
p
InfoEx
;
SMeterDataInfo
**
p
MeterDataInfo
;
Position
*
pPosition
;
Position
*
pPosition
;
SMeterQuerySupportObj
*
pSupporter
;
SMeterQuerySupportObj
*
pSupporter
;
}
SCompSupporter
;
}
SCompSupporter
;
int64_t
getCurrentTimestamp
(
SCompSupporter
*
pSupportor
,
int32_t
meterIdx
)
{
int64_t
getCurrentTimestamp
(
SCompSupporter
*
pSupportor
,
int32_t
meterIdx
)
{
Position
*
pPos
=
&
pSupportor
->
pPosition
[
meterIdx
];
Position
*
pPos
=
&
pSupportor
->
pPosition
[
meterIdx
];
tFilePage
*
pPage
=
getMeterDataPage
(
pSupportor
->
pSupporter
,
pSupportor
->
p
InfoEx
[
meterIdx
],
pPos
->
pageIdx
);
tFilePage
*
pPage
=
getMeterDataPage
(
pSupportor
->
pSupporter
,
pSupportor
->
p
MeterDataInfo
[
meterIdx
],
pPos
->
pageIdx
);
return
*
(
int64_t
*
)(
pPage
->
data
+
TSDB_KEYSIZE
*
pPos
->
rowIdx
);
return
*
(
int64_t
*
)(
pPage
->
data
+
TSDB_KEYSIZE
*
pPos
->
rowIdx
);
}
}
...
@@ -5311,10 +5310,10 @@ int32_t meterResultComparator(const void *pLeft, const void *pRight, void *param
...
@@ -5311,10 +5310,10 @@ int32_t meterResultComparator(const void *pLeft, const void *pRight, void *param
return
-
1
;
return
-
1
;
}
}
tFilePage
*
pPageLeft
=
getMeterDataPage
(
supportor
->
pSupporter
,
supportor
->
p
InfoEx
[
left
],
leftPos
.
pageIdx
);
tFilePage
*
pPageLeft
=
getMeterDataPage
(
supportor
->
pSupporter
,
supportor
->
p
MeterDataInfo
[
left
],
leftPos
.
pageIdx
);
int64_t
leftTimestamp
=
*
(
int64_t
*
)(
pPageLeft
->
data
+
TSDB_KEYSIZE
*
leftPos
.
rowIdx
);
int64_t
leftTimestamp
=
*
(
int64_t
*
)(
pPageLeft
->
data
+
TSDB_KEYSIZE
*
leftPos
.
rowIdx
);
tFilePage
*
pPageRight
=
getMeterDataPage
(
supportor
->
pSupporter
,
supportor
->
p
InfoEx
[
right
],
rightPos
.
pageIdx
);
tFilePage
*
pPageRight
=
getMeterDataPage
(
supportor
->
pSupporter
,
supportor
->
p
MeterDataInfo
[
right
],
rightPos
.
pageIdx
);
int64_t
rightTimestamp
=
*
(
int64_t
*
)(
pPageRight
->
data
+
TSDB_KEYSIZE
*
rightPos
.
rowIdx
);
int64_t
rightTimestamp
=
*
(
int64_t
*
)(
pPageRight
->
data
+
TSDB_KEYSIZE
*
rightPos
.
rowIdx
);
if
(
leftTimestamp
==
rightTimestamp
)
{
if
(
leftTimestamp
==
rightTimestamp
)
{
...
@@ -5465,17 +5464,17 @@ int32_t doMergeMetersResultsToGroupRes(SMeterQuerySupportObj *pSupporter, SQuery
...
@@ -5465,17 +5464,17 @@ int32_t doMergeMetersResultsToGroupRes(SMeterQuerySupportObj *pSupporter, SQuery
cs
.
pPosition
[
pos
].
pageIdx
+=
1
;
// try next page
cs
.
pPosition
[
pos
].
pageIdx
+=
1
;
// try next page
// check if current page is empty or not. if it is empty, ignore it and try next
// check if current page is empty or not. if it is empty, ignore it and try next
if
(
cs
.
pPosition
[
pos
].
pageIdx
<=
cs
.
p
InfoEx
[
pos
]
->
pMeterQInfo
->
numOfPages
-
1
)
{
if
(
cs
.
pPosition
[
pos
].
pageIdx
<=
cs
.
p
MeterDataInfo
[
pos
]
->
pMeterQInfo
->
numOfPages
-
1
)
{
tFilePage
*
newPage
=
getMeterDataPage
(
cs
.
pSupporter
,
pValidMeter
[
pos
],
position
->
pageIdx
);
tFilePage
*
newPage
=
getMeterDataPage
(
cs
.
pSupporter
,
pValidMeter
[
pos
],
position
->
pageIdx
);
if
(
newPage
->
numOfElems
<=
0
)
{
if
(
newPage
->
numOfElems
<=
0
)
{
// if current source data page is null, it must be the last page of source output page
// if current source data page is null, it must be the last page of source output page
cs
.
pPosition
[
pos
].
pageIdx
+=
1
;
cs
.
pPosition
[
pos
].
pageIdx
+=
1
;
assert
(
cs
.
pPosition
[
pos
].
pageIdx
>=
cs
.
p
InfoEx
[
pos
]
->
pMeterQInfo
->
numOfPages
-
1
);
assert
(
cs
.
pPosition
[
pos
].
pageIdx
>=
cs
.
p
MeterDataInfo
[
pos
]
->
pMeterQInfo
->
numOfPages
-
1
);
}
}
}
}
// the following code must be executed if current source pages are exhausted
// the following code must be executed if current source pages are exhausted
if
(
cs
.
pPosition
[
pos
].
pageIdx
>=
cs
.
p
InfoEx
[
pos
]
->
pMeterQInfo
->
numOfPages
)
{
if
(
cs
.
pPosition
[
pos
].
pageIdx
>=
cs
.
p
MeterDataInfo
[
pos
]
->
pMeterQInfo
->
numOfPages
)
{
cs
.
pPosition
[
pos
].
pageIdx
=
-
1
;
cs
.
pPosition
[
pos
].
pageIdx
=
-
1
;
cs
.
pPosition
[
pos
].
rowIdx
=
-
1
;
cs
.
pPosition
[
pos
].
rowIdx
=
-
1
;
...
@@ -5635,7 +5634,7 @@ int32_t doCloseAllOpenedResults(SMeterQuerySupportObj *pSupporter) {
...
@@ -5635,7 +5634,7 @@ int32_t doCloseAllOpenedResults(SMeterQuerySupportObj *pSupporter) {
void
disableFunctForSuppleScan
(
SQueryRuntimeEnv
*
pRuntimeEnv
,
int32_t
order
)
{
void
disableFunctForSuppleScan
(
SQueryRuntimeEnv
*
pRuntimeEnv
,
int32_t
order
)
{
SQuery
*
pQuery
=
pRuntimeEnv
->
pQuery
;
SQuery
*
pQuery
=
pRuntimeEnv
->
pQuery
;
if
(
isGroupbyNormalCol
(
pQuery
->
pGroupbyExpr
)
||
pQuery
->
slidingTime
>
0
&&
pQuery
->
nAggTimeInterval
>
0
)
{
if
(
isGroupbyNormalCol
(
pQuery
->
pGroupbyExpr
)
||
(
pQuery
->
slidingTime
>
0
&&
pQuery
->
nAggTimeInterval
>
0
)
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutputCols
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutputCols
;
++
i
)
{
pRuntimeEnv
->
pCtx
[
i
].
order
=
(
pRuntimeEnv
->
pCtx
[
i
].
order
)
^
1
;
pRuntimeEnv
->
pCtx
[
i
].
order
=
(
pRuntimeEnv
->
pCtx
[
i
].
order
)
^
1
;
}
}
...
@@ -6693,11 +6692,11 @@ void cleanBlockOrderSupporter(SBlockOrderSupporter *pSupporter, int32_t numOfTab
...
@@ -6693,11 +6692,11 @@ void cleanBlockOrderSupporter(SBlockOrderSupporter *pSupporter, int32_t numOfTab
int32_t
createDataBlocksInfoEx
(
SMeterDataInfo
**
pMeterDataInfo
,
int32_t
numOfMeters
,
int32_t
createDataBlocksInfoEx
(
SMeterDataInfo
**
pMeterDataInfo
,
int32_t
numOfMeters
,
SMeterDataBlockInfoEx
**
pDataBlockInfoEx
,
int32_t
numOfCompBlocks
,
SMeterDataBlockInfoEx
**
pDataBlockInfoEx
,
int32_t
numOfCompBlocks
,
int32_t
*
n
AllocBlocksInfoSize
,
int64_t
addr
)
{
int32_t
*
n
umOfAllocBlocks
,
int64_t
addr
)
{
// release allocated memory first
// release allocated memory first
freeDataBlockFieldInfo
(
*
pDataBlockInfoEx
,
*
n
AllocBlocksInfoSize
);
freeDataBlockFieldInfo
(
*
pDataBlockInfoEx
,
*
n
umOfAllocBlocks
);
if
(
*
n
AllocBlocksInfoSize
==
0
||
*
nAllocBlocksInfoSize
<
numOfCompBlocks
)
{
if
(
*
n
umOfAllocBlocks
==
0
||
*
numOfAllocBlocks
<
numOfCompBlocks
)
{
char
*
tmp
=
realloc
((
*
pDataBlockInfoEx
),
sizeof
(
SMeterDataBlockInfoEx
)
*
numOfCompBlocks
);
char
*
tmp
=
realloc
((
*
pDataBlockInfoEx
),
sizeof
(
SMeterDataBlockInfoEx
)
*
numOfCompBlocks
);
if
(
tmp
==
NULL
)
{
if
(
tmp
==
NULL
)
{
tfree
(
*
pDataBlockInfoEx
);
tfree
(
*
pDataBlockInfoEx
);
...
@@ -6706,7 +6705,7 @@ int32_t createDataBlocksInfoEx(SMeterDataInfo **pMeterDataInfo, int32_t numOfMet
...
@@ -6706,7 +6705,7 @@ int32_t createDataBlocksInfoEx(SMeterDataInfo **pMeterDataInfo, int32_t numOfMet
*
pDataBlockInfoEx
=
(
SMeterDataBlockInfoEx
*
)
tmp
;
*
pDataBlockInfoEx
=
(
SMeterDataBlockInfoEx
*
)
tmp
;
memset
((
*
pDataBlockInfoEx
),
0
,
sizeof
(
SMeterDataBlockInfoEx
)
*
numOfCompBlocks
);
memset
((
*
pDataBlockInfoEx
),
0
,
sizeof
(
SMeterDataBlockInfoEx
)
*
numOfCompBlocks
);
*
n
AllocBlocksInfoSize
=
numOfCompBlocks
;
*
n
umOfAllocBlocks
=
numOfCompBlocks
;
}
}
SBlockOrderSupporter
supporter
=
{
0
};
SBlockOrderSupporter
supporter
=
{
0
};
...
@@ -6740,14 +6739,14 @@ int32_t createDataBlocksInfoEx(SMeterDataInfo **pMeterDataInfo, int32_t numOfMet
...
@@ -6740,14 +6739,14 @@ int32_t createDataBlocksInfoEx(SMeterDataInfo **pMeterDataInfo, int32_t numOfMet
supporter
.
pDataBlockInfoEx
[
numOfQualMeters
]
=
(
SMeterDataBlockInfoEx
*
)
buf
;
supporter
.
pDataBlockInfoEx
[
numOfQualMeters
]
=
(
SMeterDataBlockInfoEx
*
)
buf
;
for
(
int32_t
k
=
0
;
k
<
pMeterDataInfo
[
j
]
->
numOfBlocks
;
++
k
)
{
for
(
int32_t
k
=
0
;
k
<
pMeterDataInfo
[
j
]
->
numOfBlocks
;
++
k
)
{
SMeterDataBlockInfoEx
*
pInfoEx
=
&
supporter
.
pDataBlockInfoEx
[
numOfQualMeters
][
k
];
SMeterDataBlockInfoEx
*
p
Block
InfoEx
=
&
supporter
.
pDataBlockInfoEx
[
numOfQualMeters
][
k
];
pInfoEx
->
pBlock
.
compBlock
=
&
pBlock
[
k
];
p
Block
InfoEx
->
pBlock
.
compBlock
=
&
pBlock
[
k
];
pInfoEx
->
pBlock
.
fields
=
NULL
;
p
Block
InfoEx
->
pBlock
.
fields
=
NULL
;
pInfoEx
->
pMeterDataInfo
=
pMeterDataInfo
[
j
];
p
Block
InfoEx
->
pMeterDataInfo
=
pMeterDataInfo
[
j
];
pInfoEx
->
groupIdx
=
pMeterDataInfo
[
j
]
->
groupIdx
;
// set the group index
p
Block
InfoEx
->
groupIdx
=
pMeterDataInfo
[
j
]
->
groupIdx
;
// set the group index
pInfoEx
->
blockIndex
=
pMeterDataInfo
[
j
]
->
start
+
k
;
// set the block index in original meter
p
Block
InfoEx
->
blockIndex
=
pMeterDataInfo
[
j
]
->
start
+
k
;
// set the block index in original meter
cnt
++
;
cnt
++
;
}
}
...
@@ -7013,7 +7012,7 @@ int32_t setIntervalQueryExecutionContext(SMeterQuerySupportObj *pSupporter, int3
...
@@ -7013,7 +7012,7 @@ int32_t setIntervalQueryExecutionContext(SMeterQuerySupportObj *pSupporter, int3
return
0
;
return
0
;
}
}
static
void
doApplyIntervalQueryOnBlock
(
SMeterQuerySupportObj
*
pSupporter
,
SMeterQueryInfo
*
pInfo
,
static
void
doApplyIntervalQueryOnBlock
(
SMeterQuerySupportObj
*
pSupporter
,
SMeterQueryInfo
*
p
MeterQuery
Info
,
SBlockInfo
*
pBlockInfo
,
int64_t
*
pPrimaryCol
,
SField
*
pFields
,
SBlockInfo
*
pBlockInfo
,
int64_t
*
pPrimaryCol
,
SField
*
pFields
,
__block_search_fn_t
searchFn
)
{
__block_search_fn_t
searchFn
)
{
SQueryRuntimeEnv
*
pRuntimeEnv
=
&
pSupporter
->
runtimeEnv
;
SQueryRuntimeEnv
*
pRuntimeEnv
=
&
pSupporter
->
runtimeEnv
;
...
@@ -7029,10 +7028,10 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
...
@@ -7029,10 +7028,10 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
assert
(
steps
>
0
);
assert
(
steps
>
0
);
// NOTE: in case of stable query, only ONE(or ZERO) row of result generated for each query range
// NOTE: in case of stable query, only ONE(or ZERO) row of result generated for each query range
if
(
pInfo
->
lastResRows
==
0
)
{
if
(
p
MeterQuery
Info
->
lastResRows
==
0
)
{
pInfo
->
lastResRows
=
numOfRes
;
p
MeterQuery
Info
->
lastResRows
=
numOfRes
;
}
else
{
}
else
{
assert
(
pInfo
->
lastResRows
==
1
);
assert
(
p
MeterQuery
Info
->
lastResRows
==
1
);
}
}
int32_t
pos
=
pQuery
->
pos
+
steps
*
factor
;
int32_t
pos
=
pQuery
->
pos
+
steps
*
factor
;
...
@@ -7071,20 +7070,20 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
...
@@ -7071,20 +7070,20 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
if
(
pQuery
->
lastKey
>
pSupporter
->
rawEKey
||
nextKey
>
pSupporter
->
rawEKey
)
{
if
(
pQuery
->
lastKey
>
pSupporter
->
rawEKey
||
nextKey
>
pSupporter
->
rawEKey
)
{
/* whole query completed, save result and abort */
/* whole query completed, save result and abort */
assert
(
queryCompleted
);
assert
(
queryCompleted
);
saveResult
(
pSupporter
,
p
Info
,
p
Info
->
lastResRows
);
saveResult
(
pSupporter
,
p
MeterQueryInfo
,
pMeterQuery
Info
->
lastResRows
);
// save the pQuery->lastKey for retrieve data in cache, actually, there will be no qualified data in cache.
// save the pQuery->lastKey for retrieve data in cache, actually, there will be no qualified data in cache.
saveIntervalQueryRange
(
pRuntimeEnv
,
pInfo
);
saveIntervalQueryRange
(
pRuntimeEnv
,
p
MeterQuery
Info
);
}
else
if
(
pQuery
->
ekey
==
pBlockInfo
->
keyLast
)
{
}
else
if
(
pQuery
->
ekey
==
pBlockInfo
->
keyLast
)
{
/* current interval query is completed, set the next query range on other data blocks if exist */
/* current interval query is completed, set the next query range on other data blocks if exist */
int64_t
prevEKey
=
pQuery
->
ekey
;
int64_t
prevEKey
=
pQuery
->
ekey
;
getAlignedIntervalQueryRange
(
pRuntimeEnv
,
pQuery
->
lastKey
,
pSupporter
->
rawSKey
,
pSupporter
->
rawEKey
);
getAlignedIntervalQueryRange
(
pRuntimeEnv
,
pQuery
->
lastKey
,
pSupporter
->
rawSKey
,
pSupporter
->
rawEKey
);
saveIntervalQueryRange
(
pRuntimeEnv
,
pInfo
);
saveIntervalQueryRange
(
pRuntimeEnv
,
p
MeterQuery
Info
);
assert
(
queryCompleted
&&
prevEKey
<
pQuery
->
skey
);
assert
(
queryCompleted
&&
prevEKey
<
pQuery
->
skey
);
if
(
pInfo
->
lastResRows
>
0
)
{
if
(
p
MeterQuery
Info
->
lastResRows
>
0
)
{
saveResult
(
pSupporter
,
p
Info
,
p
Info
->
lastResRows
);
saveResult
(
pSupporter
,
p
MeterQueryInfo
,
pMeterQuery
Info
->
lastResRows
);
}
}
}
else
{
}
else
{
/*
/*
...
@@ -7095,7 +7094,7 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
...
@@ -7095,7 +7094,7 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
* With the information of the directly next data block, whether locates in cache or disk,
* With the information of the directly next data block, whether locates in cache or disk,
* current interval query being completed or not can be decided.
* current interval query being completed or not can be decided.
*/
*/
saveIntervalQueryRange
(
pRuntimeEnv
,
pInfo
);
saveIntervalQueryRange
(
pRuntimeEnv
,
p
MeterQuery
Info
);
assert
(
pQuery
->
lastKey
>
pBlockInfo
->
keyLast
&&
pQuery
->
lastKey
<=
pQuery
->
ekey
);
assert
(
pQuery
->
lastKey
>
pBlockInfo
->
keyLast
&&
pQuery
->
lastKey
<=
pQuery
->
ekey
);
/*
/*
...
@@ -7103,7 +7102,7 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
...
@@ -7103,7 +7102,7 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
* merge with other meters in the same group
* merge with other meters in the same group
*/
*/
if
(
queryCompleted
)
{
if
(
queryCompleted
)
{
saveResult
(
pSupporter
,
p
Info
,
p
Info
->
lastResRows
);
saveResult
(
pSupporter
,
p
MeterQueryInfo
,
pMeterQuery
Info
->
lastResRows
);
}
}
}
}
...
@@ -7119,23 +7118,23 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
...
@@ -7119,23 +7118,23 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
if
(
pQuery
->
lastKey
<
pSupporter
->
rawEKey
||
(
nextKey
<
pSupporter
->
rawEKey
&&
nextKey
!=
-
1
))
{
if
(
pQuery
->
lastKey
<
pSupporter
->
rawEKey
||
(
nextKey
<
pSupporter
->
rawEKey
&&
nextKey
!=
-
1
))
{
/* whole query completed, save result and abort */
/* whole query completed, save result and abort */
assert
(
queryCompleted
);
assert
(
queryCompleted
);
saveResult
(
pSupporter
,
p
Info
,
p
Info
->
lastResRows
);
saveResult
(
pSupporter
,
p
MeterQueryInfo
,
pMeterQuery
Info
->
lastResRows
);
/*
/*
* save the pQuery->lastKey for retrieve data in cache, actually,
* save the pQuery->lastKey for retrieve data in cache, actually,
* there will be no qualified data in cache.
* there will be no qualified data in cache.
*/
*/
saveIntervalQueryRange
(
pRuntimeEnv
,
pInfo
);
saveIntervalQueryRange
(
pRuntimeEnv
,
p
MeterQuery
Info
);
}
else
if
(
pQuery
->
ekey
==
pBlockInfo
->
keyFirst
)
{
}
else
if
(
pQuery
->
ekey
==
pBlockInfo
->
keyFirst
)
{
// current interval query is completed, set the next query range on other data blocks if exist
// current interval query is completed, set the next query range on other data blocks if exist
int64_t
prevEKey
=
pQuery
->
ekey
;
int64_t
prevEKey
=
pQuery
->
ekey
;
getAlignedIntervalQueryRange
(
pRuntimeEnv
,
pQuery
->
lastKey
,
pSupporter
->
rawSKey
,
pSupporter
->
rawEKey
);
getAlignedIntervalQueryRange
(
pRuntimeEnv
,
pQuery
->
lastKey
,
pSupporter
->
rawSKey
,
pSupporter
->
rawEKey
);
saveIntervalQueryRange
(
pRuntimeEnv
,
pInfo
);
saveIntervalQueryRange
(
pRuntimeEnv
,
p
MeterQuery
Info
);
assert
(
queryCompleted
&&
prevEKey
>
pQuery
->
skey
);
assert
(
queryCompleted
&&
prevEKey
>
pQuery
->
skey
);
if
(
pInfo
->
lastResRows
>
0
)
{
if
(
p
MeterQuery
Info
->
lastResRows
>
0
)
{
saveResult
(
pSupporter
,
p
Info
,
p
Info
->
lastResRows
);
saveResult
(
pSupporter
,
p
MeterQueryInfo
,
pMeterQuery
Info
->
lastResRows
);
}
}
}
else
{
}
else
{
/*
/*
...
@@ -7146,7 +7145,7 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
...
@@ -7146,7 +7145,7 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
* With the information of the directly next data block, whether locates in cache or disk,
* With the information of the directly next data block, whether locates in cache or disk,
* current interval query being completed or not can be decided.
* current interval query being completed or not can be decided.
*/
*/
saveIntervalQueryRange
(
pRuntimeEnv
,
pInfo
);
saveIntervalQueryRange
(
pRuntimeEnv
,
p
MeterQuery
Info
);
assert
(
pQuery
->
lastKey
<
pBlockInfo
->
keyFirst
&&
pQuery
->
lastKey
>=
pQuery
->
ekey
);
assert
(
pQuery
->
lastKey
<
pBlockInfo
->
keyFirst
&&
pQuery
->
lastKey
>=
pQuery
->
ekey
);
/*
/*
...
@@ -7154,7 +7153,7 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
...
@@ -7154,7 +7153,7 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
* flag, and merge with other meters in the same group
* flag, and merge with other meters in the same group
*/
*/
if
(
queryCompleted
)
{
if
(
queryCompleted
)
{
saveResult
(
pSupporter
,
p
Info
,
p
Info
->
lastResRows
);
saveResult
(
pSupporter
,
p
MeterQueryInfo
,
pMeterQuery
Info
->
lastResRows
);
}
}
}
}
...
@@ -7163,14 +7162,14 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
...
@@ -7163,14 +7162,14 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
}
}
assert
(
queryCompleted
);
assert
(
queryCompleted
);
saveResult
(
pSupporter
,
p
Info
,
p
Info
->
lastResRows
);
saveResult
(
pSupporter
,
p
MeterQueryInfo
,
pMeterQuery
Info
->
lastResRows
);
assert
((
nextKey
>=
pQuery
->
lastKey
&&
QUERY_IS_ASC_QUERY
(
pQuery
))
||
assert
((
nextKey
>=
pQuery
->
lastKey
&&
QUERY_IS_ASC_QUERY
(
pQuery
))
||
(
nextKey
<=
pQuery
->
lastKey
&&
!
QUERY_IS_ASC_QUERY
(
pQuery
)));
(
nextKey
<=
pQuery
->
lastKey
&&
!
QUERY_IS_ASC_QUERY
(
pQuery
)));
/* still in the same block to query */
/* still in the same block to query */
getAlignedIntervalQueryRange
(
pRuntimeEnv
,
nextKey
,
pSupporter
->
rawSKey
,
pSupporter
->
rawEKey
);
getAlignedIntervalQueryRange
(
pRuntimeEnv
,
nextKey
,
pSupporter
->
rawSKey
,
pSupporter
->
rawEKey
);
saveIntervalQueryRange
(
pRuntimeEnv
,
pInfo
);
saveIntervalQueryRange
(
pRuntimeEnv
,
p
MeterQuery
Info
);
int32_t
newPos
=
searchFn
((
char
*
)
pPrimaryCol
,
pBlockInfo
->
size
,
pQuery
->
skey
,
pQuery
->
order
.
order
);
int32_t
newPos
=
searchFn
((
char
*
)
pPrimaryCol
,
pBlockInfo
->
size
,
pQuery
->
skey
,
pQuery
->
order
.
order
);
assert
(
newPos
==
pQuery
->
pos
+
steps
*
factor
);
assert
(
newPos
==
pQuery
->
pos
+
steps
*
factor
);
...
@@ -7179,26 +7178,25 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
...
@@ -7179,26 +7178,25 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
}
}
}
}
static
void
doApplyIntervalQueryOnBlock_rv
(
SMeterQuerySupportObj
*
pSupporter
,
SMeterQueryInfo
*
pInfo
,
static
void
doApplyIntervalQueryOnBlock_rv
(
SMeterQuerySupportObj
*
pSupporter
,
SMeterQueryInfo
*
p
MeterQuery
Info
,
SBlockInfo
*
pBlockInfo
,
int64_t
*
pPrimaryCol
,
SField
*
pFields
,
SBlockInfo
*
pBlockInfo
,
int64_t
*
pPrimaryCol
,
SField
*
pFields
,
__block_search_fn_t
searchFn
)
{
__block_search_fn_t
searchFn
)
{
SQueryRuntimeEnv
*
pRuntimeEnv
=
&
pSupporter
->
runtimeEnv
;
SQueryRuntimeEnv
*
pRuntimeEnv
=
&
pSupporter
->
runtimeEnv
;
SQuery
*
pQuery
=
pRuntimeEnv
->
pQuery
;
SQuery
*
pQuery
=
pRuntimeEnv
->
pQuery
;
int32_t
factor
=
GET_FORWARD_DIRECTION_FACTOR
(
pQuery
->
order
.
order
);
int32_t
factor
=
GET_FORWARD_DIRECTION_FACTOR
(
pQuery
->
order
.
order
);
int64_t
nextKey
=
-
1
;
bool
completed
=
false
;
while
(
1
)
{
while
(
1
)
{
int64_t
nextKey
=
-
1
;
int32_t
numOfRes
=
0
;
int32_t
numOfRes
=
0
;
int32_t
steps
=
applyFunctionsOnBlock
(
pRuntimeEnv
,
pBlockInfo
,
pPrimaryCol
,
pFields
,
searchFn
,
&
numOfRes
);
int32_t
steps
=
applyFunctionsOnBlock
(
pRuntimeEnv
,
pBlockInfo
,
pPrimaryCol
,
pFields
,
searchFn
,
&
numOfRes
);
assert
(
steps
>
0
);
assert
(
steps
>
0
);
// NOTE: in case of stable query, only ONE(or ZERO) row of result generated for each query range
// NOTE: in case of stable query, only ONE(or ZERO) row of result generated for each query range
if
(
pInfo
->
lastResRows
==
0
)
{
if
(
p
MeterQuery
Info
->
lastResRows
==
0
)
{
pInfo
->
lastResRows
=
numOfRes
;
p
MeterQuery
Info
->
lastResRows
=
numOfRes
;
}
else
{
}
else
{
assert
(
pInfo
->
lastResRows
==
1
);
assert
(
p
MeterQuery
Info
->
lastResRows
==
1
);
}
}
int32_t
pos
=
pQuery
->
pos
+
steps
*
factor
;
int32_t
pos
=
pQuery
->
pos
+
steps
*
factor
;
...
@@ -7212,6 +7210,7 @@ static void doApplyIntervalQueryOnBlock_rv(SMeterQuerySupportObj *pSupporter, SM
...
@@ -7212,6 +7210,7 @@ static void doApplyIntervalQueryOnBlock_rv(SMeterQuerySupportObj *pSupporter, SM
}
}
// all data satisfy current query are checked, query completed
// all data satisfy current query are checked, query completed
bool
completed
=
false
;
if
(
QUERY_IS_ASC_QUERY
(
pQuery
))
{
if
(
QUERY_IS_ASC_QUERY
(
pQuery
))
{
completed
=
(
pQuery
->
lastKey
>
pQuery
->
ekey
);
completed
=
(
pQuery
->
lastKey
>
pQuery
->
ekey
);
}
else
{
}
else
{
...
@@ -7232,7 +7231,7 @@ static void doApplyIntervalQueryOnBlock_rv(SMeterQuerySupportObj *pSupporter, SM
...
@@ -7232,7 +7231,7 @@ static void doApplyIntervalQueryOnBlock_rv(SMeterQuerySupportObj *pSupporter, SM
* With the information of the directly next data block, whether locates in cache or disk,
* With the information of the directly next data block, whether locates in cache or disk,
* current interval query being completed or not can be decided.
* current interval query being completed or not can be decided.
*/
*/
saveIntervalQueryRange
(
pRuntimeEnv
,
pInfo
);
saveIntervalQueryRange
(
pRuntimeEnv
,
p
MeterQuery
Info
);
if
(
QUERY_IS_ASC_QUERY
(
pQuery
))
{
if
(
QUERY_IS_ASC_QUERY
(
pQuery
))
{
assert
(
pQuery
->
lastKey
>
pBlockInfo
->
keyLast
&&
pQuery
->
lastKey
<=
pQuery
->
ekey
);
assert
(
pQuery
->
lastKey
>
pBlockInfo
->
keyLast
&&
pQuery
->
lastKey
<=
pQuery
->
ekey
);
...
@@ -7247,36 +7246,28 @@ static void doApplyIntervalQueryOnBlock_rv(SMeterQuerySupportObj *pSupporter, SM
...
@@ -7247,36 +7246,28 @@ static void doApplyIntervalQueryOnBlock_rv(SMeterQuerySupportObj *pSupporter, SM
if
(
pQuery
->
ekey
==
pSupporter
->
rawEKey
)
{
if
(
pQuery
->
ekey
==
pSupporter
->
rawEKey
)
{
/* whole query completed, save result and abort */
/* whole query completed, save result and abort */
saveResult
(
pSupporter
,
p
Info
,
p
Info
->
lastResRows
);
saveResult
(
pSupporter
,
p
MeterQueryInfo
,
pMeterQuery
Info
->
lastResRows
);
// save the pQuery->lastKey for retrieve data in cache, actually, there will be no qualified data in cache.
// save the pQuery->lastKey for retrieve data in cache, actually, there will be no qualified data in cache.
saveIntervalQueryRange
(
pRuntimeEnv
,
pInfo
);
saveIntervalQueryRange
(
pRuntimeEnv
,
p
MeterQuery
Info
);
return
;
return
;
}
else
if
((
QUERY_IS_ASC_QUERY
(
pQuery
)
&&
pQuery
->
ekey
==
pBlockInfo
->
keyLast
)
||
}
else
if
((
QUERY_IS_ASC_QUERY
(
pQuery
)
&&
pQuery
->
ekey
==
pBlockInfo
->
keyLast
)
||
(
QUERY_IS_ASC_QUERY
(
pQuery
)
&&
pQuery
->
ekey
==
pBlockInfo
->
keyFirst
))
{
(
!
QUERY_IS_ASC_QUERY
(
pQuery
)
&&
pQuery
->
ekey
==
pBlockInfo
->
keyFirst
))
{
// /* current interval query is completed, set the next query range on other data blocks if exist */
/* current interval query is completed, set the next query range on other data blocks if exist */
// int64_t prevEKey = pQuery->ekey;
saveIntervalQueryRange
(
pRuntimeEnv
,
pMeterQueryInfo
);
//
// getAlignedIntervalQueryRange(pRuntimeEnv, pQuery->lastKey, pSupporter->rawSKey, pSupporter->rawEKey);
saveIntervalQueryRange
(
pRuntimeEnv
,
pInfo
);
//
// assert(prevEKey < pQuery->skey);
// if (pInfo->lastResRows > 0) {
// saveResult(pSupporter, pInfo, pInfo->lastResRows);
// }
//
return
;
return
;
}
}
saveResult
(
pSupporter
,
p
Info
,
p
Info
->
lastResRows
);
saveResult
(
pSupporter
,
p
MeterQueryInfo
,
pMeterQuery
Info
->
lastResRows
);
assert
(
pos
>=
0
&&
pos
<
pBlockInfo
->
size
);
assert
((
nextKey
>=
pQuery
->
lastKey
&&
QUERY_IS_ASC_QUERY
(
pQuery
))
||
assert
((
nextKey
>=
pQuery
->
lastKey
&&
QUERY_IS_ASC_QUERY
(
pQuery
))
||
(
nextKey
<=
pQuery
->
lastKey
&&
!
QUERY_IS_ASC_QUERY
(
pQuery
)));
(
nextKey
<=
pQuery
->
lastKey
&&
!
QUERY_IS_ASC_QUERY
(
pQuery
)));
/* still in the same block to query */
/* still in the same block to query */
getAlignedIntervalQueryRange
(
pRuntimeEnv
,
nextKey
,
pSupporter
->
rawSKey
,
pSupporter
->
rawEKey
);
getAlignedIntervalQueryRange
(
pRuntimeEnv
,
nextKey
,
pSupporter
->
rawSKey
,
pSupporter
->
rawEKey
);
saveIntervalQueryRange
(
pRuntimeEnv
,
pInfo
);
saveIntervalQueryRange
(
pRuntimeEnv
,
p
MeterQuery
Info
);
int32_t
newPos
=
searchFn
((
char
*
)
pPrimaryCol
,
pBlockInfo
->
size
,
pQuery
->
skey
,
pQuery
->
order
.
order
);
int32_t
newPos
=
searchFn
((
char
*
)
pPrimaryCol
,
pBlockInfo
->
size
,
pQuery
->
skey
,
pQuery
->
order
.
order
);
assert
(
newPos
==
pQuery
->
pos
+
steps
*
factor
);
assert
(
newPos
==
pQuery
->
pos
+
steps
*
factor
);
...
@@ -7691,19 +7682,20 @@ void copyFromGroupBuf(SQInfo *pQInfo, SOutputRes *result) {
...
@@ -7691,19 +7682,20 @@ void copyFromGroupBuf(SQInfo *pQInfo, SOutputRes *result) {
assert
(
pQuery
->
pointsRead
<=
pQuery
->
pointsToRead
);
assert
(
pQuery
->
pointsRead
<=
pQuery
->
pointsToRead
);
}
}
static
void
applyIntervalQueryOnBlock
(
SMeterQuerySupportObj
*
pSupporter
,
SMeterDataInfo
*
p
InfoEx
,
int64_t
*
pPrimaryData
,
static
void
applyIntervalQueryOnBlock
(
SMeterQuerySupportObj
*
pSupporter
,
SMeterDataInfo
*
p
MeterDataInfo
,
SBlockInfo
*
pBlockInfo
,
int32_t
blockStatus
,
SField
*
pFields
,
SBlockInfo
*
pBlockInfo
,
int32_t
blockStatus
,
SField
*
pFields
,
__block_search_fn_t
searchFn
)
{
__block_search_fn_t
searchFn
)
{
SQueryRuntimeEnv
*
pRuntimeEnv
=
&
pSupporter
->
runtimeEnv
;
SQueryRuntimeEnv
*
pRuntimeEnv
=
&
pSupporter
->
runtimeEnv
;
SQuery
*
pQuery
=
pRuntimeEnv
->
pQuery
;
SQuery
*
pQuery
=
pRuntimeEnv
->
pQuery
;
SMeterQueryInfo
*
p
Info
=
pInfoEx
->
pMeterQInfo
;
SMeterQueryInfo
*
p
MeterQueryInfo
=
pMeterDataInfo
->
pMeterQInfo
;
int64_t
*
pPrimaryKey
=
(
int64_t
*
)
pRuntimeEnv
->
primaryColBuffer
->
data
;
/*
/*
* for each block, we need to handle the previous query, since the determination of previous query being completed
* for each block, we need to handle the previous query, since the determination of previous query being completed
* or not is based on the start key of current block.
* or not is based on the start key of current block.
*/
*/
TSKEY
key
=
getNextAccessedKeyInData
(
pQuery
,
pPrimary
Data
,
pBlockInfo
,
blockStatus
);
TSKEY
key
=
getNextAccessedKeyInData
(
pQuery
,
pPrimary
Key
,
pBlockInfo
,
blockStatus
);
setIntervalQueryRange
(
p
InfoEx
->
pMeterQInfo
,
pSupporter
,
key
);
setIntervalQueryRange
(
p
MeterDataInfo
->
pMeterQInfo
,
pSupporter
,
key
);
if
(((
pQuery
->
skey
>
pQuery
->
ekey
)
&&
QUERY_IS_ASC_QUERY
(
pQuery
))
||
if
(((
pQuery
->
skey
>
pQuery
->
ekey
)
&&
QUERY_IS_ASC_QUERY
(
pQuery
))
||
((
pQuery
->
skey
<
pQuery
->
ekey
)
&&
!
QUERY_IS_ASC_QUERY
(
pQuery
)))
{
((
pQuery
->
skey
<
pQuery
->
ekey
)
&&
!
QUERY_IS_ASC_QUERY
(
pQuery
)))
{
...
@@ -7714,18 +7706,18 @@ static void applyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMeterD
...
@@ -7714,18 +7706,18 @@ static void applyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMeterD
((
pBlockInfo
->
keyFirst
>
pQuery
->
ekey
)
&&
!
QUERY_IS_ASC_QUERY
(
pQuery
)))
{
((
pBlockInfo
->
keyFirst
>
pQuery
->
ekey
)
&&
!
QUERY_IS_ASC_QUERY
(
pQuery
)))
{
int32_t
numOfRes
=
0
;
int32_t
numOfRes
=
0
;
/* current block is included in this interval */
/* current block is included in this interval */
int32_t
steps
=
applyFunctionsOnBlock
(
pRuntimeEnv
,
pBlockInfo
,
pPrimary
Data
,
pFields
,
searchFn
,
&
numOfRes
);
int32_t
steps
=
applyFunctionsOnBlock
(
pRuntimeEnv
,
pBlockInfo
,
pPrimary
Key
,
pFields
,
searchFn
,
&
numOfRes
);
assert
(
numOfRes
<=
1
&&
numOfRes
>=
0
&&
steps
>
0
);
assert
(
numOfRes
<=
1
&&
numOfRes
>=
0
&&
steps
>
0
);
if
(
pInfo
->
lastResRows
==
0
)
{
if
(
p
MeterQuery
Info
->
lastResRows
==
0
)
{
pInfo
->
lastResRows
=
numOfRes
;
p
MeterQuery
Info
->
lastResRows
=
numOfRes
;
}
else
{
}
else
{
assert
(
pInfo
->
lastResRows
==
1
);
assert
(
p
MeterQuery
Info
->
lastResRows
==
1
);
}
}
saveIntervalQueryRange
(
pRuntimeEnv
,
pInfo
);
saveIntervalQueryRange
(
pRuntimeEnv
,
p
MeterQuery
Info
);
}
else
{
}
else
{
doApplyIntervalQueryOnBlock
(
pSupporter
,
p
Info
,
pBlockInfo
,
pPrimaryData
,
pFields
,
searchFn
);
doApplyIntervalQueryOnBlock
(
pSupporter
,
p
MeterQueryInfo
,
pBlockInfo
,
pPrimaryKey
,
pFields
,
searchFn
);
}
}
}
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录