Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
47cb043f
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看板
提交
47cb043f
编写于
7月 22, 2022
作者:
5
54liuyao
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
feat(stream):optimize update scan range
上级
164dea54
变更
6
显示空白变更内容
内联
并排
Showing
6 changed file
with
206 addition
and
332 deletion
+206
-332
source/libs/executor/inc/executorimpl.h
source/libs/executor/inc/executorimpl.h
+10
-10
source/libs/executor/src/executorimpl.c
source/libs/executor/src/executorimpl.c
+1
-2
source/libs/executor/src/scanoperator.c
source/libs/executor/src/scanoperator.c
+132
-256
source/libs/executor/src/timewindowoperator.c
source/libs/executor/src/timewindowoperator.c
+51
-52
source/libs/function/src/builtinsimpl.c
source/libs/function/src/builtinsimpl.c
+10
-10
tests/script/tsim/stream/sliding.sim
tests/script/tsim/stream/sliding.sim
+2
-2
未找到文件。
source/libs/executor/inc/executorimpl.h
浏览文件 @
47cb043f
...
...
@@ -55,8 +55,9 @@ typedef int32_t (*__block_search_fn_t)(char* data, int32_t num, int64_t key, int
#define START_TS_COLUMN_INDEX 0
#define END_TS_COLUMN_INDEX 1
#define UID_COLUMN_INDEX 2
#define GROUPID_COLUMN_INDEX UID_COLUMN_INDEX
#define DELETE_GROUPID_COLUMN_INDEX 2
#define GROUPID_COLUMN_INDEX 3
#define CALCULATE_START_TS_COLUMN_INDEX 4
#define CALCULATE_END_TS_COLUMN_INDEX 5
enum
{
// when this task starts to execute, this status will set
...
...
@@ -346,7 +347,6 @@ typedef enum EStreamScanMode {
STREAM_SCAN_FROM_READERHANDLE
=
1
,
STREAM_SCAN_FROM_RES
,
STREAM_SCAN_FROM_UPDATERES
,
STREAM_SCAN_FROM_DATAREADER
,
// todo(liuyao) delete it
STREAM_SCAN_FROM_DATAREADER_RETRIEVE
,
STREAM_SCAN_FROM_DATAREADER_RANGE
,
}
EStreamScanMode
;
...
...
@@ -366,7 +366,7 @@ typedef struct SStreamAggSupporter {
char
*
pKeyBuf
;
// window key buffer
SDiskbasedBuf
*
pResultBuf
;
// query result buffer based on blocked-wised disk file
int32_t
resultRowSize
;
// the result buffer size for each result row, with the meta data size for each row
S
Array
*
pScanWindow
;
S
SDataBlock
*
pScanBlock
;
}
SStreamAggSupporter
;
typedef
struct
SessionWindowSupporter
{
...
...
@@ -419,7 +419,7 @@ typedef struct SStreamScanInfo {
int32_t
deleteDataIndex
;
STimeWindow
updateWin
;
STimeWindowAggSupp
twAggSup
;
SSDataBlock
*
pUpdateDataRes
;
// status for tmq
// SSchemaWrapper schema;
STqOffset
offset
;
...
...
@@ -712,7 +712,6 @@ typedef struct SStreamStateAggOperatorInfo {
SSDataBlock
*
pDelRes
;
SHashObj
*
pSeDeleted
;
void
*
pDelIterator
;
SArray
*
pScanWindow
;
SArray
*
pChildren
;
// cache for children's result;
bool
ignoreExpiredData
;
}
SStreamStateAggOperatorInfo
;
...
...
@@ -954,6 +953,7 @@ int32_t updateSessionWindowInfo(SResultWindowInfo* pWinInfo, TSKEY* pStartTs,
TSKEY
*
pEndTs
,
int32_t
rows
,
int32_t
start
,
int64_t
gap
,
SHashObj
*
pStDeleted
);
bool
functionNeedToExecute
(
SqlFunctionCtx
*
pCtx
);
bool
isCloseWindow
(
STimeWindow
*
pWin
,
STimeWindowAggSupp
*
pSup
);
void
appendOneRow
(
SSDataBlock
*
pBlock
,
TSKEY
*
pStartTs
,
TSKEY
*
pEndTs
,
uint64_t
*
pUid
);
int32_t
finalizeResultRowIntoResultDataBlock
(
SDiskbasedBuf
*
pBuf
,
SResultRowPosition
*
resultRowPosition
,
SqlFunctionCtx
*
pCtx
,
SExprInfo
*
pExprInfo
,
int32_t
numOfExprs
,
const
int32_t
*
rowCellOffset
,
...
...
@@ -970,7 +970,7 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN
void
copyUpdateDataBlock
(
SSDataBlock
*
pDest
,
SSDataBlock
*
pSource
,
int32_t
tsColIndex
);
int32_t
generateGroupIdMap
(
STableListInfo
*
pTableListInfo
,
SReadHandle
*
pHandle
,
SNodeList
*
groupKey
);
SSDataBlock
*
create
PullDataBlock
(
);
SSDataBlock
*
create
SpecialDataBlock
(
EStreamType
type
);
#ifdef __cplusplus
}
...
...
source/libs/executor/src/executorimpl.c
浏览文件 @
47cb043f
...
...
@@ -5136,8 +5136,7 @@ int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, const char* pKey, SqlF
}
pSup
->
valueSize
=
size
;
pSup
->
pScanWindow
=
taosArrayInit
(
4
,
sizeof
(
STimeWindow
));
pSup
->
pScanBlock
=
createSpecialDataBlock
(
STREAM_CLEAR
);
int32_t
pageSize
=
4096
;
while
(
pageSize
<
pSup
->
resultRowSize
*
4
)
{
pageSize
<<=
1u
;
...
...
source/libs/executor/src/scanoperator.c
浏览文件 @
47cb043f
...
...
@@ -25,7 +25,6 @@
#include "tdatablock.h"
#include "tmsg.h"
#include "executorimpl.h"
#include "query.h"
#include "tcompare.h"
#include "thash.h"
...
...
@@ -812,6 +811,10 @@ static bool isSignleIntervalWindow(SStreamScanInfo* pInfo) {
return
pInfo
->
sessionSup
.
parentType
==
QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL
;
}
static
bool
isSlidingWindow
(
SStreamScanInfo
*
pInfo
)
{
return
isIntervalWindow
(
pInfo
)
&&
pInfo
->
interval
.
interval
!=
pInfo
->
interval
.
sliding
;
}
static
uint64_t
getGroupId
(
SOperatorInfo
*
pOperator
,
uint64_t
uid
)
{
uint64_t
*
groupId
=
taosHashGet
(
pOperator
->
pTaskInfo
->
tableqinfoList
.
map
,
&
uid
,
sizeof
(
int64_t
));
if
(
groupId
)
{
...
...
@@ -834,17 +837,10 @@ static uint64_t getGroupId(SOperatorInfo* pOperator, uint64_t uid) {
}
static
void
setGroupId
(
SStreamScanInfo
*
pInfo
,
SSDataBlock
*
pBlock
,
int32_t
groupColIndex
,
int32_t
rowIndex
)
{
ASSERT
(
rowIndex
<
pBlock
->
info
.
rows
);
switch
(
pBlock
->
info
.
type
)
{
case
STREAM_DELETE_DATA
:
case
STREAM_RETRIEVE
:
{
SColumnInfoData
*
pColInfo
=
taosArrayGet
(
pBlock
->
pDataBlock
,
groupColIndex
);
uint64_t
*
groupCol
=
(
uint64_t
*
)
pColInfo
->
pData
;
ASSERT
(
rowIndex
<
pBlock
->
info
.
rows
);
pInfo
->
groupId
=
groupCol
[
rowIndex
];
}
break
;
default:
break
;
}
}
void
resetTableScanInfo
(
STableScanInfo
*
pTableScanInfo
,
STimeWindow
*
pWin
)
{
...
...
@@ -864,7 +860,17 @@ static bool prepareRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pBlock, int32_
SColumnInfoData
*
pEndTsCol
=
taosArrayGet
(
pBlock
->
pDataBlock
,
END_TS_COLUMN_INDEX
);
TSKEY
*
endData
=
(
TSKEY
*
)
pEndTsCol
->
pData
;
STimeWindow
win
=
{.
skey
=
startData
[
*
pRowIndex
],
.
ekey
=
endData
[
*
pRowIndex
]};
SColumnInfoData
*
pCalStartTsCol
=
taosArrayGet
(
pBlock
->
pDataBlock
,
CALCULATE_START_TS_COLUMN_INDEX
);
TSKEY
*
calStartData
=
(
TSKEY
*
)
pCalStartTsCol
->
pData
;
SColumnInfoData
*
pCalEndTsCol
=
taosArrayGet
(
pBlock
->
pDataBlock
,
CALCULATE_END_TS_COLUMN_INDEX
);
TSKEY
*
calEndData
=
(
TSKEY
*
)
pCalEndTsCol
->
pData
;
setGroupId
(
pInfo
,
pBlock
,
GROUPID_COLUMN_INDEX
,
*
pRowIndex
);
if
(
isSlidingWindow
(
pInfo
))
{
pInfo
->
updateWin
.
skey
=
calStartData
[
*
pRowIndex
];
pInfo
->
updateWin
.
ekey
=
calEndData
[
*
pRowIndex
];
}
(
*
pRowIndex
)
++
;
for
(;
*
pRowIndex
<
pBlock
->
info
.
rows
;
(
*
pRowIndex
)
++
)
{
...
...
@@ -876,8 +882,8 @@ static bool prepareRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pBlock, int32_
win
.
skey
=
TMIN
(
win
.
skey
,
startData
[
*
pRowIndex
]);
continue
;
}
ASSERT
((
win
.
skey
>
startData
[
*
pRowIndex
]
&&
win
.
ekey
<
endData
[
*
pRowIndex
])
||
(
isInTimeWindow
(
&
win
,
startData
[
*
pRowIndex
],
0
)
||
isInTimeWindow
(
&
win
,
endData
[
*
pRowIndex
],
0
)));
ASSERT
(
!
(
win
.
skey
>
startData
[
*
pRowIndex
]
&&
win
.
ekey
<
endData
[
*
pRowIndex
])
||
!
(
isInTimeWindow
(
&
win
,
startData
[
*
pRowIndex
],
0
)
||
isInTimeWindow
(
&
win
,
endData
[
*
pRowIndex
],
0
)));
break
;
}
...
...
@@ -908,68 +914,6 @@ static STimeWindow getSlidingWindow(TSKEY* tsCol, SInterval* pInterval, SDataBlo
win
.
ekey
=
endWin
.
ekey
;
}
}
static
bool
prepareDataScan
(
SStreamScanInfo
*
pInfo
,
SSDataBlock
*
pSDB
,
int32_t
tsColIndex
,
int32_t
*
pRowIndex
)
{
STimeWindow
win
=
{
.
skey
=
INT64_MIN
,
.
ekey
=
INT64_MAX
,
};
bool
needRead
=
false
;
if
(
!
isStateWindow
(
pInfo
)
&&
(
*
pRowIndex
)
<
pSDB
->
info
.
rows
)
{
SColumnInfoData
*
pColDataInfo
=
taosArrayGet
(
pSDB
->
pDataBlock
,
tsColIndex
);
TSKEY
*
tsCols
=
(
TSKEY
*
)
pColDataInfo
->
pData
;
SResultRowInfo
dumyInfo
;
dumyInfo
.
cur
.
pageId
=
-
1
;
if
(
isSessionWindow
(
pInfo
))
{
SStreamAggSupporter
*
pAggSup
=
pInfo
->
sessionSup
.
pStreamAggSup
;
int64_t
gap
=
pInfo
->
sessionSup
.
gap
;
int32_t
winIndex
=
0
;
SResultWindowInfo
*
pCurWin
=
getSessionTimeWindow
(
pAggSup
,
tsCols
[
*
pRowIndex
],
INT64_MIN
,
pSDB
->
info
.
groupId
,
gap
,
&
winIndex
);
win
=
pCurWin
->
win
;
setGroupId
(
pInfo
,
pSDB
,
GROUPID_COLUMN_INDEX
,
*
pRowIndex
);
(
*
pRowIndex
)
+=
updateSessionWindowInfo
(
pCurWin
,
tsCols
,
NULL
,
pSDB
->
info
.
rows
,
*
pRowIndex
,
gap
,
NULL
);
}
else
{
setGroupId
(
pInfo
,
pSDB
,
GROUPID_COLUMN_INDEX
,
*
pRowIndex
);
pInfo
->
updateWin
.
skey
=
tsCols
[
*
pRowIndex
];
win
=
getSlidingWindow
(
tsCols
,
&
pInfo
->
interval
,
&
pSDB
->
info
,
pRowIndex
);
pInfo
->
updateWin
.
ekey
=
tsCols
[
*
pRowIndex
-
1
];
// win = getActiveTimeWindow(NULL, &dumyInfo, tsCols[*pRowIndex], &pInfo->interval, TSDB_ORDER_ASC);
// (*pRowIndex) +=
// getNumOfRowsInTimeWindow(&pSDB->info, tsCols, *pRowIndex, win.ekey, binarySearchForKey, NULL,
// TSDB_ORDER_ASC);
}
needRead
=
true
;
}
else
if
(
isStateWindow
(
pInfo
))
{
SArray
*
pWins
=
pInfo
->
sessionSup
.
pStreamAggSup
->
pScanWindow
;
int32_t
size
=
taosArrayGetSize
(
pWins
);
if
(
pInfo
->
scanWinIndex
<
size
)
{
win
=
*
(
STimeWindow
*
)
taosArrayGet
(
pWins
,
pInfo
->
scanWinIndex
);
pInfo
->
scanWinIndex
++
;
needRead
=
true
;
}
else
{
pInfo
->
scanWinIndex
=
0
;
taosArrayClear
(
pWins
);
}
}
if
(
!
needRead
)
{
return
false
;
}
resetTableScanInfo
(
pInfo
->
pTableScanOp
->
info
,
&
win
);
return
true
;
}
static
void
copyOneRow
(
SSDataBlock
*
dest
,
SSDataBlock
*
source
,
int32_t
sourceRowId
)
{
for
(
int32_t
j
=
0
;
j
<
taosArrayGetSize
(
source
->
pDataBlock
);
j
++
)
{
SColumnInfoData
*
pDestCol
=
(
SColumnInfoData
*
)
taosArrayGet
(
dest
->
pDataBlock
,
j
);
SColumnInfoData
*
pSourceCol
=
(
SColumnInfoData
*
)
taosArrayGet
(
source
->
pDataBlock
,
j
);
if
(
colDataIsNull_s
(
pSourceCol
,
sourceRowId
))
{
colDataAppendNULL
(
pDestCol
,
dest
->
info
.
rows
);
}
else
{
colDataAppend
(
pDestCol
,
dest
->
info
.
rows
,
colDataGetData
(
pSourceCol
,
sourceRowId
),
false
);
}
}
dest
->
info
.
rows
++
;
}
static
SSDataBlock
*
doRangeScan
(
SStreamScanInfo
*
pInfo
,
SSDataBlock
*
pSDB
,
int32_t
tsColIndex
,
int32_t
*
pRowIndex
)
{
while
(
1
)
{
...
...
@@ -982,29 +926,6 @@ static SSDataBlock* doRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pSDB, int32
if
(
!
pResult
)
{
blockDataCleanup
(
pSDB
);
*
pRowIndex
=
0
;
STableScanInfo
*
pTableScanInfo
=
pInfo
->
pTableScanOp
->
info
;
tsdbReaderClose
(
pTableScanInfo
->
dataReader
);
pTableScanInfo
->
dataReader
=
NULL
;
return
NULL
;
}
if
(
pResult
->
info
.
groupId
==
pInfo
->
groupId
)
{
return
pResult
;
}
}
}
static
SSDataBlock
*
doDataScan
(
SStreamScanInfo
*
pInfo
,
SSDataBlock
*
pSDB
,
int32_t
tsColIndex
,
int32_t
*
pRowIndex
)
{
while
(
1
)
{
SSDataBlock
*
pResult
=
NULL
;
pResult
=
doTableScan
(
pInfo
->
pTableScanOp
);
if
(
pResult
==
NULL
)
{
if
(
prepareDataScan
(
pInfo
,
pSDB
,
tsColIndex
,
pRowIndex
))
{
// scan next window data
pResult
=
doTableScan
(
pInfo
->
pTableScanOp
);
}
}
if
(
!
pResult
)
{
pInfo
->
updateWin
=
(
STimeWindow
){.
skey
=
INT64_MIN
,
.
ekey
=
INT64_MAX
};
STableScanInfo
*
pTableScanInfo
=
pInfo
->
pTableScanOp
->
info
;
tsdbReaderClose
(
pTableScanInfo
->
dataReader
);
...
...
@@ -1017,77 +938,31 @@ static SSDataBlock* doDataScan(SStreamScanInfo* pInfo, SSDataBlock* pSDB, int32_
return
pResult
;
}
}
/* Todo(liuyao) for partition by column
SSDataBlock* pBlock = createOneDataBlock(pResult, true);
blockDataCleanup(pResult);
for (int32_t i = 0; i < pBlock->info.rows; i++) {
uint64_t id = getGroupId(pInfo->pOperatorDumy, pBlock->info.uid);
if (id == pInfo->groupId) {
copyOneRow(pResult, pBlock, i);
}
}
return pResult;
*/
}
static
void
generateIntervalTs
(
SStreamScanInfo
*
pInfo
,
SSDataBlock
*
pDelBlock
,
SOperatorInfo
*
pOperator
,
SSDataBlock
*
pUpdateRes
)
{
if
(
pDelBlock
->
info
.
rows
==
0
)
{
return
;
}
blockDataCleanup
(
pUpdateRes
);
blockDataEnsureCapacity
(
pUpdateRes
,
64
);
ASSERT
(
taosArrayGetSize
(
pDelBlock
->
pDataBlock
)
>=
3
);
SColumnInfoData
*
pStartTsCol
=
taosArrayGet
(
pDelBlock
->
pDataBlock
,
START_TS_COLUMN_INDEX
);
TSKEY
*
startData
=
(
TSKEY
*
)
pStartTsCol
->
pData
;
SColumnInfoData
*
pEndTsCol
=
taosArrayGet
(
pDelBlock
->
pDataBlock
,
END_TS_COLUMN_INDEX
);
TSKEY
*
endData
=
(
TSKEY
*
)
pEndTsCol
->
pData
;
SColumnInfoData
*
pGpCol
=
taosArrayGet
(
pDelBlock
->
pDataBlock
,
UID_COLUMN_INDEX
);
uint64_t
*
uidCol
=
(
uint64_t
*
)
pGpCol
->
pData
;
SColumnInfoData
*
pDestTsCol
=
taosArrayGet
(
pUpdateRes
->
pDataBlock
,
START_TS_COLUMN_INDEX
);
SColumnInfoData
*
pDestGpCol
=
taosArrayGet
(
pUpdateRes
->
pDataBlock
,
GROUPID_COLUMN_INDEX
);
for
(
int32_t
i
=
pInfo
->
deleteDataIndex
;
i
<
pDelBlock
->
info
.
rows
&&
i
<
pDelBlock
->
info
.
capacity
-
(
endData
[
i
]
-
startData
[
i
])
/
pInfo
->
interval
.
interval
-
1
;
i
++
)
{
uint64_t
groupId
=
getGroupId
(
pOperator
,
uidCol
[
i
]);
for
(
TSKEY
startTs
=
startData
[
i
];
startTs
<=
endData
[
i
];)
{
colDataAppend
(
pDestTsCol
,
pUpdateRes
->
info
.
rows
,
(
const
char
*
)
&
startTs
,
false
);
colDataAppend
(
pDestGpCol
,
pUpdateRes
->
info
.
rows
,
(
const
char
*
)
&
groupId
,
false
);
pUpdateRes
->
info
.
rows
++
;
startTs
=
taosTimeAdd
(
startTs
,
pInfo
->
interval
.
interval
,
pInfo
->
interval
.
intervalUnit
,
pInfo
->
interval
.
precision
);
}
pInfo
->
deleteDataIndex
++
;
}
if
(
pInfo
->
deleteDataIndex
>
0
&&
pInfo
->
deleteDataIndex
==
pDelBlock
->
info
.
rows
)
{
blockDataCleanup
(
pDelBlock
);
pInfo
->
deleteDataIndex
=
0
;
}
}
static
void
generateScanRange
(
SStreamScanInfo
*
pInfo
,
SSDataBlock
*
pBlock
,
SOperatorInfo
*
pOperator
,
SSDataBlock
*
pUpdateRes
)
{
if
(
pBlock
->
info
.
rows
==
0
)
{
return
;
static
int32_t
generateSessionScanRange
(
SStreamScanInfo
*
pInfo
,
SSDataBlock
*
pSrcBlock
,
SSDataBlock
*
pDestBlock
)
{
if
(
pSrcBlock
->
info
.
rows
==
0
)
{
return
TSDB_CODE_SUCCESS
;
}
blockDataCleanup
(
pUpdateRes
);
blockDataEnsureCapacity
(
pUpdateRes
,
pBlock
->
info
.
rows
);
ASSERT
(
taosArrayGetSize
(
pBlock
->
pDataBlock
)
>=
3
);
SColumnInfoData
*
pStartTsCol
=
taosArrayGet
(
pBlock
->
pDataBlock
,
START_TS_COLUMN_INDEX
);
blockDataCleanup
(
pDestBlock
);
int32_t
code
=
blockDataEnsureCapacity
(
pDestBlock
,
pSrcBlock
->
info
.
rows
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
return
code
;
}
ASSERT
(
taosArrayGetSize
(
pSrcBlock
->
pDataBlock
)
>=
3
);
SColumnInfoData
*
pStartTsCol
=
taosArrayGet
(
pSrcBlock
->
pDataBlock
,
START_TS_COLUMN_INDEX
);
TSKEY
*
startData
=
(
TSKEY
*
)
pStartTsCol
->
pData
;
SColumnInfoData
*
pEndTsCol
=
taosArrayGet
(
pBlock
->
pDataBlock
,
END_TS_COLUMN_INDEX
);
SColumnInfoData
*
pEndTsCol
=
taosArrayGet
(
p
Src
Block
->
pDataBlock
,
END_TS_COLUMN_INDEX
);
TSKEY
*
endData
=
(
TSKEY
*
)
pEndTsCol
->
pData
;
SColumnInfoData
*
p
GpCol
=
taosArrayGet
(
p
Block
->
pDataBlock
,
UID_COLUMN_INDEX
);
uint64_t
*
uidCol
=
(
uint64_t
*
)
p
Gp
Col
->
pData
;
SColumnInfoData
*
p
UidCol
=
taosArrayGet
(
pSrc
Block
->
pDataBlock
,
UID_COLUMN_INDEX
);
uint64_t
*
uidCol
=
(
uint64_t
*
)
p
Uid
Col
->
pData
;
SColumnInfoData
*
pDestStartCol
=
taosArrayGet
(
p
UpdateRes
->
pDataBlock
,
START_TS_COLUMN_INDEX
);
SColumnInfoData
*
pDestEndCol
=
taosArrayGet
(
p
UpdateRes
->
pDataBlock
,
END_TS_COLUMN_INDEX
);
SColumnInfoData
*
pDestGpCol
=
taosArrayGet
(
p
UpdateRes
->
pDataBlock
,
GROUPID_COLUMN_INDEX
);
SColumnInfoData
*
pDestStartCol
=
taosArrayGet
(
p
DestBlock
->
pDataBlock
,
START_TS_COLUMN_INDEX
);
SColumnInfoData
*
pDestEndCol
=
taosArrayGet
(
p
DestBlock
->
pDataBlock
,
END_TS_COLUMN_INDEX
);
SColumnInfoData
*
pDestGpCol
=
taosArrayGet
(
p
DestBlock
->
pDataBlock
,
GROUPID_COLUMN_INDEX
);
int32_t
dummy
=
0
;
for
(
int32_t
i
=
0
;
i
<
pBlock
->
info
.
rows
;
i
++
)
{
uint64_t
groupId
=
getGroupId
(
p
Operator
,
uidCol
[
i
]);
for
(
int32_t
i
=
0
;
i
<
p
Src
Block
->
info
.
rows
;
i
++
)
{
uint64_t
groupId
=
getGroupId
(
p
Info
->
pTableScanOp
,
uidCol
[
i
]);
// gap must be 0.
SResultWindowInfo
*
pStartWin
=
getCurSessionWindow
(
pInfo
->
sessionSup
.
pStreamAggSup
,
startData
[
i
],
endData
[
i
],
groupId
,
0
,
&
dummy
);
...
...
@@ -1101,46 +976,75 @@ static void generateScanRange(SStreamScanInfo* pInfo, SSDataBlock* pBlock, SOper
colDataAppend
(
pDestStartCol
,
i
,
(
const
char
*
)
&
pStartWin
->
win
.
skey
,
false
);
colDataAppend
(
pDestEndCol
,
i
,
(
const
char
*
)
&
pEndWin
->
win
.
ekey
,
false
);
colDataAppend
(
pDestGpCol
,
i
,
(
const
char
*
)
&
groupId
,
false
);
pUpdateRes
->
info
.
rows
++
;
}
}
static
void
setUpdateData
(
SStreamScanInfo
*
pInfo
,
SSDataBlock
*
pBlock
,
SSDataBlock
*
pUpdateBlock
)
{
blockDataCleanup
(
pUpdateBlock
);
int32_t
size
=
taosArrayGetSize
(
pInfo
->
tsArray
);
if
(
pInfo
->
tsArrayIndex
<
size
)
{
SColumnInfoData
*
pCol
=
(
SColumnInfoData
*
)
taosArrayGet
(
pUpdateBlock
->
pDataBlock
,
pInfo
->
primaryTsIndex
);
ASSERT
(
pCol
->
info
.
type
==
TSDB_DATA_TYPE_TIMESTAMP
);
blockDataEnsureCapacity
(
pUpdateBlock
,
size
);
int32_t
rowId
=
*
(
int32_t
*
)
taosArrayGet
(
pInfo
->
tsArray
,
pInfo
->
tsArrayIndex
);
pInfo
->
groupId
=
getGroupId
(
pInfo
->
pTableScanOp
,
pBlock
->
info
.
uid
);
int32_t
i
=
0
;
for
(;
i
<
size
;
i
++
)
{
rowId
=
*
(
int32_t
*
)
taosArrayGet
(
pInfo
->
tsArray
,
i
+
pInfo
->
tsArrayIndex
);
uint64_t
id
=
getGroupId
(
pInfo
->
pTableScanOp
,
pBlock
->
info
.
uid
);
if
(
pInfo
->
groupId
!=
id
)
{
break
;
pDestBlock
->
info
.
rows
++
;
}
copyOneRow
(
pUpdateBlock
,
pBlock
,
rowId
);
return
TSDB_CODE_SUCCESS
;
}
static
int32_t
generateIntervalScanRange
(
SStreamScanInfo
*
pInfo
,
SSDataBlock
*
pSrcBlock
,
SSDataBlock
*
pDestBlock
)
{
blockDataCleanup
(
pDestBlock
);
int32_t
rows
=
pSrcBlock
->
info
.
rows
;
if
(
rows
==
0
)
{
return
TSDB_CODE_SUCCESS
;
}
pUpdateBlock
->
info
.
rows
=
i
;
pInfo
->
tsArrayIndex
+=
i
;
pUpdateBlock
->
info
.
groupId
=
pInfo
->
groupId
;
pUpdateBlock
->
info
.
type
=
STREAM_CLEAR
;
blockDataUpdateTsWindow
(
pUpdateBlock
,
0
);
int32_t
code
=
blockDataEnsureCapacity
(
pDestBlock
,
rows
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
return
code
;
}
// all rows have same group id
ASSERT
(
pInfo
->
tsArrayIndex
>=
size
);
if
(
size
>
0
&&
pInfo
->
tsArrayIndex
==
size
)
{
taosArrayClear
(
pInfo
->
tsArray
);
SColumnInfoData
*
pTsCol
=
(
SColumnInfoData
*
)
taosArrayGet
(
pSrcBlock
->
pDataBlock
,
START_TS_COLUMN_INDEX
);
SColumnInfoData
*
pUidCol
=
taosArrayGet
(
pSrcBlock
->
pDataBlock
,
UID_COLUMN_INDEX
);
uint64_t
*
uidCol
=
(
uint64_t
*
)
pUidCol
->
pData
;
ASSERT
(
pTsCol
->
info
.
type
==
TSDB_DATA_TYPE_TIMESTAMP
);
TSKEY
*
tsCol
=
(
TSKEY
*
)
pTsCol
->
pData
;
SColumnInfoData
*
pStartTsCol
=
taosArrayGet
(
pDestBlock
->
pDataBlock
,
START_TS_COLUMN_INDEX
);
SColumnInfoData
*
pEndTsCol
=
taosArrayGet
(
pDestBlock
->
pDataBlock
,
END_TS_COLUMN_INDEX
);
SColumnInfoData
*
pGpCol
=
taosArrayGet
(
pDestBlock
->
pDataBlock
,
GROUPID_COLUMN_INDEX
);
SColumnInfoData
*
pCalStartTsCol
=
taosArrayGet
(
pDestBlock
->
pDataBlock
,
CALCULATE_START_TS_COLUMN_INDEX
);
SColumnInfoData
*
pCalEndTsCol
=
taosArrayGet
(
pDestBlock
->
pDataBlock
,
CALCULATE_END_TS_COLUMN_INDEX
);
uint64_t
groupId
=
getGroupId
(
pInfo
->
pTableScanOp
,
uidCol
[
0
]);
for
(
int32_t
i
=
0
;
i
<
rows
;
)
{
colDataAppend
(
pCalStartTsCol
,
pDestBlock
->
info
.
rows
,
(
const
char
*
)(
tsCol
+
i
),
false
);
STimeWindow
win
=
getSlidingWindow
(
tsCol
,
&
pInfo
->
interval
,
&
pSrcBlock
->
info
,
&
i
);
colDataAppend
(
pCalEndTsCol
,
pDestBlock
->
info
.
rows
,
(
const
char
*
)(
tsCol
+
i
-
1
),
false
);
colDataAppend
(
pStartTsCol
,
pDestBlock
->
info
.
rows
,
(
const
char
*
)(
&
win
.
skey
),
false
);
colDataAppend
(
pEndTsCol
,
pDestBlock
->
info
.
rows
,
(
const
char
*
)(
&
win
.
ekey
),
false
);
colDataAppend
(
pGpCol
,
pDestBlock
->
info
.
rows
,
(
const
char
*
)(
&
groupId
),
false
);
pDestBlock
->
info
.
rows
++
;
}
// all rows have same group id
pDestBlock
->
info
.
groupId
=
groupId
;
return
TSDB_CODE_SUCCESS
;
}
if
(
size
==
0
)
{
generateIntervalTs
(
pInfo
,
pInfo
->
pDeleteDataRes
,
pInfo
->
pTableScanOp
,
pUpdateBlock
);
static
int32_t
generateScanRange
(
SStreamScanInfo
*
pInfo
,
SSDataBlock
*
pSrcBlock
,
SSDataBlock
*
pDestBlock
)
{
int32_t
code
=
TSDB_CODE_SUCCESS
;
if
(
isIntervalWindow
(
pInfo
))
{
code
=
generateIntervalScanRange
(
pInfo
,
pSrcBlock
,
pDestBlock
);
}
else
{
code
=
generateSessionScanRange
(
pInfo
,
pSrcBlock
,
pDestBlock
);
}
pDestBlock
->
info
.
type
=
STREAM_CLEAR
;
blockDataUpdateTsWindow
(
pDestBlock
,
0
);
return
code
;
}
void
appendOneRow
(
SSDataBlock
*
pBlock
,
TSKEY
*
pStartTs
,
TSKEY
*
pEndTs
,
uint64_t
*
pUid
)
{
SColumnInfoData
*
pStartTsCol
=
taosArrayGet
(
pBlock
->
pDataBlock
,
START_TS_COLUMN_INDEX
);
SColumnInfoData
*
pEndTsCol
=
taosArrayGet
(
pBlock
->
pDataBlock
,
END_TS_COLUMN_INDEX
);
SColumnInfoData
*
pUidCol
=
taosArrayGet
(
pBlock
->
pDataBlock
,
UID_COLUMN_INDEX
);
colDataAppend
(
pStartTsCol
,
pBlock
->
info
.
rows
,
(
const
char
*
)
pStartTs
,
false
);
colDataAppend
(
pEndTsCol
,
pBlock
->
info
.
rows
,
(
const
char
*
)
pEndTs
,
false
);
colDataAppend
(
pUidCol
,
pBlock
->
info
.
rows
,
(
const
char
*
)
pUid
,
false
);
pBlock
->
info
.
rows
++
;
}
static
void
checkUpdateData
(
SStreamScanInfo
*
pInfo
,
bool
invertible
,
SSDataBlock
*
pBlock
,
bool
out
)
{
if
(
out
)
{
blockDataCleanup
(
pInfo
->
pUpdateDataRes
);
blockDataEnsureCapacity
(
pInfo
->
pUpdateDataRes
,
pBlock
->
info
.
rows
);
}
SColumnInfoData
*
pColDataInfo
=
taosArrayGet
(
pBlock
->
pDataBlock
,
pInfo
->
primaryTsIndex
);
ASSERT
(
pColDataInfo
->
info
.
type
==
TSDB_DATA_TYPE_TIMESTAMP
);
TSKEY
*
tsCol
=
(
TSKEY
*
)
pColDataInfo
->
pData
;
...
...
@@ -1151,9 +1055,13 @@ static void checkUpdateData(SStreamScanInfo* pInfo, bool invertible, SSDataBlock
// must check update info first.
bool
update
=
updateInfoIsUpdated
(
pInfo
->
pUpdateInfo
,
pBlock
->
info
.
uid
,
tsCol
[
rowId
]);
if
((
update
||
(
isSignleIntervalWindow
(
pInfo
)
&&
isCloseWindow
(
&
win
,
&
pInfo
->
twAggSup
)))
&&
out
)
{
taosArrayPush
(
pInfo
->
tsArray
,
&
rowI
d
);
appendOneRow
(
pInfo
->
pUpdateDataRes
,
tsCol
+
rowId
,
tsCol
+
rowId
,
&
pBlock
->
info
.
ui
d
);
}
}
if
(
out
)
{
blockDataUpdateTsWindow
(
pInfo
->
pUpdateDataRes
,
0
);
pInfo
->
pUpdateDataRes
->
info
.
type
=
STREAM_CLEAR
;
}
}
static
void
setBlockGroupId
(
SOperatorInfo
*
pOperator
,
SSDataBlock
*
pBlock
,
int32_t
uidColIndex
)
{
...
...
@@ -1319,26 +1227,18 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
case
STREAM_RETRIEVE
:
{
pInfo
->
blockType
=
STREAM_INPUT__DATA_SUBMIT
;
pInfo
->
scanMode
=
STREAM_SCAN_FROM_DATAREADER_RETRIEVE
;
copyDataBlock
(
pInfo
->
pPullDataRes
,
pBlock
);
pInfo
->
pullDataResIndex
=
0
;
prepareDataScan
(
pInfo
,
pInfo
->
pPullDataRes
,
START_TS_COLUMN_INDEX
,
&
pInfo
->
pullDataResIndex
);
copyDataBlock
(
pInfo
->
pUpdateRes
,
pBlock
);
prepareRangeScan
(
pInfo
,
pInfo
->
pUpdateRes
,
&
pInfo
->
updateResIndex
);
updateInfoAddCloseWindowSBF
(
pInfo
->
pUpdateInfo
);
}
break
;
case
STREAM_DELETE_DATA
:
{
pInfo
->
blockType
=
STREAM_INPUT__DATA_SUBMIT
;
pInfo
->
updateResIndex
=
0
;
if
(
isIntervalWindow
(
pInfo
))
{
copyDataBlock
(
pInfo
->
pDeleteDataRes
,
pBlock
);
generateIntervalTs
(
pInfo
,
pInfo
->
pDeleteDataRes
,
pInfo
->
pTableScanOp
,
pInfo
->
pUpdateRes
);
prepareDataScan
(
pInfo
,
pInfo
->
pUpdateRes
,
START_TS_COLUMN_INDEX
,
&
pInfo
->
updateResIndex
);
pInfo
->
scanMode
=
STREAM_SCAN_FROM_DATAREADER
;
}
else
{
generateScanRange
(
pInfo
,
pBlock
,
pInfo
->
pTableScanOp
,
pInfo
->
pUpdateRes
);
generateScanRange
(
pInfo
,
pBlock
,
pInfo
->
pUpdateRes
);
prepareRangeScan
(
pInfo
,
pInfo
->
pUpdateRes
,
&
pInfo
->
updateResIndex
);
copyDataBlock
(
pInfo
->
pDeleteDataRes
,
pInfo
->
pUpdateRes
);
pInfo
->
scanMode
=
STREAM_SCAN_FROM_DATAREADER_RANGE
;
}
pInfo
->
pUpdateRes
->
info
.
type
=
STREAM_DELETE_DATA
;
return
pInfo
->
pUpdateRes
;
return
pInfo
->
pDeleteDataRes
;
}
break
;
default:
break
;
...
...
@@ -1351,51 +1251,27 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
pInfo
->
scanMode
=
STREAM_SCAN_FROM_READERHANDLE
;
return
pInfo
->
pRes
;
}
else
if
(
pInfo
->
scanMode
==
STREAM_SCAN_FROM_UPDATERES
)
{
if
(
isStateWindow
(
pInfo
))
{
pInfo
->
scanMode
=
STREAM_SCAN_FROM_READERHANDLE
;
}
else
{
pInfo
->
scanMode
=
STREAM_SCAN_FROM_DATAREADER
;
prepareDataScan
(
pInfo
,
pInfo
->
pUpdateRes
,
pInfo
->
primaryTsIndex
,
&
pInfo
->
updateResIndex
);
}
return
pInfo
->
pUpdateRes
;
}
else
if
(
pInfo
->
scanMode
==
STREAM_SCAN_FROM_DATAREADER_RETRIEVE
)
{
SSDataBlock
*
pSDB
=
doDataScan
(
pInfo
,
pInfo
->
pPullDataRes
,
0
,
&
pInfo
->
pullDataResIndex
);
if
(
pSDB
!=
NULL
)
{
checkUpdateData
(
pInfo
,
true
,
pSDB
,
false
);
pSDB
->
info
.
type
=
STREAM_PULL_DATA
;
return
pSDB
;
}
pInfo
->
scanMode
=
STREAM_SCAN_FROM_DATAREADER
;
}
else
if
(
pInfo
->
scanMode
==
STREAM_SCAN_FROM_DATAREADER
)
{
SSDataBlock
*
pSDB
=
doDataScan
(
pInfo
,
pInfo
->
pUpdateRes
,
pInfo
->
primaryTsIndex
,
&
pInfo
->
updateResIndex
);
if
(
pSDB
)
{
pSDB
->
info
.
type
=
STREAM_NORMAL
;
checkUpdateData
(
pInfo
,
true
,
pSDB
,
false
);
return
pSDB
;
}
setUpdateData
(
pInfo
,
pInfo
->
pRes
,
pInfo
->
pUpdateRes
);
if
(
pInfo
->
pUpdateRes
->
info
.
rows
>
0
)
{
prepareDataScan
(
pInfo
,
pInfo
->
pUpdateRes
,
pInfo
->
primaryTsIndex
,
&
pInfo
->
updateResIndex
);
generateScanRange
(
pInfo
,
pInfo
->
pUpdateDataRes
,
pInfo
->
pUpdateRes
);
pInfo
->
scanMode
=
STREAM_SCAN_FROM_DATAREADER_RANGE
;
prepareRangeScan
(
pInfo
,
pInfo
->
pUpdateRes
,
&
pInfo
->
updateResIndex
);
return
pInfo
->
pUpdateRes
;
}
pInfo
->
scanMode
=
STREAM_SCAN_FROM_READERHANDLE
;
}
else
if
(
pInfo
->
scanMode
==
STREAM_SCAN_FROM_DATAREADER_RANGE
)
{
}
else
if
(
pInfo
->
scanMode
==
STREAM_SCAN_FROM_DATAREADER_RANGE
||
pInfo
->
scanMode
==
STREAM_SCAN_FROM_DATAREADER_RETRIEVE
)
{
SSDataBlock
*
pSDB
=
doRangeScan
(
pInfo
,
pInfo
->
pUpdateRes
,
pInfo
->
primaryTsIndex
,
&
pInfo
->
updateResIndex
);
if
(
pSDB
)
{
pSDB
->
info
.
type
=
STREAM_NORMAL
;
pSDB
->
info
.
type
=
pInfo
->
scanMode
==
STREAM_SCAN_FROM_DATAREADER_RANGE
?
STREAM_NORMAL
:
STREAM_PULL_DATA
;
checkUpdateData
(
pInfo
,
true
,
pSDB
,
false
);
return
pSDB
;
}
pInfo
->
scanMode
=
STREAM_SCAN_FROM_READERHANDLE
;
}
else
if
(
isStateWindow
(
pInfo
))
{
pInfo
->
scanMode
=
STREAM_SCAN_FROM_DATAREADER
;
pInfo
->
updateResIndex
=
pInfo
->
pUpdateRes
->
info
.
rows
;
if
(
prepareDataScan
(
pInfo
,
pInfo
->
pUpdateRes
,
pInfo
->
primaryTsIndex
,
&
pInfo
->
updateResIndex
))
{
blockDataCleanup
(
pInfo
->
pUpdateRes
);
// return empty data blcok
return
pInfo
->
pUpdateRes
;
}
pInfo
->
scanMode
=
STREAM_SCAN_FROM_READERHANDLE
;
if
(
isStateWindow
(
pInfo
)
&&
pInfo
->
sessionSup
.
pStreamAggSup
->
pScanBlock
->
info
.
rows
>
0
)
{
pInfo
->
scanMode
=
STREAM_SCAN_FROM_DATAREADER_RANGE
;
pInfo
->
updateResIndex
=
0
;
copyDataBlock
(
pInfo
->
pUpdateRes
,
pInfo
->
sessionSup
.
pStreamAggSup
->
pScanBlock
);
blockDataCleanup
(
pInfo
->
sessionSup
.
pStreamAggSup
->
pScanBlock
);
prepareRangeScan
(
pInfo
,
pInfo
->
pUpdateRes
,
&
pInfo
->
updateResIndex
);
return
pInfo
->
pUpdateRes
;
}
SDataBlockInfo
*
pBlockInfo
=
&
pInfo
->
pRes
->
info
;
...
...
@@ -1454,15 +1330,14 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
}
else
if
(
pInfo
->
pUpdateInfo
)
{
pInfo
->
tsArrayIndex
=
0
;
checkUpdateData
(
pInfo
,
true
,
pInfo
->
pRes
,
true
);
setUpdateData
(
pInfo
,
pInfo
->
pRes
,
pInfo
->
pUpdateRes
);
pInfo
->
twAggSup
.
maxTs
=
TMAX
(
pInfo
->
twAggSup
.
maxTs
,
pBlockInfo
->
window
.
ekey
);
if
(
pInfo
->
pUpdateRes
->
info
.
rows
>
0
)
{
if
(
pInfo
->
pUpdateRes
->
info
.
type
==
STREAM_CLEAR
)
{
if
(
pInfo
->
pUpdate
Data
Res
->
info
.
rows
>
0
)
{
if
(
pInfo
->
pUpdate
Data
Res
->
info
.
type
==
STREAM_CLEAR
)
{
pInfo
->
updateResIndex
=
0
;
pInfo
->
scanMode
=
STREAM_SCAN_FROM_UPDATERES
;
}
else
if
(
pInfo
->
pUpdateRes
->
info
.
type
==
STREAM_INVERT
)
{
}
else
if
(
pInfo
->
pUpdate
Data
Res
->
info
.
type
==
STREAM_INVERT
)
{
pInfo
->
scanMode
=
STREAM_SCAN_FROM_RES
;
return
pInfo
->
pUpdateRes
;
return
pInfo
->
pUpdate
Data
Res
;
}
}
}
...
...
@@ -1628,17 +1503,18 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
}
pInfo
->
pRes
=
createResDataBlock
(
pDescNode
);
pInfo
->
pUpdateRes
=
create
ResDataBlock
(
pDescNode
);
pInfo
->
pUpdateRes
=
create
SpecialDataBlock
(
STREAM_CLEAR
);
pInfo
->
pCondition
=
pScanPhyNode
->
node
.
pConditions
;
pInfo
->
scanMode
=
STREAM_SCAN_FROM_READERHANDLE
;
pInfo
->
sessionSup
=
(
SessionWindowSupporter
){.
pStreamAggSup
=
NULL
,
.
gap
=
-
1
,
.
parentType
=
QUERY_NODE_PHYSICAL_PLAN
};
pInfo
->
groupId
=
0
;
pInfo
->
pPullDataRes
=
create
PullDataBlock
(
);
pInfo
->
pPullDataRes
=
create
SpecialDataBlock
(
STREAM_RETRIEVE
);
pInfo
->
pStreamScanOp
=
pOperator
;
pInfo
->
deleteDataIndex
=
0
;
pInfo
->
pDeleteDataRes
=
create
PullDataBlock
(
);
pInfo
->
pDeleteDataRes
=
create
SpecialDataBlock
(
STREAM_DELETE_DATA
);
pInfo
->
updateWin
=
(
STimeWindow
){.
skey
=
INT64_MAX
,
.
ekey
=
INT64_MAX
};
pInfo
->
pUpdateDataRes
=
createSpecialDataBlock
(
STREAM_CLEAR
);
pOperator
->
name
=
"StreamScanOperator"
;
pOperator
->
operatorType
=
QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN
;
...
...
source/libs/executor/src/timewindowoperator.c
浏览文件 @
47cb043f
...
...
@@ -1373,8 +1373,10 @@ void doDeleteSpecifyIntervalWindow(SAggSupporter* pAggSup, SSDataBlock* pBlock,
static
void
doClearWindows
(
SAggSupporter
*
pAggSup
,
SExprSupp
*
pSup1
,
SInterval
*
pInterval
,
int32_t
numOfOutput
,
SSDataBlock
*
pBlock
,
SArray
*
pUpWins
)
{
SColumnInfoData
*
pTsCol
=
taosArrayGet
(
pBlock
->
pDataBlock
,
START_TS_COLUMN_INDEX
);
TSKEY
*
tsCols
=
(
TSKEY
*
)
pTsCol
->
pData
;
SColumnInfoData
*
pStartTsCol
=
taosArrayGet
(
pBlock
->
pDataBlock
,
START_TS_COLUMN_INDEX
);
TSKEY
*
startTsCols
=
(
TSKEY
*
)
pStartTsCol
->
pData
;
SColumnInfoData
*
pEndTsCol
=
taosArrayGet
(
pBlock
->
pDataBlock
,
END_TS_COLUMN_INDEX
);
TSKEY
*
endTsCols
=
(
TSKEY
*
)
pEndTsCol
->
pData
;
uint64_t
*
pGpDatas
=
NULL
;
if
(
pBlock
->
info
.
type
==
STREAM_RETRIEVE
)
{
SColumnInfoData
*
pGpCol
=
taosArrayGet
(
pBlock
->
pDataBlock
,
GROUPID_COLUMN_INDEX
);
...
...
@@ -1382,22 +1384,18 @@ static void doClearWindows(SAggSupporter* pAggSup, SExprSupp* pSup1, SInterval*
}
int32_t
step
=
0
;
int32_t
startPos
=
0
;
for
(
int32_t
i
=
0
;
i
<
pBlock
->
info
.
rows
;
i
++
)
{
SResultRowInfo
dumyInfo
;
dumyInfo
.
cur
.
pageId
=
-
1
;
STimeWindow
win
=
getActiveTimeWindow
(
NULL
,
&
dumyInfo
,
tsCols
[
0
],
pInterval
,
TSDB_ORDER_ASC
);
while
(
1
)
{
step
=
getNumOfRowsInTimeWindow
(
&
pBlock
->
info
,
tsCols
,
startPos
,
win
.
ekey
,
binarySearchForKey
,
NULL
,
TSDB_ORDER_ASC
);
STimeWindow
win
=
getActiveTimeWindow
(
NULL
,
&
dumyInfo
,
startTsCols
[
i
],
pInterval
,
TSDB_ORDER_ASC
);
while
(
win
.
ekey
<=
endTsCols
[
i
])
{
uint64_t
winGpId
=
pGpDatas
?
pGpDatas
[
startPos
]
:
pBlock
->
info
.
groupId
;
bool
res
=
doClearWindow
(
pAggSup
,
pSup1
,
(
char
*
)
&
win
.
skey
,
sizeof
(
TSKEY
),
winGpId
,
numOfOutput
);
if
(
pUpWins
&&
res
)
{
SWinRes
winRes
=
{.
ts
=
win
.
skey
,
.
groupId
=
winGpId
};
taosArrayPush
(
pUpWins
,
&
winRes
);
}
int32_t
prevEndPos
=
step
-
1
+
startPos
;
startPos
=
getNextQualifiedWindow
(
pInterval
,
&
win
,
&
pBlock
->
info
,
tsCols
,
prevEndPos
,
TSDB_ORDER_ASC
);
if
(
startPos
<
0
)
{
break
;
getNextTimeWindow
(
pInterval
,
pInterval
->
precision
,
TSDB_ORDER_ASC
,
&
win
);
}
}
}
...
...
@@ -1501,7 +1499,7 @@ static void doBuildDeleteResult(SArray* pWins, int32_t* index, SSDataBlock* pBlo
}
blockDataEnsureCapacity
(
pBlock
,
size
-
*
index
);
SColumnInfoData
*
pTsCol
=
taosArrayGet
(
pBlock
->
pDataBlock
,
START_TS_COLUMN_INDEX
);
SColumnInfoData
*
pGroupCol
=
taosArrayGet
(
pBlock
->
pDataBlock
,
DELETE_
GROUPID_COLUMN_INDEX
);
SColumnInfoData
*
pGroupCol
=
taosArrayGet
(
pBlock
->
pDataBlock
,
GROUPID_COLUMN_INDEX
);
for
(
int32_t
i
=
*
index
;
i
<
size
;
i
++
)
{
SWinRes
*
pWin
=
taosArrayGet
(
pWins
,
i
);
colDataAppend
(
pTsCol
,
pBlock
->
info
.
rows
,
(
const
char
*
)
&
pWin
->
ts
,
false
);
...
...
@@ -1793,10 +1791,7 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo*
pInfo
->
pRecycledPages
=
taosArrayInit
(
4
,
sizeof
(
int32_t
));
pInfo
->
pDelWins
=
taosArrayInit
(
4
,
sizeof
(
SWinRes
));
pInfo
->
delIndex
=
0
;
// pInfo->pDelRes = createPullDataBlock(); todo(liuyao) for delete
pInfo
->
pDelRes
=
createOneDataBlock
(
pInfo
->
binfo
.
pRes
,
false
);
// todo(liuyao) for delete
pInfo
->
pDelRes
->
info
.
type
=
STREAM_DELETE_RESULT
;
// todo(liuyao) for delete
pInfo
->
pDelRes
=
createSpecialDataBlock
(
STREAM_DELETE_RESULT
);
initResultRowInfo
(
&
pInfo
->
binfo
.
resultRowInfo
);
pOperator
->
name
=
"TimeIntervalAggOperator"
;
...
...
@@ -2598,14 +2593,6 @@ static void doHashInterval(SOperatorInfo* pOperatorInfo, SSDataBlock* pSDataBloc
chId
=
getChildIndex
(
pSDataBlock
);
index
=
taosArraySearchIdx
(
chArray
,
&
chId
,
compareInt32Val
,
TD_EQ
);
}
// if (index != -1 && pSDataBlock->info.type == STREAM_PULL_DATA) {
// qDebug("===stream===delete child id %d", chId);
// taosArrayRemove(chArray, index);
// if (taosArrayGetSize(chArray) == 0) {
// // pull data is over
// taosHashRemove(pInfo->pPullDataMap, &winRes, sizeof(SWinRes));
// }
// }
if
(
index
==
-
1
||
pSDataBlock
->
info
.
type
==
STREAM_PULL_DATA
)
{
ignore
=
false
;
}
...
...
@@ -2697,16 +2684,18 @@ static void doBuildPullDataBlock(SArray* array, int32_t* pIndex, SSDataBlock* pB
}
blockDataEnsureCapacity
(
pBlock
,
size
-
(
*
pIndex
));
ASSERT
(
3
<=
taosArrayGetSize
(
pBlock
->
pDataBlock
));
SColumnInfoData
*
pStartTs
=
(
SColumnInfoData
*
)
taosArrayGet
(
pBlock
->
pDataBlock
,
START_TS_COLUMN_INDEX
);
SColumnInfoData
*
pEndTs
=
(
SColumnInfoData
*
)
taosArrayGet
(
pBlock
->
pDataBlock
,
END_TS_COLUMN_INDEX
);
SColumnInfoData
*
pGroupId
=
(
SColumnInfoData
*
)
taosArrayGet
(
pBlock
->
pDataBlock
,
GROUPID_COLUMN_INDEX
);
SColumnInfoData
*
pCalStartTs
=
(
SColumnInfoData
*
)
taosArrayGet
(
pBlock
->
pDataBlock
,
CALCULATE_START_TS_COLUMN_INDEX
);
SColumnInfoData
*
pCalEndTs
=
(
SColumnInfoData
*
)
taosArrayGet
(
pBlock
->
pDataBlock
,
CALCULATE_END_TS_COLUMN_INDEX
);
for
(;
(
*
pIndex
)
<
size
;
(
*
pIndex
)
++
)
{
SPullWindowInfo
*
pWin
=
taosArrayGet
(
array
,
(
*
pIndex
));
SColumnInfoData
*
pStartTs
=
(
SColumnInfoData
*
)
taosArrayGet
(
pBlock
->
pDataBlock
,
START_TS_COLUMN_INDEX
);
colDataAppend
(
pStartTs
,
pBlock
->
info
.
rows
,
(
const
char
*
)
&
pWin
->
window
.
skey
,
false
);
SColumnInfoData
*
pEndTs
=
(
SColumnInfoData
*
)
taosArrayGet
(
pBlock
->
pDataBlock
,
END_TS_COLUMN_INDEX
);
colDataAppend
(
pEndTs
,
pBlock
->
info
.
rows
,
(
const
char
*
)
&
pWin
->
window
.
ekey
,
false
);
SColumnInfoData
*
pGroupId
=
(
SColumnInfoData
*
)
taosArrayGet
(
pBlock
->
pDataBlock
,
GROUPID_COLUMN_INDEX
);
colDataAppend
(
pGroupId
,
pBlock
->
info
.
rows
,
(
const
char
*
)
&
pWin
->
groupId
,
false
);
colDataAppend
(
pCalStartTs
,
pBlock
->
info
.
rows
,
(
const
char
*
)
&
pWin
->
window
.
skey
,
false
);
colDataAppend
(
pCalEndTs
,
pBlock
->
info
.
rows
,
(
const
char
*
)
&
pWin
->
window
.
ekey
,
false
);
pBlock
->
info
.
rows
++
;
}
if
((
*
pIndex
)
==
size
)
{
...
...
@@ -2825,7 +2814,8 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
continue
;
}
removeResults
(
pUpWins
,
pUpdated
);
copyUpdateDataBlock
(
pInfo
->
pUpdateRes
,
pBlock
,
pInfo
->
primaryTsIndex
);
copyDataBlock
(
pInfo
->
pUpdateRes
,
pBlock
);
// copyUpdateDataBlock(pInfo->pUpdateRes, pBlock, pInfo->primaryTsIndex);
pInfo
->
returnUpdate
=
true
;
taosArrayDestroy
(
pUpWins
);
break
;
...
...
@@ -2933,12 +2923,12 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
return
NULL
;
}
SSDataBlock
*
create
PullDataBlock
(
)
{
SSDataBlock
*
create
SpecialDataBlock
(
EStreamType
type
)
{
SSDataBlock
*
pBlock
=
taosMemoryCalloc
(
1
,
sizeof
(
SSDataBlock
));
pBlock
->
info
.
hasVarCol
=
false
;
pBlock
->
info
.
groupId
=
0
;
pBlock
->
info
.
rows
=
0
;
pBlock
->
info
.
type
=
STREAM_RETRIEVE
;
pBlock
->
info
.
type
=
type
;
pBlock
->
info
.
rowSize
=
sizeof
(
TSKEY
)
+
sizeof
(
TSKEY
)
+
sizeof
(
uint64_t
);
pBlock
->
pDataBlock
=
taosArrayInit
(
3
,
sizeof
(
SColumnInfoData
));
...
...
@@ -2952,6 +2942,14 @@ SSDataBlock* createPullDataBlock() {
infoData
.
info
.
type
=
TSDB_DATA_TYPE_UBIGINT
;
infoData
.
info
.
bytes
=
sizeof
(
uint64_t
);
// uid
taosArrayPush
(
pBlock
->
pDataBlock
,
&
infoData
);
// group id
taosArrayPush
(
pBlock
->
pDataBlock
,
&
infoData
);
// calculate start ts
taosArrayPush
(
pBlock
->
pDataBlock
,
&
infoData
);
// calculate end ts
taosArrayPush
(
pBlock
->
pDataBlock
,
&
infoData
);
return
pBlock
;
...
...
@@ -3019,8 +3017,7 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
goto
_error
;
}
}
pInfo
->
pUpdateRes
=
createResDataBlock
(
pPhyNode
->
pOutputDataBlockDesc
);
pInfo
->
pUpdateRes
->
info
.
type
=
STREAM_CLEAR
;
pInfo
->
pUpdateRes
=
createSpecialDataBlock
(
STREAM_CLEAR
);
blockDataEnsureCapacity
(
pInfo
->
pUpdateRes
,
128
);
pInfo
->
returnUpdate
=
false
;
...
...
@@ -3042,11 +3039,9 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
pInfo
->
pullIndex
=
0
;
_hash_fn_t
hashFn
=
taosGetDefaultHashFunction
(
TSDB_DATA_TYPE_BINARY
);
pInfo
->
pPullDataMap
=
taosHashInit
(
64
,
hashFn
,
false
,
HASH_NO_LOCK
);
pInfo
->
pPullDataRes
=
create
PullDataBlock
(
);
pInfo
->
pPullDataRes
=
create
SpecialDataBlock
(
STREAM_RETRIEVE
);
pInfo
->
ignoreExpiredData
=
pIntervalPhyNode
->
window
.
igExpired
;
// pInfo->pDelRes = createPullDataBlock(); // todo(liuyao) for delete
pInfo
->
pDelRes
=
createOneDataBlock
(
pInfo
->
binfo
.
pRes
,
false
);
// todo(liuyao) for delete
pInfo
->
pDelRes
->
info
.
type
=
STREAM_DELETE_RESULT
;
// todo(liuyao) for delete
pInfo
->
pDelRes
=
createSpecialDataBlock
(
STREAM_DELETE_RESULT
);
pInfo
->
delIndex
=
0
;
pInfo
->
pDelWins
=
taosArrayInit
(
4
,
sizeof
(
SWinRes
));
...
...
@@ -3061,7 +3056,9 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
pOperator
->
fpSet
=
createOperatorFpSet
(
NULL
,
doStreamFinalIntervalAgg
,
NULL
,
NULL
,
destroyStreamFinalIntervalOperatorInfo
,
aggEncodeResultRow
,
aggDecodeResultRow
,
NULL
);
if
(
pPhyNode
->
type
==
QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL
)
{
initIntervalDownStream
(
downstream
,
pPhyNode
->
type
);
}
code
=
appendDownstream
(
pOperator
,
&
downstream
,
1
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
goto
_error
;
...
...
@@ -3086,6 +3083,7 @@ void destroyStreamAggSupporter(SStreamAggSupporter* pSup) {
}
taosHashCleanup
(
pSup
->
pResultRows
);
destroyDiskbasedBuf
(
pSup
->
pResultBuf
);
blockDataDestroy
(
pSup
->
pScanBlock
);
}
void
destroyStreamSessionAggOperatorInfo
(
void
*
param
,
int32_t
numOfOutput
)
{
...
...
@@ -3200,7 +3198,7 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh
_hash_fn_t
hashFn
=
taosGetDefaultHashFunction
(
TSDB_DATA_TYPE_BINARY
);
pInfo
->
pStDeleted
=
taosHashInit
(
64
,
hashFn
,
true
,
HASH_NO_LOCK
);
pInfo
->
pDelIterator
=
NULL
;
// pInfo->pDelRes = create
PullDataBlock(
);
// pInfo->pDelRes = create
SpecialDataBlock(STREAM_DELETE_RESULT
);
pInfo
->
pDelRes
=
createOneDataBlock
(
pInfo
->
binfo
.
pRes
,
false
);
// todo(liuyao) for delete
pInfo
->
pDelRes
->
info
.
type
=
STREAM_DELETE_RESULT
;
// todo(liuyao) for delete
pInfo
->
pChildren
=
NULL
;
...
...
@@ -3559,7 +3557,7 @@ static void doDeleteTimeWindows(SStreamAggSupporter* pAggSup, SSDataBlock* pBloc
TSKEY
*
startDatas
=
(
TSKEY
*
)
pStartTsCol
->
pData
;
SColumnInfoData
*
pEndTsCol
=
taosArrayGet
(
pBlock
->
pDataBlock
,
END_TS_COLUMN_INDEX
);
TSKEY
*
endDatas
=
(
TSKEY
*
)
pEndTsCol
->
pData
;
SColumnInfoData
*
pGroupCol
=
taosArrayGet
(
pBlock
->
pDataBlock
,
U
ID_COLUMN_INDEX
);
SColumnInfoData
*
pGroupCol
=
taosArrayGet
(
pBlock
->
pDataBlock
,
GROUP
ID_COLUMN_INDEX
);
uint64_t
*
gpDatas
=
(
uint64_t
*
)
pGroupCol
->
pData
;
for
(
int32_t
i
=
0
;
i
<
pBlock
->
info
.
rows
;
i
++
)
{
int32_t
winIndex
=
0
;
...
...
@@ -4255,7 +4253,6 @@ static void doClearStateWindows(SStreamAggSupporter* pAggSup, SSDataBlock* pBloc
step
=
updateStateWindowInfo
(
pAggSup
->
pCurWins
,
winIndex
,
tsCol
,
pKeyColInfo
,
pBlock
->
info
.
rows
,
i
,
&
allEqual
,
pSeDeleted
);
ASSERT
(
isTsInWindow
(
pCurWin
,
tsCol
[
i
])
||
isEqualStateKey
(
pCurWin
,
pKeyData
));
taosArrayPush
(
pAggSup
->
pScanWindow
,
&
pCurWin
->
winInfo
.
win
);
taosHashRemove
(
pSeUpdated
,
&
pCurWin
->
winInfo
.
pos
,
sizeof
(
SResultRowPosition
));
deleteWindow
(
pAggSup
->
pCurWins
,
winIndex
);
}
...
...
@@ -4282,6 +4279,7 @@ static void doStreamStateAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl
}
SStreamAggSupporter
*
pAggSup
=
&
pInfo
->
streamAggSup
;
blockDataEnsureCapacity
(
pAggSup
->
pScanBlock
,
pSDataBlock
->
info
.
rows
);
SColumnInfoData
*
pKeyColInfo
=
taosArrayGet
(
pSDataBlock
->
pDataBlock
,
pInfo
->
stateCol
.
slotId
);
for
(
int32_t
i
=
0
;
i
<
pSDataBlock
->
info
.
rows
;
i
+=
winRows
)
{
if
(
pInfo
->
ignoreExpiredData
&&
isOverdue
(
tsCols
[
i
],
&
pInfo
->
twAggSup
))
{
...
...
@@ -4296,7 +4294,8 @@ static void doStreamStateAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl
winRows
=
updateStateWindowInfo
(
pAggSup
->
pCurWins
,
winIndex
,
tsCols
,
pKeyColInfo
,
pSDataBlock
->
info
.
rows
,
i
,
&
allEqual
,
pInfo
->
pSeDeleted
);
if
(
!
allEqual
)
{
taosArrayPush
(
pAggSup
->
pScanWindow
,
&
pCurWin
->
winInfo
.
win
);
appendOneRow
(
pAggSup
->
pScanBlock
,
&
pCurWin
->
winInfo
.
win
.
skey
,
&
pCurWin
->
winInfo
.
win
.
ekey
,
&
pSDataBlock
->
info
.
groupId
);
taosHashRemove
(
pSeUpdated
,
&
pCurWin
->
winInfo
.
pos
,
sizeof
(
SResultRowPosition
));
deleteWindow
(
pAggSup
->
pCurWins
,
winIndex
);
continue
;
...
...
@@ -4460,7 +4459,7 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys
_hash_fn_t
hashFn
=
taosGetDefaultHashFunction
(
TSDB_DATA_TYPE_BINARY
);
pInfo
->
pSeDeleted
=
taosHashInit
(
64
,
hashFn
,
true
,
HASH_NO_LOCK
);
pInfo
->
pDelIterator
=
NULL
;
// pInfo->pDelRes = create
PullDataBlock(); // todo(liuyao) for delete
// pInfo->pDelRes = create
SpecialDataBlock(STREAM_DELETE_RESULT);
pInfo
->
pDelRes
=
createOneDataBlock
(
pInfo
->
binfo
.
pRes
,
false
);
// todo(liuyao) for delete
pInfo
->
pDelRes
->
info
.
type
=
STREAM_DELETE_RESULT
;
// todo(liuyao) for delete
pInfo
->
pChildren
=
NULL
;
...
...
source/libs/function/src/builtinsimpl.c
浏览文件 @
47cb043f
...
...
@@ -476,16 +476,16 @@ int32_t functionFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
int32_t
firstCombine
(
SqlFunctionCtx
*
pDestCtx
,
SqlFunctionCtx
*
pSourceCtx
)
{
SResultRowEntryInfo
*
pDResInfo
=
GET_RES_INFO
(
pDestCtx
);
char
*
pDBuf
=
GET_ROWCELL_INTERBUF
(
pDResInfo
);
SFirstLastRes
*
pDBuf
=
GET_ROWCELL_INTERBUF
(
pDResInfo
);
int32_t
type
=
pDestCtx
->
input
.
pData
[
0
]
->
info
.
type
;
int32_t
bytes
=
pDestCtx
->
input
.
pData
[
0
]
->
info
.
bytes
;
SResultRowEntryInfo
*
pSResInfo
=
GET_RES_INFO
(
pSourceCtx
);
char
*
pSBuf
=
GET_ROWCELL_INTERBUF
(
pSResInfo
);
SFirstLastRes
*
pSBuf
=
GET_ROWCELL_INTERBUF
(
pSResInfo
);
if
(
pSResInfo
->
numOfRes
!=
0
&&
(
pDResInfo
->
numOfRes
==
0
||
*
(
TSKEY
*
)(
pDBuf
+
bytes
)
>
*
(
TSKEY
*
)(
pSBuf
+
bytes
)
))
{
memcpy
(
pDBuf
,
pSB
uf
,
bytes
);
*
(
TSKEY
*
)(
pDBuf
+
bytes
)
=
*
(
TSKEY
*
)(
pSBuf
+
bytes
)
;
if
(
pSResInfo
->
numOfRes
!=
0
&&
(
pDResInfo
->
numOfRes
==
0
||
pDBuf
->
ts
>
pSBuf
->
ts
))
{
memcpy
(
pDBuf
->
buf
,
pSBuf
->
b
uf
,
bytes
);
pDBuf
->
ts
=
pSBuf
->
ts
;
pDResInfo
->
numOfRes
=
1
;
}
return
TSDB_CODE_SUCCESS
;
...
...
@@ -2994,16 +2994,16 @@ int32_t firstLastPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
// todo rewrite:
int32_t
lastCombine
(
SqlFunctionCtx
*
pDestCtx
,
SqlFunctionCtx
*
pSourceCtx
)
{
SResultRowEntryInfo
*
pDResInfo
=
GET_RES_INFO
(
pDestCtx
);
char
*
pDBuf
=
GET_ROWCELL_INTERBUF
(
pDResInfo
);
SFirstLastRes
*
pDBuf
=
GET_ROWCELL_INTERBUF
(
pDResInfo
);
int32_t
type
=
pDestCtx
->
input
.
pData
[
0
]
->
info
.
type
;
int32_t
bytes
=
pDestCtx
->
input
.
pData
[
0
]
->
info
.
bytes
;
SResultRowEntryInfo
*
pSResInfo
=
GET_RES_INFO
(
pSourceCtx
);
char
*
pSBuf
=
GET_ROWCELL_INTERBUF
(
pSResInfo
);
SFirstLastRes
*
pSBuf
=
GET_ROWCELL_INTERBUF
(
pSResInfo
);
if
(
pSResInfo
->
numOfRes
!=
0
&&
(
pDResInfo
->
numOfRes
==
0
||
*
(
TSKEY
*
)(
pDBuf
+
bytes
)
<
*
(
TSKEY
*
)(
pSBuf
+
bytes
)
))
{
memcpy
(
pDBuf
,
pSB
uf
,
bytes
);
*
(
TSKEY
*
)(
pDBuf
+
bytes
)
=
*
(
TSKEY
*
)(
pSBuf
+
bytes
)
;
if
(
pSResInfo
->
numOfRes
!=
0
&&
(
pDResInfo
->
numOfRes
==
0
||
pDBuf
->
ts
<
pSBuf
->
ts
))
{
memcpy
(
pDBuf
->
buf
,
pSBuf
->
b
uf
,
bytes
);
pDBuf
->
ts
=
pSBuf
->
ts
;
pDResInfo
->
numOfRes
=
1
;
}
return
TSDB_CODE_SUCCESS
;
...
...
tests/script/tsim/stream/sliding.sim
浏览文件 @
47cb043f
...
...
@@ -367,7 +367,7 @@ if $data32 != 8 then
endi
#$loop_all = 0
#looptest:
#
=
looptest:
sql drop database IF EXISTS test2;
sql drop stream IF EXISTS streams21;
...
...
@@ -511,6 +511,6 @@ endi
$loop_all = $loop_all + 1
print ============loop_all=$loop_all
#goto looptest
#
=
goto looptest
system sh/stop_dnodes.sh
\ No newline at end of file
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录