Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
875989aa
TDengine
项目概览
taosdata
/
TDengine
12 个月 前同步成功
通知
1181
Star
22014
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
875989aa
编写于
11月 18, 2022
作者:
H
Haojun Liao
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
refactor: do some internal refactor.
上级
2055879d
变更
3
隐藏空白更改
内联
并排
Showing
3 changed file
with
13 addition
and
72 deletion
+13
-72
source/libs/executor/inc/executorimpl.h
source/libs/executor/inc/executorimpl.h
+4
-6
source/libs/executor/src/executorimpl.c
source/libs/executor/src/executorimpl.c
+1
-56
source/libs/executor/src/scanoperator.c
source/libs/executor/src/scanoperator.c
+8
-10
未找到文件。
source/libs/executor/inc/executorimpl.h
浏览文件 @
875989aa
...
...
@@ -163,7 +163,7 @@ typedef struct {
SArray
*
pStopInfo
;
}
STaskStopInfo
;
typedef
struct
SExecTaskInfo
{
struct
SExecTaskInfo
{
STaskIdInfo
id
;
uint32_t
status
;
STimeWindow
window
;
...
...
@@ -182,7 +182,7 @@ typedef struct SExecTaskInfo {
struct
SOperatorInfo
*
pRoot
;
SLocalFetch
localFetch
;
STaskStopInfo
stopInfo
;
}
SExecTaskInfo
;
};
enum
{
OP_NOT_OPENED
=
0x0
,
...
...
@@ -343,7 +343,6 @@ typedef struct STableScanInfo {
}
STableScanInfo
;
typedef
struct
STableMergeScanInfo
{
STableListInfo
*
tableListInfo
;
int32_t
tableStartIndex
;
int32_t
tableEndIndex
;
bool
hasGroupId
;
...
...
@@ -363,7 +362,6 @@ typedef struct STableMergeScanInfo {
int32_t
scanTimes
;
SSDataBlock
*
pResBlock
;
int32_t
numOfOutput
;
int32_t
dataBlockLoadFlag
;
// if the upstream is an interval operator, the interval info is also kept here to get the time
// window to check if current data block needs to be loaded.
...
...
@@ -1041,8 +1039,8 @@ int32_t finalizeResultRows(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPos
SOperatorInfo
*
createGroupSortOperatorInfo
(
SOperatorInfo
*
downstream
,
SGroupSortPhysiNode
*
pSortPhyNode
,
SExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createTableMergeScanOperatorInfo
(
STableScanPhysiNode
*
pTableScanNode
,
S
TableListInfo
*
pTableListInfo
,
S
ReadHandle
*
readHandle
,
S
ExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createTableMergeScanOperatorInfo
(
STableScanPhysiNode
*
pTableScanNode
,
S
ReadHandle
*
readHandle
,
SExecTaskInfo
*
pTaskInfo
);
void
copyUpdateDataBlock
(
SSDataBlock
*
pDest
,
SSDataBlock
*
pSource
,
int32_t
tsColIndex
);
...
...
source/libs/executor/src/executorimpl.c
浏览文件 @
875989aa
...
...
@@ -1000,12 +1000,6 @@ int32_t loadDataBlockOnDemand(SExecTaskInfo* pTaskInfo, STableScanInfo* pTableSc
return
TSDB_CODE_SUCCESS
;
}
static
void
updateTableQueryInfoForReverseScan
(
STableQueryInfo
*
pTableQueryInfo
)
{
if
(
pTableQueryInfo
==
NULL
)
{
return
;
}
}
void
setTaskStatus
(
SExecTaskInfo
*
pTaskInfo
,
int8_t
status
)
{
if
(
status
==
TASK_NOT_COMPLETED
)
{
pTaskInfo
->
status
=
status
;
...
...
@@ -1665,55 +1659,6 @@ int32_t appendDownstream(SOperatorInfo* p, SOperatorInfo** pDownstream, int32_t
static
int32_t
doInitAggInfoSup
(
SAggSupporter
*
pAggSup
,
SqlFunctionCtx
*
pCtx
,
int32_t
numOfOutput
,
size_t
keyBufSize
,
const
char
*
pKey
);
static
bool
needToMerge
(
SSDataBlock
*
pBlock
,
SArray
*
groupInfo
,
char
**
buf
,
int32_t
rowIndex
)
{
size_t
size
=
taosArrayGetSize
(
groupInfo
);
if
(
size
==
0
)
{
return
true
;
}
for
(
int32_t
i
=
0
;
i
<
size
;
++
i
)
{
int32_t
*
index
=
taosArrayGet
(
groupInfo
,
i
);
SColumnInfoData
*
pColInfo
=
taosArrayGet
(
pBlock
->
pDataBlock
,
*
index
);
bool
isNull
=
colDataIsNull
(
pColInfo
,
rowIndex
,
pBlock
->
info
.
rows
,
NULL
);
if
((
isNull
&&
buf
[
i
]
!=
NULL
)
||
(
!
isNull
&&
buf
[
i
]
==
NULL
))
{
return
false
;
}
char
*
pCell
=
colDataGetData
(
pColInfo
,
rowIndex
);
if
(
IS_VAR_DATA_TYPE
(
pColInfo
->
info
.
type
))
{
if
(
varDataLen
(
pCell
)
!=
varDataLen
(
buf
[
i
]))
{
return
false
;
}
else
{
if
(
memcmp
(
varDataVal
(
pCell
),
varDataVal
(
buf
[
i
]),
varDataLen
(
pCell
))
!=
0
)
{
return
false
;
}
}
}
else
{
if
(
memcmp
(
pCell
,
buf
[
i
],
pColInfo
->
info
.
bytes
)
!=
0
)
{
return
false
;
}
}
}
return
0
;
}
static
bool
saveCurrentTuple
(
char
**
rowColData
,
SArray
*
pColumnList
,
SSDataBlock
*
pBlock
,
int32_t
rowIndex
)
{
int32_t
size
=
(
int32_t
)
taosArrayGetSize
(
pColumnList
);
for
(
int32_t
i
=
0
;
i
<
size
;
++
i
)
{
int32_t
*
index
=
taosArrayGet
(
pColumnList
,
i
);
SColumnInfoData
*
pColInfo
=
taosArrayGet
(
pBlock
->
pDataBlock
,
*
index
);
char
*
data
=
colDataGetData
(
pColInfo
,
rowIndex
);
memcpy
(
rowColData
[
i
],
data
,
colDataGetLength
(
pColInfo
,
rowIndex
));
}
return
true
;
}
int32_t
getTableScanInfo
(
SOperatorInfo
*
pOperator
,
int32_t
*
order
,
int32_t
*
scanFlag
)
{
// todo add more information about exchange operation
int32_t
type
=
pOperator
->
operatorType
;
...
...
@@ -2762,7 +2707,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
return
NULL
;
}
pOperator
=
createTableMergeScanOperatorInfo
(
pTableScanNode
,
p
TableListInfo
,
p
Handle
,
pTaskInfo
);
pOperator
=
createTableMergeScanOperatorInfo
(
pTableScanNode
,
pHandle
,
pTaskInfo
);
if
(
NULL
==
pOperator
)
{
pTaskInfo
->
code
=
terrno
;
return
NULL
;
...
...
source/libs/executor/src/scanoperator.c
浏览文件 @
875989aa
...
...
@@ -4392,7 +4392,7 @@ static SSDataBlock* getTableDataBlockImpl(void* param) {
int64_t
st
=
taosGetTimestampUs
();
void
*
p
=
tableListGetInfo
(
p
Info
->
tableListInfo
,
readIdx
+
pInfo
->
tableStartIndex
);
void
*
p
=
tableListGetInfo
(
p
TaskInfo
->
pTableInfoList
,
readIdx
+
pInfo
->
tableStartIndex
);
SReadHandle
*
pHandle
=
&
pInfo
->
base
.
readHandle
;
int32_t
code
=
tsdbReaderOpen
(
pHandle
->
vnode
,
pQueryCond
,
p
,
1
,
&
pInfo
->
base
.
dataReader
,
GET_TASKID
(
pTaskInfo
));
...
...
@@ -4486,10 +4486,10 @@ int32_t startGroupTableMergeScan(SOperatorInfo* pOperator) {
SExecTaskInfo
*
pTaskInfo
=
pOperator
->
pTaskInfo
;
{
size_t
numOfTables
=
tableListGetSize
(
p
Info
->
tableListInfo
);
size_t
numOfTables
=
tableListGetSize
(
p
TaskInfo
->
pTableInfoList
);
int32_t
i
=
pInfo
->
tableStartIndex
+
1
;
for
(;
i
<
numOfTables
;
++
i
)
{
STableKeyInfo
*
tableKeyInfo
=
tableListGetInfo
(
p
Info
->
tableListInfo
,
i
);
STableKeyInfo
*
tableKeyInfo
=
tableListGetInfo
(
p
TaskInfo
->
pTableInfoList
,
i
);
if
(
tableKeyInfo
->
groupId
!=
pInfo
->
groupId
)
{
break
;
}
...
...
@@ -4613,7 +4613,7 @@ SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) {
T_LONG_JMP
(
pTaskInfo
->
env
,
code
);
}
size_t
tableListSize
=
tableListGetSize
(
p
Info
->
tableListInfo
);
size_t
tableListSize
=
tableListGetSize
(
p
TaskInfo
->
pTableInfoList
);
if
(
!
pInfo
->
hasGroupId
)
{
pInfo
->
hasGroupId
=
true
;
...
...
@@ -4622,7 +4622,7 @@ SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) {
return
NULL
;
}
pInfo
->
tableStartIndex
=
0
;
pInfo
->
groupId
=
((
STableKeyInfo
*
)
tableListGetInfo
(
p
Info
->
tableListInfo
,
pInfo
->
tableStartIndex
))
->
groupId
;
pInfo
->
groupId
=
((
STableKeyInfo
*
)
tableListGetInfo
(
p
TaskInfo
->
pTableInfoList
,
pInfo
->
tableStartIndex
))
->
groupId
;
startGroupTableMergeScan
(
pOperator
);
}
...
...
@@ -4641,7 +4641,7 @@ SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) {
break
;
}
pInfo
->
tableStartIndex
=
pInfo
->
tableEndIndex
+
1
;
pInfo
->
groupId
=
tableListGetInfo
(
p
Info
->
tableListInfo
,
pInfo
->
tableStartIndex
)
->
groupId
;
pInfo
->
groupId
=
tableListGetInfo
(
p
TaskInfo
->
pTableInfoList
,
pInfo
->
tableStartIndex
)
->
groupId
;
startGroupTableMergeScan
(
pOperator
);
}
}
...
...
@@ -4698,8 +4698,8 @@ int32_t getTableMergeScanExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExpla
return
TSDB_CODE_SUCCESS
;
}
SOperatorInfo
*
createTableMergeScanOperatorInfo
(
STableScanPhysiNode
*
pTableScanNode
,
S
TableListInfo
*
pTableListInfo
,
S
ReadHandle
*
readHandle
,
S
ExecTaskInfo
*
pTaskInfo
)
{
SOperatorInfo
*
createTableMergeScanOperatorInfo
(
STableScanPhysiNode
*
pTableScanNode
,
S
ReadHandle
*
readHandle
,
SExecTaskInfo
*
pTaskInfo
)
{
STableMergeScanInfo
*
pInfo
=
taosMemoryCalloc
(
1
,
sizeof
(
STableMergeScanInfo
));
SOperatorInfo
*
pOperator
=
taosMemoryCalloc
(
1
,
sizeof
(
SOperatorInfo
));
if
(
pInfo
==
NULL
||
pOperator
==
NULL
)
{
...
...
@@ -4733,14 +4733,12 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN
pInfo
->
interval
=
extractIntervalInfo
(
pTableScanNode
);
pInfo
->
sample
.
sampleRatio
=
pTableScanNode
->
ratio
;
pInfo
->
sample
.
seed
=
taosGetTimestampSec
();
pInfo
->
dataBlockLoadFlag
=
pTableScanNode
->
dataRequired
;
code
=
filterInitFromNode
((
SNode
*
)
pTableScanNode
->
scan
.
node
.
pConditions
,
&
pOperator
->
exprSupp
.
pFilterInfo
,
0
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
goto
_error
;
}
pInfo
->
tableListInfo
=
pTableListInfo
;
pInfo
->
base
.
scanFlag
=
MAIN_SCAN
;
initResultSizeInfo
(
&
pOperator
->
resultInfo
,
1024
);
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录