Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
f23abec3
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看板
未验证
提交
f23abec3
编写于
6月 18, 2022
作者:
H
Haojun Liao
提交者:
GitHub
6月 18, 2022
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #13957 from taosdata/feature/3_liaohj
refactor: do some internal refactor.
上级
d4f5dc86
956173e2
变更
11
展开全部
隐藏空白更改
内联
并排
Showing
11 changed file
with
539 addition
and
729 deletion
+539
-729
include/libs/function/function.h
include/libs/function/function.h
+4
-4
source/libs/executor/inc/executil.h
source/libs/executor/inc/executil.h
+2
-2
source/libs/executor/inc/executorimpl.h
source/libs/executor/inc/executorimpl.h
+36
-110
source/libs/executor/src/executil.c
source/libs/executor/src/executil.c
+5
-20
source/libs/executor/src/executorimpl.c
source/libs/executor/src/executorimpl.c
+141
-292
source/libs/executor/src/groupoperator.c
source/libs/executor/src/groupoperator.c
+42
-36
source/libs/executor/src/joinoperator.c
source/libs/executor/src/joinoperator.c
+4
-4
source/libs/executor/src/scanoperator.c
source/libs/executor/src/scanoperator.c
+20
-19
source/libs/executor/src/sortoperator.c
source/libs/executor/src/sortoperator.c
+5
-5
source/libs/executor/src/timewindowoperator.c
source/libs/executor/src/timewindowoperator.c
+251
-208
tests/script/tsim/testsuit.sim
tests/script/tsim/testsuit.sim
+29
-29
未找到文件。
include/libs/function/function.h
浏览文件 @
f23abec3
...
...
@@ -158,10 +158,10 @@ typedef struct tExprNode {
int32_t
nodeType
;
union
{
struct
{
// function node
char
functionName
[
FUNCTIONS_NAME_MAX_LENGTH
];
// todo refactor
int32_t
functionId
;
int32_t
num
;
struct
SFunctionNode
*
pFunctNode
;
char
functionName
[
FUNCTIONS_NAME_MAX_LENGTH
];
// todo refactor
int32_t
functionId
;
int32_t
num
;
struct
SFunctionNode
*
pFunctNode
;
}
_function
;
struct
{
...
...
source/libs/executor/inc/executil.h
浏览文件 @
f23abec3
...
...
@@ -113,7 +113,7 @@ SArray* extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNod
SExprInfo
*
createExprInfo
(
SNodeList
*
pNodeList
,
SNodeList
*
pGroupKeys
,
int32_t
*
numOfExprs
);
SqlFunctionCtx
*
createSqlFunctionCtx
(
SExprInfo
*
pExprInfo
,
int32_t
numOfOutput
,
int32_t
**
row
Cell
InfoOffset
);
SqlFunctionCtx
*
createSqlFunctionCtx
(
SExprInfo
*
pExprInfo
,
int32_t
numOfOutput
,
int32_t
**
row
Entry
InfoOffset
);
void
relocateColumnData
(
SSDataBlock
*
pBlock
,
const
SArray
*
pColMatchInfo
,
SArray
*
pCols
);
void
initExecTimeWindowInfo
(
SColumnInfoData
*
pColData
,
STimeWindow
*
pQueryWindow
);
...
...
@@ -121,6 +121,6 @@ SInterval extractIntervalInfo(const STableScanPhysiNode* pTableScanNode);
SColumn
extractColumnFromColumnNode
(
SColumnNode
*
pColNode
);
int32_t
initQueryTableDataCond
(
SQueryTableDataCond
*
pCond
,
const
STableScanPhysiNode
*
pTableScanNode
);
void
cleanupQueryTableDataCond
(
SQueryTableDataCond
*
pCond
);
void
cleanupQueryTableDataCond
(
SQueryTableDataCond
*
pCond
);
#endif // TDENGINE_QUERYUTIL_H
source/libs/executor/inc/executorimpl.h
浏览文件 @
f23abec3
...
...
@@ -49,8 +49,6 @@ typedef int32_t (*__block_search_fn_t)(char* data, int32_t num, int64_t key, int
#define Q_STATUS_EQUAL(p, s) (((p) & (s)) != 0u)
#define QUERY_IS_ASC_QUERY(q) (GET_FORWARD_DIRECTION_FACTOR((q)->order.order) == QUERY_ASC_FORWARD_STEP)
//#define GET_TABLEGROUP(q, _index) ((SArray*)taosArrayGetP((q)->tableqinfoGroupInfo.pGroupList, (_index)))
#define NEEDTO_COMPRESS_QUERY(size) ((size) > tsCompressColData ? 1 : 0)
enum
{
...
...
@@ -64,11 +62,6 @@ enum {
TASK_COMPLETED
=
0x2u
,
};
typedef
struct
SResultRowCell
{
uint64_t
groupId
;
SResultRowPosition
pos
;
}
SResultRowCell
;
/**
* If the number of generated results is greater than this value,
* query query will be halt and return results to client immediate.
...
...
@@ -83,7 +76,6 @@ typedef struct SResultInfo { // TODO refactor
typedef
struct
STableQueryInfo
{
TSKEY
lastKey
;
// last check ts, todo remove it later
SResultRowPosition
pos
;
// current active time window
// SVariant tag;
}
STableQueryInfo
;
typedef
struct
SLimit
{
...
...
@@ -123,41 +115,7 @@ typedef struct SOperatorCostInfo {
double
totalCost
;
}
SOperatorCostInfo
;
// The basic query information extracted from the SQueryInfo tree to support the
// execution of query in a data node.
typedef
struct
STaskAttr
{
SLimit
limit
;
SLimit
slimit
;
bool
stableQuery
;
// super table query or not
bool
topBotQuery
;
// TODO used bitwise flag
bool
groupbyColumn
;
// denote if this is a groupby normal column query
bool
timeWindowInterpo
;
// if the time window start/end required interpolation
bool
tsCompQuery
;
// is tscomp query
bool
diffQuery
;
// is diff query
bool
pointInterpQuery
;
// point interpolation query
int32_t
havingNum
;
// having expr number
int16_t
numOfCols
;
int16_t
numOfTags
;
STimeWindow
window
;
SInterval
interval
;
int16_t
precision
;
int16_t
numOfOutput
;
int16_t
fillType
;
int32_t
resultRowSize
;
int32_t
tagLen
;
// tag value length of current query
SExprInfo
*
pExpr1
;
SColumnInfo
*
tagColList
;
int32_t
numOfFilterCols
;
int64_t
*
fillVal
;
void
*
tsdb
;
// STableListInfo tableGroupInfo; // table list
int32_t
vgId
;
}
STaskAttr
;
struct
SOperatorInfo
;
//struct SAggSupporter;
//struct SOptrBasicInfo;
typedef
int32_t
(
*
__optr_encode_fn_t
)(
struct
SOperatorInfo
*
pOperator
,
char
**
result
,
int32_t
*
length
);
typedef
int32_t
(
*
__optr_decode_fn_t
)(
struct
SOperatorInfo
*
pOperator
,
char
*
result
);
...
...
@@ -195,31 +153,6 @@ typedef struct SExecTaskInfo {
struct
SOperatorInfo
*
pRoot
;
}
SExecTaskInfo
;
typedef
struct
STaskRuntimeEnv
{
STaskAttr
*
pQueryAttr
;
uint32_t
status
;
// query status
uint8_t
scanFlag
;
// denotes reversed scan of data or not
SDiskbasedBuf
*
pResultBuf
;
// query result buffer based on blocked-wised disk file
SHashObj
*
pResultRowHashTable
;
// quick locate the window object for each result
SHashObj
*
pResultRowListSet
;
// used to check if current ResultRowInfo has ResultRow object or not
SArray
*
pResultRowArrayList
;
// The array list that contains the Result rows
char
*
keyBuf
;
// window key buffer
// The window result objects pool, all the resultRow Objects are allocated and managed by this object.
char
**
prevRow
;
STSBuf
*
pTsBuf
;
// timestamp filter list
STSCursor
cur
;
char
*
tagVal
;
// tag value of current data block
// STableGroupInfo tableqinfoGroupInfo; // this is a table list
struct
SOperatorInfo
*
proot
;
SGroupResInfo
groupResInfo
;
int64_t
currentOffset
;
// dynamic offset value
STableQueryInfo
*
current
;
SResultInfo
resultInfo
;
struct
SUdfInfo
*
pUdfInfo
;
}
STaskRuntimeEnv
;
enum
{
OP_NOT_OPENED
=
0x0
,
OP_OPENED
=
0x1
,
...
...
@@ -238,14 +171,20 @@ typedef struct SOperatorFpSet {
__optr_explain_fn_t
getExplainFn
;
}
SOperatorFpSet
;
typedef
struct
SExprSupp
{
SExprInfo
*
pExprInfo
;
int32_t
numOfExprs
;
// the number of scalar expression in group operator
SqlFunctionCtx
*
pCtx
;
int32_t
*
rowEntryInfoOffset
;
// offset value for each row result cell info
}
SExprSupp
;
typedef
struct
SOperatorInfo
{
uint8_t
operatorType
;
bool
blocking
;
// block operator or not
uint8_t
status
;
// denote if current operator is completed
int32_t
numOfExprs
;
// number of columns of the current operator results
char
*
name
;
// name, used to show the query execution plan
char
*
name
;
// name, for debug purpose
void
*
info
;
// extension attribution
SExpr
Info
*
pExpr
;
SExpr
Supp
exprSupp
;
SExecTaskInfo
*
pTaskInfo
;
SOperatorCostInfo
cost
;
SResultInfo
resultInfo
;
...
...
@@ -260,6 +199,9 @@ typedef enum {
EX_SOURCE_DATA_EXHAUSTED
=
0x3
,
}
EX_SOURCE_STATUS
;
#define COL_MATCH_FROM_COL_ID 0x1
#define COL_MATCH_FROM_SLOT_ID 0x2
typedef
struct
SSourceDataInfo
{
int32_t
index
;
SRetrieveTableRsp
*
pRsp
;
...
...
@@ -287,9 +229,6 @@ typedef struct SExchangeInfo {
uint64_t
self
;
}
SExchangeInfo
;
#define COL_MATCH_FROM_COL_ID 0x1
#define COL_MATCH_FROM_SLOT_ID 0x2
typedef
struct
SColMatchInfo
{
int32_t
srcSlotId
;
// source slot id
int32_t
colId
;
...
...
@@ -299,8 +238,8 @@ typedef struct SColMatchInfo {
}
SColMatchInfo
;
typedef
struct
SScanInfo
{
int32_t
numOfAsc
;
int32_t
numOfDesc
;
int32_t
numOfAsc
;
int32_t
numOfDesc
;
}
SScanInfo
;
typedef
struct
SSampleExecInfo
{
...
...
@@ -320,17 +259,13 @@ typedef struct STableScanInfo {
SNode
*
pFilterNode
;
// filter info, which is push down by optimizer
SqlFunctionCtx
*
pCtx
;
// which belongs to the direct upstream operator operator query context
SResultRowInfo
*
pResultRowInfo
;
int32_t
*
row
Cell
InfoOffset
;
int32_t
*
row
Entry
InfoOffset
;
SExprInfo
*
pExpr
;
SSDataBlock
*
pResBlock
;
SArray
*
pColMatchInfo
;
int32_t
numOfOutput
;
SExprInfo
*
pPseudoExpr
;
int32_t
numOfPseudoExpr
;
SqlFunctionCtx
*
pPseudoCtx
;
// int32_t* rowCellInfoOffset;
SExprSupp
pseudoSup
;
SQueryTableDataCond
cond
;
int32_t
scanFlag
;
// table scan flag to denote if it is a repeat/reverse/main scan
int32_t
dataBlockLoadFlag
;
...
...
@@ -436,14 +371,12 @@ typedef struct SBlockDistInfo {
void
*
pHandle
;
}
SBlockDistInfo
;
// todo remove this
typedef
struct
SOptrBasicInfo
{
SResultRowInfo
resultRowInfo
;
int32_t
*
rowCellInfoOffset
;
// offset value for each row result cell info
SqlFunctionCtx
*
pCtx
;
SSDataBlock
*
pRes
;
}
SOptrBasicInfo
;
// TODO move the resultrowsiz together with SOptrBasicInfo:rowCellInfoOffset
typedef
struct
SAggSupporter
{
SHashObj
*
pResultRowHashTable
;
// quick locate the window object for each result
char
*
keyBuf
;
// window key buffer
...
...
@@ -500,10 +433,7 @@ typedef struct SAggOperatorInfo {
STableQueryInfo
*
current
;
uint64_t
groupId
;
SGroupResInfo
groupResInfo
;
SExprInfo
*
pScalarExprInfo
;
int32_t
numOfScalarExpr
;
// the number of scalar expression before the aggregate function can be applied
SqlFunctionCtx
*
pScalarCtx
;
// scalar function requried sql function struct.
int32_t
*
rowCellInfoOffset
;
// offset value for each row result cell info
SExprSupp
scalarExprSup
;
}
SAggOperatorInfo
;
typedef
struct
SProjectOperatorInfo
{
...
...
@@ -528,11 +458,7 @@ typedef struct SIndefOperatorInfo {
SOptrBasicInfo
binfo
;
SAggSupporter
aggSup
;
SArray
*
pPseudoColInfo
;
SExprInfo
*
pScalarExpr
;
int32_t
numOfScalarExpr
;
SqlFunctionCtx
*
pScalarCtx
;
int32_t
*
rowCellInfoOffset
;
SExprSupp
scalarSup
;
}
SIndefOperatorInfo
;
typedef
struct
SFillOperatorInfo
{
...
...
@@ -544,13 +470,6 @@ typedef struct SFillOperatorInfo {
bool
multigroupResult
;
}
SFillOperatorInfo
;
typedef
struct
SScalarSupp
{
SExprInfo
*
pScalarExprInfo
;
int32_t
numOfScalarExpr
;
// the number of scalar expression in group operator
SqlFunctionCtx
*
pScalarFuncCtx
;
int32_t
*
rowCellInfoOffset
;
// offset value for each row result cell info
}
SScalarSupp
;
typedef
struct
SGroupbyOperatorInfo
{
// SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
SOptrBasicInfo
binfo
;
...
...
@@ -563,7 +482,7 @@ typedef struct SGroupbyOperatorInfo {
char
*
keyBuf
;
// group by keys for hash
int32_t
groupKeyLen
;
// total group by column width
SGroupResInfo
groupResInfo
;
S
ScalarSupp
scalarSup
;
S
ExprSupp
scalarSup
;
}
SGroupbyOperatorInfo
;
typedef
struct
SDataGroupInfo
{
...
...
@@ -587,7 +506,7 @@ typedef struct SPartitionOperatorInfo {
void
*
pGroupIter
;
// group iterator
int32_t
pageIndex
;
// page index of current group
SSDataBlock
*
pUpdateRes
;
S
ScalarSupp
scalarSup
p
;
S
ExprSupp
scalarSu
p
;
}
SPartitionOperatorInfo
;
typedef
struct
SWindowRowsSup
{
...
...
@@ -740,15 +659,19 @@ SOperatorFpSet createOperatorFpSet(__optr_open_fn_t openFn, __optr_fn_t nextFn,
int32_t
operatorDummyOpenFn
(
SOperatorInfo
*
pOperator
);
void
operatorDummyCloseFn
(
void
*
param
,
int32_t
numOfCols
);
int32_t
appendDownstream
(
SOperatorInfo
*
p
,
SOperatorInfo
**
pDownstream
,
int32_t
num
);
int32_t
initAggInfo
(
SOptrBasicInfo
*
pBasicInfo
,
SAggSupporter
*
pAggSup
,
SExprInfo
*
pExprInfo
,
int32_t
numOfCols
,
SSDataBlock
*
pResultBlock
,
size_t
keyBufSize
,
const
char
*
pkey
);
void
initBasicInfo
(
SOptrBasicInfo
*
pInfo
,
SSDataBlock
*
pBlock
);
void
cleanupBasicInfo
(
SOptrBasicInfo
*
pInfo
);
void
initExprSupp
(
SExprSupp
*
pSup
,
SExprInfo
*
pExprInfo
,
int32_t
numOfExpr
);
void
cleanupExprSup
(
SExprSupp
*
pSup
);
int32_t
initAggInfo
(
SExprSupp
*
pSup
,
SAggSupporter
*
pAggSup
,
SExprInfo
*
pExprInfo
,
int32_t
numOfCols
,
size_t
keyBufSize
,
const
char
*
pkey
);
void
initResultSizeInfo
(
SOperatorInfo
*
pOperator
,
int32_t
numOfRows
);
void
doBuildResultDatablock
(
SOperatorInfo
*
pOperator
,
SOptrBasicInfo
*
pbInfo
,
SGroupResInfo
*
pGroupResInfo
,
SDiskbasedBuf
*
pBuf
);
void
doApplyFunctions
(
SExecTaskInfo
*
taskInfo
,
SqlFunctionCtx
*
pCtx
,
STimeWindow
*
pWin
,
SColumnInfoData
*
pTimeWindowData
,
int32_t
offset
,
int32_t
forwardStep
,
TSKEY
*
tsCol
,
int32_t
numOfTotal
,
int32_t
numOfOutput
,
int32_t
order
);
void
doDestroyBasicInfo
(
SOptrBasicInfo
*
pInfo
,
int32_t
numOfOutput
);
int32_t
extractDataBlockFromFetchRsp
(
SSDataBlock
*
pRes
,
SLoadRemoteDataInfo
*
pLoadInfo
,
int32_t
numOfRows
,
char
*
pData
,
int32_t
compLen
,
int32_t
numOfOutput
,
int64_t
startTs
,
uint64_t
*
total
,
SArray
*
pColList
);
...
...
@@ -764,9 +687,11 @@ void destroyBasicOperatorInfo(void* param, int32_t numOfOutput);
void
appendOneRowToDataBlock
(
SSDataBlock
*
pBlock
,
STupleHandle
*
pTupleHandle
);
void
setTbNameColData
(
void
*
pMeta
,
const
SSDataBlock
*
pBlock
,
SColumnInfoData
*
pColInfoData
,
int32_t
functionId
);
void
cleanupExecSupp
(
SExprSupp
*
pSupp
);
SSDataBlock
*
loadNextDataBlock
(
void
*
param
);
void
setResultRowInitCtx
(
SResultRow
*
pResult
,
SqlFunctionCtx
*
pCtx
,
int32_t
numOfOutput
,
int32_t
*
row
Cell
InfoOffset
);
void
setResultRowInitCtx
(
SResultRow
*
pResult
,
SqlFunctionCtx
*
pCtx
,
int32_t
numOfOutput
,
int32_t
*
row
Entry
InfoOffset
);
SResultRow
*
doSetResultOutBufByKey
(
SDiskbasedBuf
*
pResultBuf
,
SResultRowInfo
*
pResultRowInfo
,
char
*
pData
,
int16_t
bytes
,
bool
masterscan
,
uint64_t
groupId
,
...
...
@@ -885,12 +810,13 @@ STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowI
int32_t
getNumOfRowsInTimeWindow
(
SDataBlockInfo
*
pDataBlockInfo
,
TSKEY
*
pPrimaryColumn
,
int32_t
startPos
,
TSKEY
ekey
,
__block_search_fn_t
searchFn
,
STableQueryInfo
*
item
,
int32_t
order
);
int32_t
binarySearchForKey
(
char
*
pValue
,
int
num
,
TSKEY
key
,
int
order
);
int32_t
initStreamAggSupporter
(
SStreamAggSupporter
*
pSup
,
const
char
*
pKey
,
SqlFunctionCtx
*
pCtx
,
int32_t
numOfOutput
,
int32_t
size
);
int32_t
initStreamAggSupporter
(
SStreamAggSupporter
*
pSup
,
const
char
*
pKey
,
SqlFunctionCtx
*
pCtx
,
int32_t
numOfOutput
,
int32_t
size
);
SResultRow
*
getNewResultRow
(
SDiskbasedBuf
*
pResultBuf
,
int64_t
tableGroupId
,
int32_t
interBufSize
);
SResultWindowInfo
*
getSessionTimeWindow
(
SStreamAggSupporter
*
pAggSup
,
TSKEY
ts
,
uint64_t
groupId
,
int64_t
gap
,
int32_t
*
pIndex
);
int32_t
updateSessionWindowInfo
(
SResultWindowInfo
*
pWinInfo
,
TSKEY
*
pTs
,
int32_t
rows
,
int32_t
start
,
int64_t
gap
,
SHashObj
*
pStDeleted
);
int32_t
updateSessionWindowInfo
(
SResultWindowInfo
*
pWinInfo
,
TSKEY
*
pTs
,
int32_t
rows
,
int32_t
start
,
int64_t
gap
,
SHashObj
*
pStDeleted
);
bool
functionNeedToExecute
(
SqlFunctionCtx
*
pCtx
);
int32_t
compareTimeWindow
(
const
void
*
p1
,
const
void
*
p2
,
const
void
*
param
);
...
...
source/libs/executor/src/executil.c
浏览文件 @
f23abec3
...
...
@@ -42,21 +42,6 @@ void cleanupResultRowInfo(SResultRowInfo *pResultRowInfo) {
}
}
void
resetResultRowInfo
(
STaskRuntimeEnv
*
pRuntimeEnv
,
SResultRowInfo
*
pResultRowInfo
)
{
for
(
int32_t
i
=
0
;
i
<
pResultRowInfo
->
size
;
++
i
)
{
// SResultRow *pWindowRes = pResultRowInfo->pResult[i];
// clearResultRow(pRuntimeEnv, pWindowRes);
int32_t
groupIndex
=
0
;
int64_t
uid
=
0
;
SET_RES_WINDOW_KEY
(
pRuntimeEnv
->
keyBuf
,
&
groupIndex
,
sizeof
(
groupIndex
),
uid
);
taosHashRemove
(
pRuntimeEnv
->
pResultRowHashTable
,
(
const
char
*
)
pRuntimeEnv
->
keyBuf
,
GET_RES_WINDOW_KEY_LEN
(
sizeof
(
groupIndex
)));
}
pResultRowInfo
->
size
=
0
;
}
void
closeAllResultRows
(
SResultRowInfo
*
pResultRowInfo
)
{
// do nothing
}
...
...
@@ -518,14 +503,14 @@ static int32_t setSelectValueColumnInfo(SqlFunctionCtx* pCtx, int32_t numOfOutpu
return
TSDB_CODE_SUCCESS
;
}
SqlFunctionCtx
*
createSqlFunctionCtx
(
SExprInfo
*
pExprInfo
,
int32_t
numOfOutput
,
int32_t
**
row
Cell
InfoOffset
)
{
SqlFunctionCtx
*
createSqlFunctionCtx
(
SExprInfo
*
pExprInfo
,
int32_t
numOfOutput
,
int32_t
**
row
Entry
InfoOffset
)
{
SqlFunctionCtx
*
pFuncCtx
=
(
SqlFunctionCtx
*
)
taosMemoryCalloc
(
numOfOutput
,
sizeof
(
SqlFunctionCtx
));
if
(
pFuncCtx
==
NULL
)
{
return
NULL
;
}
*
row
Cell
InfoOffset
=
taosMemoryCalloc
(
numOfOutput
,
sizeof
(
int32_t
));
if
(
*
row
Cell
InfoOffset
==
0
)
{
*
row
Entry
InfoOffset
=
taosMemoryCalloc
(
numOfOutput
,
sizeof
(
int32_t
));
if
(
*
row
Entry
InfoOffset
==
0
)
{
taosMemoryFreeClear
(
pFuncCtx
);
return
NULL
;
}
...
...
@@ -584,8 +569,8 @@ SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput,
}
for
(
int32_t
i
=
1
;
i
<
numOfOutput
;
++
i
)
{
(
*
row
Cell
InfoOffset
)[
i
]
=
(
int32_t
)((
*
row
Cell
InfoOffset
)[
i
-
1
]
+
sizeof
(
SResultRowEntryInfo
)
+
pFuncCtx
[
i
-
1
].
resDataInfo
.
interBufSize
);
(
*
row
Entry
InfoOffset
)[
i
]
=
(
int32_t
)((
*
row
Entry
InfoOffset
)[
i
-
1
]
+
sizeof
(
SResultRowEntryInfo
)
+
pFuncCtx
[
i
-
1
].
resDataInfo
.
interBufSize
);
}
setSelectValueColumnInfo
(
pFuncCtx
,
numOfOutput
);
...
...
source/libs/executor/src/executorimpl.c
浏览文件 @
f23abec3
此差异已折叠。
点击以展开。
source/libs/executor/src/groupoperator.c
浏览文件 @
f23abec3
...
...
@@ -28,15 +28,16 @@
static
void
*
getCurrentDataGroupInfo
(
const
SPartitionOperatorInfo
*
pInfo
,
SDataGroupInfo
**
pGroupInfo
,
int32_t
len
);
static
int32_t
*
setupColumnOffset
(
const
SSDataBlock
*
pBlock
,
int32_t
rowCapacity
);
static
int32_t
setGroupResultOutputBuf
(
SOp
trBasicInfo
*
binfo
,
int32_t
numOfCols
,
char
*
pData
,
int16_t
type
,
int16_t
bytes
,
int32_t
groupId
,
SDiskbasedBuf
*
pBuf
,
SExecTaskInfo
*
pTaskInfo
,
SAggSupporter
*
pAggSup
);
static
int32_t
setGroupResultOutputBuf
(
SOp
eratorInfo
*
pOperator
,
SOptrBasicInfo
*
binfo
,
int32_t
numOfCols
,
char
*
pData
,
int16_t
bytes
,
int32_t
groupId
,
SDiskbasedBuf
*
pBuf
,
SAggSupporter
*
pAggSup
);
static
void
destroyGroupOperatorInfo
(
void
*
param
,
int32_t
numOfOutput
)
{
SGroupbyOperatorInfo
*
pInfo
=
(
SGroupbyOperatorInfo
*
)
param
;
doDestroyBasicInfo
(
&
pInfo
->
binfo
,
numOfOutput
);
cleanupBasicInfo
(
&
pInfo
->
binfo
);
taosMemoryFreeClear
(
pInfo
->
keyBuf
);
taosArrayDestroy
(
pInfo
->
pGroupCols
);
taosArrayDestroy
(
pInfo
->
pGroupColVals
);
cleanupExecSupp
(
&
pInfo
->
scalarSup
);
}
static
int32_t
initGroupOptrInfo
(
SArray
**
pGroupColVals
,
int32_t
*
keyLen
,
char
**
keyBuf
,
const
SArray
*
pGroupColList
)
{
...
...
@@ -216,7 +217,7 @@ static void doHashGroupbyAgg(SOperatorInfo* pOperator, SSDataBlock* pBlock) {
SExecTaskInfo
*
pTaskInfo
=
pOperator
->
pTaskInfo
;
SGroupbyOperatorInfo
*
pInfo
=
pOperator
->
info
;
SqlFunctionCtx
*
pCtx
=
p
Info
->
binfo
.
pCtx
;
SqlFunctionCtx
*
pCtx
=
p
Operator
->
exprSupp
.
pCtx
;
int32_t
numOfGroupCols
=
taosArrayGetSize
(
pInfo
->
pGroupCols
);
// if (type == TSDB_DATA_TYPE_FLOAT || type == TSDB_DATA_TYPE_DOUBLE) {
// qError("QInfo:0x%"PRIx64" group by not supported on double/float columns, abort", GET_TASKID(pRuntimeEnv));
...
...
@@ -250,16 +251,16 @@ static void doHashGroupbyAgg(SOperatorInfo* pOperator, SSDataBlock* pBlock) {
}
len
=
buildGroupKeys
(
pInfo
->
keyBuf
,
pInfo
->
pGroupColVals
);
int32_t
ret
=
setGroupResultOutputBuf
(
&
(
pInfo
->
binfo
),
pOperator
->
numOfExprs
,
pInfo
->
keyBuf
,
TSDB_DATA_TYPE_VARCHAR
,
len
,
0
,
pInfo
->
aggSup
.
pResultBuf
,
pTaskInfo
,
&
pInfo
->
aggSup
);
int32_t
ret
=
setGroupResultOutputBuf
(
pOperator
,
&
(
pInfo
->
binfo
),
pOperator
->
exprSupp
.
numOfExprs
,
pInfo
->
keyBuf
,
len
,
0
,
pInfo
->
aggSup
.
pResultBuf
,
&
pInfo
->
aggSup
);
if
(
ret
!=
TSDB_CODE_SUCCESS
)
{
// null data, too many state code
longjmp
(
pTaskInfo
->
env
,
TSDB_CODE_QRY_APP_ERROR
);
}
int32_t
rowIndex
=
j
-
num
;
doApplyFunctions
(
pTaskInfo
,
pCtx
,
&
w
,
NULL
,
rowIndex
,
num
,
NULL
,
pBlock
->
info
.
rows
,
pOperator
->
numOfExprs
,
TSDB_ORDER_ASC
);
doApplyFunctions
(
pTaskInfo
,
pCtx
,
&
w
,
NULL
,
rowIndex
,
num
,
NULL
,
pBlock
->
info
.
rows
,
pOperator
->
exprSupp
.
numOfExprs
,
TSDB_ORDER_ASC
);
// assign the group keys or user input constant values if required
doAssignGroupKeys
(
pCtx
,
pOperator
->
numOfExprs
,
pBlock
->
info
.
rows
,
rowIndex
);
doAssignGroupKeys
(
pCtx
,
pOperator
->
exprSupp
.
numOfExprs
,
pBlock
->
info
.
rows
,
rowIndex
);
recordNewGroupKeys
(
pInfo
->
pGroupCols
,
pInfo
->
pGroupColVals
,
pBlock
,
j
);
num
=
1
;
}
...
...
@@ -267,15 +268,15 @@ static void doHashGroupbyAgg(SOperatorInfo* pOperator, SSDataBlock* pBlock) {
if
(
num
>
0
)
{
len
=
buildGroupKeys
(
pInfo
->
keyBuf
,
pInfo
->
pGroupColVals
);
int32_t
ret
=
setGroupResultOutputBuf
(
&
(
pInfo
->
binfo
),
pOperator
->
numOfExprs
,
pInfo
->
keyBuf
,
TSDB_DATA_TYPE_VARCHAR
,
len
,
0
,
pInfo
->
aggSup
.
pResultBuf
,
pTaskInfo
,
&
pInfo
->
aggSup
);
setGroupResultOutputBuf
(
pOperator
,
&
(
pInfo
->
binfo
),
pOperator
->
exprSupp
.
numOfExprs
,
pInfo
->
keyBuf
,
len
,
0
,
pInfo
->
aggSup
.
pResultBuf
,
&
pInfo
->
aggSup
);
if
(
ret
!=
TSDB_CODE_SUCCESS
)
{
longjmp
(
pTaskInfo
->
env
,
TSDB_CODE_QRY_APP_ERROR
);
}
int32_t
rowIndex
=
pBlock
->
info
.
rows
-
num
;
doApplyFunctions
(
pTaskInfo
,
pCtx
,
&
w
,
NULL
,
rowIndex
,
num
,
NULL
,
pBlock
->
info
.
rows
,
pOperator
->
numOfExprs
,
TSDB_ORDER_ASC
);
doAssignGroupKeys
(
pCtx
,
pOperator
->
numOfExprs
,
pBlock
->
info
.
rows
,
rowIndex
);
doApplyFunctions
(
pTaskInfo
,
pCtx
,
&
w
,
NULL
,
rowIndex
,
num
,
NULL
,
pBlock
->
info
.
rows
,
pOperator
->
exprSupp
.
numOfExprs
,
TSDB_ORDER_ASC
);
doAssignGroupKeys
(
pCtx
,
pOperator
->
exprSupp
.
numOfExprs
,
pBlock
->
info
.
rows
,
rowIndex
);
}
}
...
...
@@ -319,11 +320,11 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator) {
}
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock
(
pOperator
,
p
Info
->
binfo
.
pCtx
,
pBlock
,
order
,
scanFlag
,
true
);
setInputDataBlock
(
pOperator
,
p
Operator
->
exprSupp
.
pCtx
,
pBlock
,
order
,
scanFlag
,
true
);
// there is an scalar expression that needs to be calculated right before apply the group aggregation.
if
(
pInfo
->
scalarSup
.
p
Scalar
ExprInfo
!=
NULL
)
{
pTaskInfo
->
code
=
projectApplyFunctions
(
pInfo
->
scalarSup
.
p
ScalarExprInfo
,
pBlock
,
pBlock
,
pInfo
->
scalarSup
.
pScalarFuncCtx
,
pInfo
->
scalarSup
.
numOfScalarExpr
,
NULL
);
if
(
pInfo
->
scalarSup
.
pExprInfo
!=
NULL
)
{
pTaskInfo
->
code
=
projectApplyFunctions
(
pInfo
->
scalarSup
.
p
ExprInfo
,
pBlock
,
pBlock
,
pInfo
->
scalarSup
.
pCtx
,
pInfo
->
scalarSup
.
numOfExprs
,
NULL
);
if
(
pTaskInfo
->
code
!=
TSDB_CODE_SUCCESS
)
{
longjmp
(
pTaskInfo
->
env
,
pTaskInfo
->
code
);
}
...
...
@@ -386,9 +387,9 @@ SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pEx
pInfo
->
pGroupCols
=
pGroupColList
;
pInfo
->
pCondition
=
pCondition
;
pInfo
->
scalarSup
.
p
Scalar
ExprInfo
=
pScalarExprInfo
;
pInfo
->
scalarSup
.
numOf
ScalarExpr
=
numOfScalarExpr
;
pInfo
->
scalarSup
.
p
ScalarFuncCtx
=
createSqlFunctionCtx
(
pScalarExprInfo
,
numOfScalarExpr
,
&
pInfo
->
scalarSup
.
rowCell
InfoOffset
);
pInfo
->
scalarSup
.
pExprInfo
=
pScalarExprInfo
;
pInfo
->
scalarSup
.
numOf
Exprs
=
numOfScalarExpr
;
pInfo
->
scalarSup
.
p
Ctx
=
createSqlFunctionCtx
(
pScalarExprInfo
,
numOfScalarExpr
,
&
pInfo
->
scalarSup
.
rowEntry
InfoOffset
);
int32_t
code
=
initGroupOptrInfo
(
&
pInfo
->
pGroupColVals
,
&
pInfo
->
groupKeyLen
,
&
pInfo
->
keyBuf
,
pGroupColList
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
...
...
@@ -396,15 +397,16 @@ SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pEx
}
initResultSizeInfo
(
pOperator
,
4096
);
initAggInfo
(
&
pInfo
->
binfo
,
&
pInfo
->
aggSup
,
pExprInfo
,
numOfCols
,
pResultBlock
,
pInfo
->
groupKeyLen
,
pTaskInfo
->
id
.
str
);
initAggInfo
(
&
pOperator
->
exprSupp
,
&
pInfo
->
aggSup
,
pExprInfo
,
numOfCols
,
pInfo
->
groupKeyLen
,
pTaskInfo
->
id
.
str
);
initBasicInfo
(
&
pInfo
->
binfo
,
pResultBlock
);
initResultRowInfo
(
&
pInfo
->
binfo
.
resultRowInfo
);
pOperator
->
name
=
"GroupbyAggOperator"
;
pOperator
->
blocking
=
true
;
pOperator
->
status
=
OP_NOT_OPENED
;
// pOperator->operatorType = OP_Groupby;
pOperator
->
pExpr
=
pExprInfo
;
pOperator
->
numOfExprs
=
numOfCols
;
pOperator
->
exprSupp
.
pExprInfo
=
pExprInfo
;
pOperator
->
exprSupp
.
numOfExprs
=
numOfCols
;
pOperator
->
info
=
pInfo
;
pOperator
->
pTaskInfo
=
pTaskInfo
;
...
...
@@ -441,9 +443,9 @@ static void doHashPartition(SOperatorInfo* pOperator, SSDataBlock* pBlock) {
// group id
size_t
numOfCols
=
pOperator
->
numOfExprs
;
size_t
numOfCols
=
pOperator
->
exprSupp
.
numOfExprs
;
for
(
int32_t
i
=
0
;
i
<
numOfCols
;
++
i
)
{
SExprInfo
*
pExpr
=
&
pOperator
->
pExpr
[
i
];
SExprInfo
*
pExpr
=
&
pOperator
->
exprSupp
.
pExprInfo
[
i
];
int32_t
slotId
=
pExpr
->
base
.
pParam
[
0
].
pCol
->
slotId
;
SColumnInfoData
*
pColInfoData
=
taosArrayGet
(
pBlock
->
pDataBlock
,
slotId
);
...
...
@@ -645,8 +647,8 @@ static SSDataBlock* hashPartition(SOperatorInfo* pOperator) {
}
// there is an scalar expression that needs to be calculated right before apply the group aggregation.
if
(
pInfo
->
scalarSup
p
.
pScalar
ExprInfo
!=
NULL
)
{
pTaskInfo
->
code
=
projectApplyFunctions
(
pInfo
->
scalarSup
p
.
pScalarExprInfo
,
pBlock
,
pBlock
,
pInfo
->
scalarSupp
.
pScalarFuncCtx
,
pInfo
->
scalarSupp
.
numOfScalarExpr
,
NULL
);
if
(
pInfo
->
scalarSup
.
p
ExprInfo
!=
NULL
)
{
pTaskInfo
->
code
=
projectApplyFunctions
(
pInfo
->
scalarSup
.
pExprInfo
,
pBlock
,
pBlock
,
pInfo
->
scalarSup
.
pCtx
,
pInfo
->
scalarSup
.
numOfExprs
,
NULL
);
if
(
pTaskInfo
->
code
!=
TSDB_CODE_SUCCESS
)
{
longjmp
(
pTaskInfo
->
env
,
pTaskInfo
->
code
);
}
...
...
@@ -664,16 +666,20 @@ static SSDataBlock* hashPartition(SOperatorInfo* pOperator) {
static
void
destroyPartitionOperatorInfo
(
void
*
param
,
int32_t
numOfOutput
)
{
SPartitionOperatorInfo
*
pInfo
=
(
SPartitionOperatorInfo
*
)
param
;
doDestroyBasicInfo
(
&
pInfo
->
binfo
,
numOfOutput
);
cleanupBasicInfo
(
&
pInfo
->
binfo
);
taosArrayDestroy
(
pInfo
->
pGroupCols
);
for
(
int
i
=
0
;
i
<
taosArrayGetSize
(
pInfo
->
pGroupColVals
);
i
++
){
SGroupKeys
key
=
*
(
SGroupKeys
*
)
taosArrayGet
(
pInfo
->
pGroupColVals
,
i
);
taosMemoryFree
(
key
.
pData
);
}
taosArrayDestroy
(
pInfo
->
pGroupColVals
);
taosMemoryFree
(
pInfo
->
keyBuf
);
taosHashCleanup
(
pInfo
->
pGroupSet
);
taosMemoryFree
(
pInfo
->
columnOffset
);
cleanupExecSupp
(
&
pInfo
->
scalarSup
);
}
SOperatorInfo
*
createPartitionOperatorInfo
(
SOperatorInfo
*
downstream
,
SPartitionPhysiNode
*
pPartNode
,
SExecTaskInfo
*
pTaskInfo
)
{
...
...
@@ -691,10 +697,10 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition
pInfo
->
pGroupCols
=
extractPartitionColInfo
(
pPartNode
->
pPartitionKeys
);
if
(
pPartNode
->
pExprs
!=
NULL
)
{
pInfo
->
scalarSup
p
.
numOfScalarExpr
=
0
;
pInfo
->
scalarSup
p
.
pScalarExprInfo
=
createExprInfo
(
pPartNode
->
pExprs
,
NULL
,
&
pInfo
->
scalarSupp
.
numOfScalarExpr
);
pInfo
->
scalarSup
p
.
pScalarFunc
Ctx
=
createSqlFunctionCtx
(
pInfo
->
scalarSup
p
.
pScalarExprInfo
,
pInfo
->
scalarSupp
.
numOfScalarExpr
,
&
pInfo
->
scalarSupp
.
rowCell
InfoOffset
);
pInfo
->
scalarSup
.
numOfExprs
=
0
;
pInfo
->
scalarSup
.
pExprInfo
=
createExprInfo
(
pPartNode
->
pExprs
,
NULL
,
&
pInfo
->
scalarSup
.
numOfExprs
);
pInfo
->
scalarSup
.
p
Ctx
=
createSqlFunctionCtx
(
pInfo
->
scalarSup
.
pExprInfo
,
pInfo
->
scalarSup
.
numOfExprs
,
&
pInfo
->
scalarSup
.
rowEntry
InfoOffset
);
}
_hash_fn_t
hashFn
=
taosGetDefaultHashFunction
(
TSDB_DATA_TYPE_BINARY
);
...
...
@@ -724,8 +730,8 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition
pOperator
->
status
=
OP_NOT_OPENED
;
pOperator
->
operatorType
=
QUERY_NODE_PHYSICAL_PLAN_PARTITION
;
pInfo
->
binfo
.
pRes
=
pResBlock
;
pOperator
->
numOfExprs
=
numOfCols
;
pOperator
->
pExpr
=
pExprInfo
;
pOperator
->
exprSupp
.
numOfExprs
=
numOfCols
;
pOperator
->
exprSupp
.
pExprInfo
=
pExprInfo
;
pOperator
->
info
=
pInfo
;
pOperator
->
pTaskInfo
=
pTaskInfo
;
...
...
@@ -742,16 +748,16 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition
return
NULL
;
}
int32_t
setGroupResultOutputBuf
(
SOp
trBasicInfo
*
binfo
,
int32_t
numOfCols
,
char
*
pData
,
int16_t
type
,
int16_t
bytes
,
int32_t
groupId
,
SDiskbasedBuf
*
pBuf
,
S
ExecTaskInfo
*
pTaskInfo
,
SAggSupporter
*
pAggSup
)
{
int32_t
setGroupResultOutputBuf
(
SOp
eratorInfo
*
pOperator
,
SOptrBasicInfo
*
binfo
,
int32_t
numOfCols
,
char
*
pData
,
int16_t
bytes
,
int32_t
groupId
,
SDiskbasedBuf
*
pBuf
,
S
AggSupporter
*
pAggSup
)
{
SExecTaskInfo
*
pTaskInfo
=
pOperator
->
pTaskInfo
;
SResultRowInfo
*
pResultRowInfo
=
&
binfo
->
resultRowInfo
;
SqlFunctionCtx
*
pCtx
=
binfo
->
pCtx
;
SqlFunctionCtx
*
pCtx
=
pOperator
->
exprSupp
.
pCtx
;
SResultRow
*
pResultRow
=
doSetResultOutBufByKey
(
pBuf
,
pResultRowInfo
,
(
char
*
)
pData
,
bytes
,
true
,
groupId
,
pTaskInfo
,
false
,
pAggSup
);
assert
(
pResultRow
!=
NULL
);
setResultRowInitCtx
(
pResultRow
,
pCtx
,
numOfCols
,
binfo
->
rowCell
InfoOffset
);
setResultRowInitCtx
(
pResultRow
,
pCtx
,
numOfCols
,
pOperator
->
exprSupp
.
rowEntry
InfoOffset
);
return
TSDB_CODE_SUCCESS
;
}
\ No newline at end of file
source/libs/executor/src/joinoperator.c
浏览文件 @
f23abec3
...
...
@@ -48,8 +48,8 @@ SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t
pOperator
->
operatorType
=
QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN
;
pOperator
->
blocking
=
false
;
pOperator
->
status
=
OP_NOT_OPENED
;
pOperator
->
pExpr
=
pExprInfo
;
pOperator
->
numOfExprs
=
numOfCols
;
pOperator
->
exprSupp
.
pExprInfo
=
pExprInfo
;
pOperator
->
exprSupp
.
numOfExprs
=
numOfCols
;
pOperator
->
info
=
pInfo
;
pOperator
->
pTaskInfo
=
pTaskInfo
;
...
...
@@ -132,10 +132,10 @@ SSDataBlock* doMergeJoin(struct SOperatorInfo* pOperator) {
// only the timestamp match support for ordinary table
ASSERT
(
pLeftCol
->
info
.
type
==
TSDB_DATA_TYPE_TIMESTAMP
);
if
(
*
(
int64_t
*
)
pLeftVal
==
*
(
int64_t
*
)
pRightVal
)
{
for
(
int32_t
i
=
0
;
i
<
pOperator
->
numOfExprs
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pOperator
->
exprSupp
.
numOfExprs
;
++
i
)
{
SColumnInfoData
*
pDst
=
taosArrayGet
(
pRes
->
pDataBlock
,
i
);
SExprInfo
*
pExprInfo
=
&
pOperator
->
pExpr
[
i
];
SExprInfo
*
pExprInfo
=
&
pOperator
->
exprSupp
.
pExprInfo
[
i
];
int32_t
blockId
=
pExprInfo
->
base
.
pParam
[
0
].
pCol
->
dataBlockId
;
int32_t
slotId
=
pExprInfo
->
base
.
pParam
[
0
].
pCol
->
slotId
;
...
...
source/libs/executor/src/scanoperator.c
浏览文件 @
f23abec3
...
...
@@ -261,9 +261,9 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableSca
relocateColumnData
(
pBlock
,
pTableScanInfo
->
pColMatchInfo
,
pCols
);
// currently only the tbname pseudo column
if
(
pTableScanInfo
->
numOfPseudoExpr
>
0
)
{
addTagPseudoColumnData
(
&
pTableScanInfo
->
readHandle
,
pTableScanInfo
->
pPseudoExpr
,
pTableScanInfo
->
numOfPseudoExpr
,
pBlock
);
if
(
pTableScanInfo
->
pseudoSup
.
numOfExprs
>
0
)
{
SExprSupp
*
pSup
=
&
pTableScanInfo
->
pseudoSup
;
addTagPseudoColumnData
(
&
pTableScanInfo
->
readHandle
,
pSup
->
pExprInfo
,
pSup
->
numOfExprs
,
pBlock
);
}
int64_t
st
=
taosGetTimestampMs
();
...
...
@@ -538,8 +538,9 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
}
if
(
pTableScanNode
->
scan
.
pScanPseudoCols
!=
NULL
)
{
pInfo
->
pPseudoExpr
=
createExprInfo
(
pTableScanNode
->
scan
.
pScanPseudoCols
,
NULL
,
&
pInfo
->
numOfPseudoExpr
);
pInfo
->
pPseudoCtx
=
createSqlFunctionCtx
(
pInfo
->
pPseudoExpr
,
pInfo
->
numOfPseudoExpr
,
&
pInfo
->
rowCellInfoOffset
);
SExprSupp
*
pSup
=
&
pInfo
->
pseudoSup
;
pSup
->
pExprInfo
=
createExprInfo
(
pTableScanNode
->
scan
.
pScanPseudoCols
,
NULL
,
&
pSup
->
numOfExprs
);
pSup
->
pCtx
=
createSqlFunctionCtx
(
pSup
->
pExprInfo
,
pSup
->
numOfExprs
,
&
pSup
->
rowEntryInfoOffset
);
}
pInfo
->
scanInfo
=
(
SScanInfo
){.
numOfAsc
=
pTableScanNode
->
scanSeq
[
0
],
.
numOfDesc
=
pTableScanNode
->
scanSeq
[
1
]};
...
...
@@ -563,7 +564,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
pOperator
->
blocking
=
false
;
pOperator
->
status
=
OP_NOT_OPENED
;
pOperator
->
info
=
pInfo
;
pOperator
->
numOfExprs
=
numOfCols
;
pOperator
->
exprSupp
.
numOfExprs
=
numOfCols
;
pOperator
->
pTaskInfo
=
pTaskInfo
;
pOperator
->
fpSet
=
createOperatorFpSet
(
operatorDummyOpenFn
,
doTableScan
,
NULL
,
NULL
,
destroyTableScanOperatorInfo
,
...
...
@@ -1117,7 +1118,7 @@ SOperatorInfo* createStreamScanOperatorInfo(void* pDataReader, SReadHandle* pHan
pOperator
->
blocking
=
false
;
pOperator
->
status
=
OP_NOT_OPENED
;
pOperator
->
info
=
pInfo
;
pOperator
->
numOfExprs
=
pInfo
->
pRes
->
info
.
numOfCols
;
pOperator
->
exprSupp
.
numOfExprs
=
pInfo
->
pRes
->
info
.
numOfCols
;
pOperator
->
pTaskInfo
=
pTaskInfo
;
pOperator
->
fpSet
=
...
...
@@ -1512,7 +1513,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
}
extractDataBlockFromFetchRsp
(
pInfo
->
pRes
,
&
pInfo
->
loadInfo
,
pRsp
->
numOfRows
,
pRsp
->
data
,
pRsp
->
compLen
,
pOperator
->
numOfExprs
,
startTs
,
NULL
,
pInfo
->
scanCols
);
pOperator
->
exprSupp
.
numOfExprs
,
startTs
,
NULL
,
pInfo
->
scanCols
);
// todo log the filter info
doFilterResult
(
pInfo
);
...
...
@@ -1627,7 +1628,7 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScan
pOperator
->
blocking
=
false
;
pOperator
->
status
=
OP_NOT_OPENED
;
pOperator
->
info
=
pInfo
;
pOperator
->
numOfExprs
=
pResBlock
->
info
.
numOfCols
;
pOperator
->
exprSupp
.
numOfExprs
=
pResBlock
->
info
.
numOfCols
;
pOperator
->
pTaskInfo
=
pTaskInfo
;
pOperator
->
fpSet
=
...
...
@@ -1658,11 +1659,11 @@ static SSDataBlock* doTagScan(SOperatorInfo* pOperator) {
int32_t count = 0;
SArray* pa = GET_TABLEGROUP(pRuntimeEnv, 0);
int32_t functionId = getExprFunctionId(&pOperator->
pExpr
[0]);
int32_t functionId = getExprFunctionId(&pOperator->
exprSupp.pExprInfo
[0]);
if (functionId == FUNCTION_TID_TAG) { // return the tags & table Id
assert(pQueryAttr->numOfOutput == 1);
SExprInfo* pExprInfo = &pOperator->
pExpr
[0];
SExprInfo* pExprInfo = &pOperator->
exprSupp.pExprInfo
[0];
int32_t rsize = pExprInfo->base.resSchema.bytes;
count = 0;
...
...
@@ -1725,7 +1726,7 @@ static SSDataBlock* doTagScan(SOperatorInfo* pOperator) {
#endif
STagScanInfo
*
pInfo
=
pOperator
->
info
;
SExprInfo
*
pExprInfo
=
&
pOperator
->
pExpr
[
0
];
SExprInfo
*
pExprInfo
=
&
pOperator
->
exprSupp
.
pExprInfo
[
0
];
SSDataBlock
*
pRes
=
pInfo
->
pRes
;
int32_t
size
=
taosArrayGetSize
(
pInfo
->
pTableList
->
pTableList
);
...
...
@@ -1743,7 +1744,7 @@ static SSDataBlock* doTagScan(SOperatorInfo* pOperator) {
STableKeyInfo
*
item
=
taosArrayGet
(
pInfo
->
pTableList
->
pTableList
,
pInfo
->
curPos
);
metaGetTableEntryByUid
(
&
mr
,
item
->
uid
);
for
(
int32_t
j
=
0
;
j
<
pOperator
->
numOfExprs
;
++
j
)
{
for
(
int32_t
j
=
0
;
j
<
pOperator
->
exprSupp
.
numOfExprs
;
++
j
)
{
SColumnInfoData
*
pDst
=
taosArrayGet
(
pRes
->
pDataBlock
,
pExprInfo
[
j
].
base
.
resSchema
.
slotId
);
// refactor later
...
...
@@ -1823,8 +1824,8 @@ SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysi
pOperator
->
blocking
=
false
;
pOperator
->
status
=
OP_NOT_OPENED
;
pOperator
->
info
=
pInfo
;
pOperator
->
pExpr
=
pExprInfo
;
pOperator
->
numOfExprs
=
numOfExprs
;
pOperator
->
exprSupp
.
pExprInfo
=
pExprInfo
;
pOperator
->
exprSupp
.
numOfExprs
=
numOfExprs
;
pOperator
->
pTaskInfo
=
pTaskInfo
;
initResultSizeInfo
(
pOperator
,
4096
);
...
...
@@ -1869,7 +1870,7 @@ typedef struct STableMergeScanInfo {
SNode
*
pFilterNode
;
// filter info, which is push down by optimizer
SqlFunctionCtx
*
pCtx
;
// which belongs to the direct upstream operator operator query context
SResultRowInfo
*
pResultRowInfo
;
int32_t
*
row
Cell
InfoOffset
;
int32_t
*
row
Entry
InfoOffset
;
SExprInfo
*
pExpr
;
SSDataBlock
*
pResBlock
;
SArray
*
pColMatchInfo
;
...
...
@@ -1878,7 +1879,7 @@ typedef struct STableMergeScanInfo {
SExprInfo
*
pPseudoExpr
;
int32_t
numOfPseudoExpr
;
SqlFunctionCtx
*
pPseudoCtx
;
// int32_t* row
Cell
InfoOffset;
// int32_t* row
Entry
InfoOffset;
SQueryTableDataCond
cond
;
int32_t
scanFlag
;
// table scan flag to denote if it is a repeat/reverse/main scan
...
...
@@ -2257,7 +2258,7 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN
if
(
pTableScanNode
->
scan
.
pScanPseudoCols
!=
NULL
)
{
pInfo
->
pPseudoExpr
=
createExprInfo
(
pTableScanNode
->
scan
.
pScanPseudoCols
,
NULL
,
&
pInfo
->
numOfPseudoExpr
);
pInfo
->
pPseudoCtx
=
createSqlFunctionCtx
(
pInfo
->
pPseudoExpr
,
pInfo
->
numOfPseudoExpr
,
&
pInfo
->
row
Cell
InfoOffset
);
pInfo
->
pPseudoCtx
=
createSqlFunctionCtx
(
pInfo
->
pPseudoExpr
,
pInfo
->
numOfPseudoExpr
,
&
pInfo
->
row
Entry
InfoOffset
);
}
pInfo
->
scanInfo
=
(
SScanInfo
){.
numOfAsc
=
pTableScanNode
->
scanSeq
[
0
],
.
numOfDesc
=
pTableScanNode
->
scanSeq
[
1
]};
...
...
@@ -2303,7 +2304,7 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN
pOperator
->
blocking
=
false
;
pOperator
->
status
=
OP_NOT_OPENED
;
pOperator
->
info
=
pInfo
;
pOperator
->
numOfExprs
=
numOfCols
;
pOperator
->
exprSupp
.
numOfExprs
=
numOfCols
;
pOperator
->
pTaskInfo
=
pTaskInfo
;
initResultSizeInfo
(
pOperator
,
1024
);
...
...
source/libs/executor/src/sortoperator.c
浏览文件 @
f23abec3
...
...
@@ -39,7 +39,7 @@ SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode*
SArray
*
pColMatchColInfo
=
extractColMatchInfo
(
pSortPhyNode
->
pTargets
,
pDescNode
,
&
numOfOutputCols
,
COL_MATCH_FROM_SLOT_ID
);
p
Info
->
binfo
.
pCtx
=
createSqlFunctionCtx
(
pExprInfo
,
numOfCols
,
&
pInfo
->
binfo
.
rowCell
InfoOffset
);
p
Operator
->
exprSupp
.
pCtx
=
createSqlFunctionCtx
(
pExprInfo
,
numOfCols
,
&
pOperator
->
exprSupp
.
rowEntry
InfoOffset
);
pInfo
->
binfo
.
pRes
=
pResBlock
;
initResultSizeInfo
(
pOperator
,
1024
);
...
...
@@ -51,8 +51,8 @@ SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode*
pOperator
->
blocking
=
true
;
pOperator
->
status
=
OP_NOT_OPENED
;
pOperator
->
info
=
pInfo
;
pOperator
->
pExpr
=
pExprInfo
;
pOperator
->
numOfExprs
=
numOfCols
;
pOperator
->
exprSupp
.
pExprInfo
=
pExprInfo
;
pOperator
->
exprSupp
.
numOfExprs
=
numOfCols
;
pOperator
->
pTaskInfo
=
pTaskInfo
;
// lazy evaluation for the following parameter since the input datablock is not known till now.
...
...
@@ -145,9 +145,9 @@ SSDataBlock* loadNextDataBlock(void* param) {
void
applyScalarFunction
(
SSDataBlock
*
pBlock
,
void
*
param
)
{
SOperatorInfo
*
pOperator
=
param
;
SSortOperatorInfo
*
pSort
=
pOperator
->
info
;
if
(
pOperator
->
pExpr
!=
NULL
)
{
if
(
pOperator
->
exprSupp
.
pExprInfo
!=
NULL
)
{
int32_t
code
=
projectApplyFunctions
(
pOperator
->
pExpr
,
pBlock
,
pBlock
,
pSort
->
binfo
.
pCtx
,
pOperator
->
numOfExprs
,
NULL
);
projectApplyFunctions
(
pOperator
->
exprSupp
.
pExprInfo
,
pBlock
,
pBlock
,
pOperator
->
exprSupp
.
pCtx
,
pOperator
->
exprSupp
.
numOfExprs
,
NULL
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
longjmp
(
pOperator
->
pTaskInfo
->
env
,
code
);
}
...
...
source/libs/executor/src/timewindowoperator.c
浏览文件 @
f23abec3
此差异已折叠。
点击以展开。
tests/script/tsim/testsuit.sim
浏览文件 @
f23abec3
#
run tsim/user/pass_alter.sim
#
run tsim/user/basic1.sim
#
run tsim/user/privilege2.sim
#
run tsim/user/user_len.sim
#
run tsim/user/privilege1.sim
#
run tsim/user/pass_len.sim
#
run tsim/table/basic1.sim
#
run tsim/trans/lossdata1.sim
#
run tsim/trans/create_db.sim
#
run tsim/stable/alter_metrics.sim
#
run tsim/stable/tag_modify.sim
#
run tsim/stable/alter_comment.sim
#
run tsim/stable/column_drop.sim
#
run tsim/stable/column_modify.sim
#
run tsim/stable/tag_rename.sim
#
run tsim/stable/vnode3.sim
#
run tsim/stable/metrics.sim
#
run tsim/stable/alter_insert2.sim
#
run tsim/stable/show.sim
#
run tsim/stable/alter_import.sim
#
run tsim/stable/tag_add.sim
#
run tsim/stable/tag_drop.sim
#
run tsim/stable/column_add.sim
#
run tsim/stable/alter_count.sim
#
run tsim/stable/values.sim
run tsim/user/pass_alter.sim
run tsim/user/basic1.sim
run tsim/user/privilege2.sim
run tsim/user/user_len.sim
run tsim/user/privilege1.sim
run tsim/user/pass_len.sim
run tsim/table/basic1.sim
run tsim/trans/lossdata1.sim
run tsim/trans/create_db.sim
run tsim/stable/alter_metrics.sim
run tsim/stable/tag_modify.sim
run tsim/stable/alter_comment.sim
run tsim/stable/column_drop.sim
run tsim/stable/column_modify.sim
run tsim/stable/tag_rename.sim
run tsim/stable/vnode3.sim
run tsim/stable/metrics.sim
run tsim/stable/alter_insert2.sim
run tsim/stable/show.sim
run tsim/stable/alter_import.sim
run tsim/stable/tag_add.sim
run tsim/stable/tag_drop.sim
run tsim/stable/column_add.sim
run tsim/stable/alter_count.sim
run tsim/stable/values.sim
run tsim/stable/dnode3.sim
#
run tsim/stable/alter_insert1.sim
#
run tsim/stable/refcount.sim
#
run tsim/stable/disk.sim
run tsim/stable/alter_insert1.sim
run tsim/stable/refcount.sim
run tsim/stable/disk.sim
run tsim/db/basic1.sim
run tsim/db/basic3.sim
#
run tsim/db/basic7.sim
run tsim/db/basic7.sim
run tsim/db/basic6.sim
run tsim/db/create_all_options.sim
run tsim/db/basic2.sim
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录