Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
1d7581f6
T
TDengine
项目概览
taosdata
/
TDengine
12 个月 前同步成功
通知
1180
Star
22014
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看板
体验新版 GitCode,发现更多精彩内容 >>
提交
1d7581f6
编写于
4月 03, 2023
作者:
wmmhello
浏览文件
操作
浏览文件
下载
差异文件
fix:conflicts from main
上级
d651ba02
d0a7e02a
变更
12
展开全部
隐藏空白更改
内联
并排
Showing
12 changed file
with
789 addition
and
1004 deletion
+789
-1004
source/common/src/tdatablock.c
source/common/src/tdatablock.c
+6
-40
source/libs/executor/inc/executorimpl.h
source/libs/executor/inc/executorimpl.h
+33
-18
source/libs/executor/src/aggregateoperator.c
source/libs/executor/src/aggregateoperator.c
+563
-0
source/libs/executor/src/cachescanoperator.c
source/libs/executor/src/cachescanoperator.c
+10
-8
source/libs/executor/src/exchangeoperator.c
source/libs/executor/src/exchangeoperator.c
+3
-3
source/libs/executor/src/executor.c
source/libs/executor/src/executor.c
+16
-12
source/libs/executor/src/executorimpl.c
source/libs/executor/src/executorimpl.c
+58
-853
source/libs/executor/src/scanoperator.c
source/libs/executor/src/scanoperator.c
+68
-52
source/libs/executor/src/sysscanoperator.c
source/libs/executor/src/sysscanoperator.c
+10
-8
source/libs/executor/src/timewindowoperator.c
source/libs/executor/src/timewindowoperator.c
+1
-7
source/libs/parser/src/parTranslater.c
source/libs/parser/src/parTranslater.c
+19
-1
tests/script/tsim/query/interval-offset.sim
tests/script/tsim/query/interval-offset.sim
+2
-2
未找到文件。
source/common/src/tdatablock.c
浏览文件 @
1d7581f6
...
...
@@ -1035,6 +1035,7 @@ int32_t blockDataSort(SSDataBlock* pDataBlock, SArray* pOrderInfo) {
return
TSDB_CODE_SUCCESS
;
}
#if 0
typedef struct SHelper {
int32_t index;
union {
...
...
@@ -1083,59 +1084,20 @@ SHelper* createTupleIndex_rv(int32_t numOfRows, SArray* pOrderInfo, SSDataBlock*
int32_t dataBlockCompar_rv(const void* p1, const void* p2, const void* param) {
const SSDataBlockSortHelper* pHelper = (const SSDataBlockSortHelper*)param;
// SSDataBlock* pDataBlock = pHelper->pDataBlock;
SHelper* left = (SHelper*)p1;
SHelper* right = (SHelper*)p2;
SArray* pInfo = pHelper->orderInfo;
int32_t offset = 0;
// for(int32_t i = 0; i < pInfo->size; ++i) {
// SBlockOrderInfo* pOrder = TARRAY_GET_ELEM(pInfo, 0);
// SColumnInfoData* pColInfoData = pOrder->pColData;//TARRAY_GET_ELEM(pDataBlock->pDataBlock, pOrder->colIndex);
// if (pColInfoData->hasNull) {
// bool leftNull = colDataIsNull(pColInfoData, pDataBlock->info.rows, left, pDataBlock->pBlockAgg);
// bool rightNull = colDataIsNull(pColInfoData, pDataBlock->info.rows, right, pDataBlock->pBlockAgg);
// if (leftNull && rightNull) {
// continue; // continue to next slot
// }
//
// if (rightNull) {
// return pHelper->nullFirst? 1:-1;
// }
//
// if (leftNull) {
// return pHelper->nullFirst? -1:1;
// }
// }
// void* left1 = colDataGetData(pColInfoData, left);
// void* right1 = colDataGetData(pColInfoData, right);
// switch(pColInfoData->info.type) {
// case TSDB_DATA_TYPE_INT: {
int32_t leftx = *(int32_t*)left->pData; //*(int32_t*)(left->pData + offset);
int32_t rightx = *(int32_t*)right->pData; //*(int32_t*)(right->pData + offset);
// offset += pColInfoData->info.bytes;
if (leftx == rightx) {
// break;
return 0;
} else {
// if (pOrder->order == TSDB_ORDER_ASC) {
return (leftx < rightx) ? -1 : 1;
// } else {
// return (leftx < rightx)? 1:-1;
// }
}
// }
// default:
// assert(0);
// }
// }
}
return 0;
}
...
...
@@ -1179,6 +1141,7 @@ int32_t blockDataSort_rv(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullF
// destroyTupleIndex(index);
return 0;
}
#endif
void
blockDataCleanup
(
SSDataBlock
*
pDataBlock
)
{
blockDataEmpty
(
pDataBlock
);
...
...
@@ -1887,6 +1850,7 @@ static char* formatTimestamp(char* buf, int64_t val, int precision) {
return
buf
;
}
#if 0
void blockDebugShowDataBlock(SSDataBlock* pBlock, const char* flag) {
SArray* dataBlocks = taosArrayInit(1, sizeof(SSDataBlock*));
taosArrayPush(dataBlocks, &pBlock);
...
...
@@ -1979,6 +1943,8 @@ void blockDebugShowDataBlocks(const SArray* dataBlocks, const char* flag) {
}
}
#endif
// for debug
char
*
dumpBlockData
(
SSDataBlock
*
pDataBlock
,
const
char
*
flag
,
char
**
pDataBuf
)
{
int32_t
size
=
2048
;
...
...
source/libs/executor/inc/executorimpl.h
浏览文件 @
1d7581f6
...
...
@@ -175,7 +175,6 @@ struct SExecTaskInfo {
int64_t
version
;
// used for stream to record wal version, why not move to sschemainfo
SStreamTaskInfo
streamInfo
;
SSchemaInfo
schemaInfo
;
STableListInfo
*
pTableInfoList
;
// this is a table list
const
char
*
sql
;
// query sql string
jmp_buf
env
;
// jump to this position when error happens.
EOPTR_EXEC_MODEL
execModel
;
// operator execution model [batch model|stream model]
...
...
@@ -323,6 +322,8 @@ typedef struct STableScanBase {
int32_t
scanFlag
;
// table scan flag to denote if it is a repeat/reverse/main scan
int32_t
dataBlockLoadFlag
;
SLimitInfo
limitInfo
;
// there are more than one table list exists in one task, if only one vnode exists.
STableListInfo
*
pTableListInfo
;
}
STableScanBase
;
typedef
struct
STableScanInfo
{
...
...
@@ -363,11 +364,12 @@ typedef struct STableMergeScanInfo {
}
STableMergeScanInfo
;
typedef
struct
STagScanInfo
{
SColumnInfo
*
pCols
;
SSDataBlock
*
pRes
;
SColMatchInfo
matchInfo
;
int32_t
curPos
;
SReadHandle
readHandle
;
SColumnInfo
*
pCols
;
SSDataBlock
*
pRes
;
SColMatchInfo
matchInfo
;
int32_t
curPos
;
SReadHandle
readHandle
;
STableListInfo
*
pTableListInfo
;
}
STagScanInfo
;
typedef
enum
EStreamScanMode
{
...
...
@@ -483,10 +485,11 @@ typedef struct SStreamScanInfo {
}
SStreamScanInfo
;
typedef
struct
{
SVnode
*
vnode
;
SSDataBlock
pRes
;
// result SSDataBlock
STsdbReader
*
dataReader
;
SSnapContext
*
sContext
;
SVnode
*
vnode
;
SSDataBlock
pRes
;
// result SSDataBlock
STsdbReader
*
dataReader
;
SSnapContext
*
sContext
;
STableListInfo
*
pTableListInfo
;
}
SStreamRawScanInfo
;
typedef
struct
STableCountScanSupp
{
...
...
@@ -515,6 +518,16 @@ typedef struct SOptrBasicInfo {
bool
mergeResultBlock
;
}
SOptrBasicInfo
;
typedef
struct
SAggOperatorInfo
{
SOptrBasicInfo
binfo
;
SAggSupporter
aggSup
;
STableQueryInfo
*
current
;
uint64_t
groupId
;
SGroupResInfo
groupResInfo
;
SExprSupp
scalarExprSup
;
bool
groupKeyOptimized
;
}
SAggOperatorInfo
;
typedef
struct
SIntervalAggOperatorInfo
{
SOptrBasicInfo
binfo
;
// basic info
SAggSupporter
aggSup
;
// aggregate supporter
...
...
@@ -731,6 +744,7 @@ void updateLoadRemoteInfo(SLoadRemoteDataInfo* pInfo, int64_t numOfRows, int3
STimeWindow
getFirstQualifiedTimeWindow
(
int64_t
ts
,
STimeWindow
*
pWindow
,
SInterval
*
pInterval
,
int32_t
order
);
SOperatorInfo
*
extractOperatorInTree
(
SOperatorInfo
*
pOperator
,
int32_t
type
,
const
char
*
id
);
int32_t
getTableScanInfo
(
SOperatorInfo
*
pOperator
,
int32_t
*
order
,
int32_t
*
scanFlag
,
bool
inheritUsOrder
);
int32_t
getBufferPgSize
(
int32_t
rowSize
,
uint32_t
*
defaultPgsz
,
uint32_t
*
defaultBufsz
);
...
...
@@ -753,11 +767,11 @@ SResultRow* doSetResultOutBufByKey(SDiskbasedBuf* pResultBuf, SResultRowInfo* pR
// clang-format off
SOperatorInfo
*
createExchangeOperatorInfo
(
void
*
pTransporter
,
SExchangePhysiNode
*
pExNode
,
SExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createTableScanOperatorInfo
(
STableScanPhysiNode
*
pTableScanNode
,
SReadHandle
*
pHandle
,
SExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createTableScanOperatorInfo
(
STableScanPhysiNode
*
pTableScanNode
,
SReadHandle
*
pHandle
,
S
TableListInfo
*
pTableList
,
S
ExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createTableMergeScanOperatorInfo
(
STableScanPhysiNode
*
pTableScanNode
,
SReadHandle
*
readHandle
,
SExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createTableMergeScanOperatorInfo
(
STableScanPhysiNode
*
pTableScanNode
,
SReadHandle
*
readHandle
,
S
TableListInfo
*
pTableListInfo
,
S
ExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createTagScanOperatorInfo
(
SReadHandle
*
pReadHandle
,
STagScanPhysiNode
*
pPhyNode
,
SExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createTagScanOperatorInfo
(
SReadHandle
*
pReadHandle
,
STagScanPhysiNode
*
pPhyNode
,
S
TableListInfo
*
pTableListInfo
,
S
ExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createSysTableScanOperatorInfo
(
void
*
readHandle
,
SSystemTableScanPhysiNode
*
pScanPhyNode
,
const
char
*
pUser
,
SExecTaskInfo
*
pTaskInfo
);
...
...
@@ -773,7 +787,7 @@ SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode*
SOperatorInfo
*
createMultiwayMergeOperatorInfo
(
SOperatorInfo
**
dowStreams
,
size_t
numStreams
,
SMergePhysiNode
*
pMergePhysiNode
,
SExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createCacherowsScanOperator
(
SLastRowScanPhysiNode
*
pTableScanNode
,
SReadHandle
*
readHandle
,
SExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createCacherowsScanOperator
(
SLastRowScanPhysiNode
*
pTableScanNode
,
SReadHandle
*
readHandle
,
S
TableListInfo
*
pTableListInfo
,
S
ExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createIntervalOperatorInfo
(
SOperatorInfo
*
downstream
,
SIntervalPhysiNode
*
pPhyNode
,
SExecTaskInfo
*
pTaskInfo
);
...
...
@@ -787,9 +801,9 @@ SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SSessionW
SOperatorInfo
*
createGroupOperatorInfo
(
SOperatorInfo
*
downstream
,
SAggPhysiNode
*
pAggNode
,
SExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createDataBlockInfoScanOperator
(
SReadHandle
*
readHandle
,
SBlockDistScanPhysiNode
*
pBlockScanNode
,
SExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createDataBlockInfoScanOperator
(
SReadHandle
*
readHandle
,
SBlockDistScanPhysiNode
*
pBlockScanNode
,
S
TableListInfo
*
pTableListInfo
,
S
ExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createStreamScanOperatorInfo
(
SReadHandle
*
pHandle
,
STableScanPhysiNode
*
pTableScanNode
,
SNode
*
pTagCond
,
SExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createStreamScanOperatorInfo
(
SReadHandle
*
pHandle
,
STableScanPhysiNode
*
pTableScanNode
,
SNode
*
pTagCond
,
S
TableListInfo
*
pTableListInfo
,
S
ExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createRawScanOperatorInfo
(
SReadHandle
*
pHandle
,
SExecTaskInfo
*
pTaskInfo
);
...
...
@@ -834,9 +848,11 @@ void setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status);
char
*
buildTaskId
(
uint64_t
taskId
,
uint64_t
queryId
);
SArray
*
getTableListInfo
(
const
SExecTaskInfo
*
pTaskInfo
);
int32_t
createExecTaskInfo
(
SSubplan
*
pPlan
,
SExecTaskInfo
**
pTaskInfo
,
SReadHandle
*
pHandle
,
uint64_t
taskId
,
int32_t
vgId
,
char
*
sql
,
EOPTR_EXEC_MODEL
model
);
int32_t
createDataSinkParam
(
SDataSinkNode
*
pNode
,
void
**
pParam
,
qTaskInfo_t
*
pTask
Info
,
SReadHandle
*
readHandle
);
int32_t
createDataSinkParam
(
SDataSinkNode
*
pNode
,
void
**
pParam
,
STableListInfo
*
pTableList
Info
,
SReadHandle
*
readHandle
);
int32_t
getOperatorExplainExecInfo
(
SOperatorInfo
*
operatorInfo
,
SArray
*
pExecInfoList
);
STimeWindow
getActiveTimeWindow
(
SDiskbasedBuf
*
pBuf
,
SResultRowInfo
*
pResultRowInfo
,
int64_t
ts
,
SInterval
*
pInterval
,
...
...
@@ -850,7 +866,6 @@ bool isInTimeWindow(STimeWindow* pWin, TSKEY ts, int64_t gap);
bool
functionNeedToExecute
(
SqlFunctionCtx
*
pCtx
);
bool
isOverdue
(
TSKEY
ts
,
STimeWindowAggSupp
*
pSup
);
bool
isCloseWindow
(
STimeWindow
*
pWin
,
STimeWindowAggSupp
*
pSup
);
bool
isDeletedWindow
(
STimeWindow
*
pWin
,
uint64_t
groupId
,
SAggSupporter
*
pSup
);
bool
isDeletedStreamWindow
(
STimeWindow
*
pWin
,
uint64_t
groupId
,
SStreamState
*
pState
,
STimeWindowAggSupp
*
pTwSup
);
void
appendOneRowToStreamSpecialBlock
(
SSDataBlock
*
pBlock
,
TSKEY
*
pStartTs
,
TSKEY
*
pEndTs
,
uint64_t
*
pUid
,
uint64_t
*
pGp
,
void
*
pTbName
);
...
...
source/libs/executor/src/aggregateoperator.c
0 → 100644
浏览文件 @
1d7581f6
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "filter.h"
#include "function.h"
#include "functionMgt.h"
#include "os.h"
#include "querynodes.h"
#include "tfill.h"
#include "tname.h"
#include "tdatablock.h"
#include "tglobal.h"
#include "tmsg.h"
#include "ttime.h"
#include "executorimpl.h"
#include "index.h"
#include "query.h"
#include "tcompare.h"
#include "thash.h"
#include "ttypes.h"
#include "vnode.h"
typedef
struct
{
bool
hasAgg
;
int32_t
numOfRows
;
int32_t
startOffset
;
}
SFunctionCtxStatus
;
static
void
destroyAggOperatorInfo
(
void
*
param
);
static
void
setExecutionContext
(
SOperatorInfo
*
pOperator
,
int32_t
numOfOutput
,
uint64_t
groupId
);
static
int32_t
createDataBlockForEmptyInput
(
SOperatorInfo
*
pOperator
,
SSDataBlock
**
ppBlock
);
static
void
destroyDataBlockForEmptyInput
(
bool
blockAllocated
,
SSDataBlock
**
ppBlock
);
static
int32_t
doOpenAggregateOptr
(
SOperatorInfo
*
pOperator
);
static
int32_t
doAggregateImpl
(
SOperatorInfo
*
pOperator
,
SqlFunctionCtx
*
pCtx
);
static
SSDataBlock
*
getAggregateResult
(
SOperatorInfo
*
pOperator
);
static
int32_t
doInitAggInfoSup
(
SAggSupporter
*
pAggSup
,
SqlFunctionCtx
*
pCtx
,
int32_t
numOfOutput
,
size_t
keyBufSize
,
const
char
*
pKey
);
static
int32_t
addNewResultRowBuf
(
SResultRow
*
pWindowRes
,
SDiskbasedBuf
*
pResultBuf
,
uint32_t
size
);
static
void
doSetTableGroupOutputBuf
(
SOperatorInfo
*
pOperator
,
int32_t
numOfOutput
,
uint64_t
groupId
);
static
void
functionCtxSave
(
SqlFunctionCtx
*
pCtx
,
SFunctionCtxStatus
*
pStatus
);
static
void
functionCtxRestore
(
SqlFunctionCtx
*
pCtx
,
SFunctionCtxStatus
*
pStatus
);
SOperatorInfo
*
createAggregateOperatorInfo
(
SOperatorInfo
*
downstream
,
SAggPhysiNode
*
pAggNode
,
SExecTaskInfo
*
pTaskInfo
)
{
SAggOperatorInfo
*
pInfo
=
taosMemoryCalloc
(
1
,
sizeof
(
SAggOperatorInfo
));
SOperatorInfo
*
pOperator
=
taosMemoryCalloc
(
1
,
sizeof
(
SOperatorInfo
));
if
(
pInfo
==
NULL
||
pOperator
==
NULL
)
{
goto
_error
;
}
SSDataBlock
*
pResBlock
=
createDataBlockFromDescNode
(
pAggNode
->
node
.
pOutputDataBlockDesc
);
initBasicInfo
(
&
pInfo
->
binfo
,
pResBlock
);
size_t
keyBufSize
=
sizeof
(
int64_t
)
+
sizeof
(
int64_t
)
+
POINTER_BYTES
;
initResultSizeInfo
(
&
pOperator
->
resultInfo
,
4096
);
int32_t
num
=
0
;
SExprInfo
*
pExprInfo
=
createExprInfo
(
pAggNode
->
pAggFuncs
,
pAggNode
->
pGroupKeys
,
&
num
);
int32_t
code
=
initAggSup
(
&
pOperator
->
exprSupp
,
&
pInfo
->
aggSup
,
pExprInfo
,
num
,
keyBufSize
,
pTaskInfo
->
id
.
str
,
pTaskInfo
->
streamInfo
.
pState
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
goto
_error
;
}
int32_t
numOfScalarExpr
=
0
;
SExprInfo
*
pScalarExprInfo
=
NULL
;
if
(
pAggNode
->
pExprs
!=
NULL
)
{
pScalarExprInfo
=
createExprInfo
(
pAggNode
->
pExprs
,
NULL
,
&
numOfScalarExpr
);
}
code
=
initExprSupp
(
&
pInfo
->
scalarExprSup
,
pScalarExprInfo
,
numOfScalarExpr
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
goto
_error
;
}
code
=
filterInitFromNode
((
SNode
*
)
pAggNode
->
node
.
pConditions
,
&
pOperator
->
exprSupp
.
pFilterInfo
,
0
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
goto
_error
;
}
pInfo
->
binfo
.
mergeResultBlock
=
pAggNode
->
mergeDataBlock
;
pInfo
->
groupKeyOptimized
=
pAggNode
->
groupKeyOptimized
;
pInfo
->
groupId
=
UINT64_MAX
;
setOperatorInfo
(
pOperator
,
"TableAggregate"
,
QUERY_NODE_PHYSICAL_PLAN_HASH_AGG
,
true
,
OP_NOT_OPENED
,
pInfo
,
pTaskInfo
);
pOperator
->
fpSet
=
createOperatorFpSet
(
doOpenAggregateOptr
,
getAggregateResult
,
NULL
,
destroyAggOperatorInfo
,
optrDefaultBufFn
,
NULL
);
if
(
downstream
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN
)
{
STableScanInfo
*
pTableScanInfo
=
downstream
->
info
;
pTableScanInfo
->
base
.
pdInfo
.
pExprSup
=
&
pOperator
->
exprSupp
;
pTableScanInfo
->
base
.
pdInfo
.
pAggSup
=
&
pInfo
->
aggSup
;
}
code
=
appendDownstream
(
pOperator
,
&
downstream
,
1
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
goto
_error
;
}
return
pOperator
;
_error:
if
(
pInfo
!=
NULL
)
{
destroyAggOperatorInfo
(
pInfo
);
}
if
(
pOperator
!=
NULL
)
{
cleanupExprSupp
(
&
pOperator
->
exprSupp
);
}
taosMemoryFreeClear
(
pOperator
);
pTaskInfo
->
code
=
code
;
return
NULL
;
}
void
destroyAggOperatorInfo
(
void
*
param
)
{
SAggOperatorInfo
*
pInfo
=
(
SAggOperatorInfo
*
)
param
;
cleanupBasicInfo
(
&
pInfo
->
binfo
);
cleanupAggSup
(
&
pInfo
->
aggSup
);
cleanupExprSupp
(
&
pInfo
->
scalarExprSup
);
cleanupGroupResInfo
(
&
pInfo
->
groupResInfo
);
taosMemoryFreeClear
(
param
);
}
// this is a blocking operator
int32_t
doOpenAggregateOptr
(
SOperatorInfo
*
pOperator
)
{
if
(
OPTR_IS_OPENED
(
pOperator
))
{
return
TSDB_CODE_SUCCESS
;
}
SExecTaskInfo
*
pTaskInfo
=
pOperator
->
pTaskInfo
;
SAggOperatorInfo
*
pAggInfo
=
pOperator
->
info
;
SExprSupp
*
pSup
=
&
pOperator
->
exprSupp
;
SOperatorInfo
*
downstream
=
pOperator
->
pDownstream
[
0
];
int64_t
st
=
taosGetTimestampUs
();
int32_t
order
=
TSDB_ORDER_ASC
;
int32_t
scanFlag
=
MAIN_SCAN
;
bool
hasValidBlock
=
false
;
bool
blockAllocated
=
false
;
while
(
1
)
{
SSDataBlock
*
pBlock
=
downstream
->
fpSet
.
getNextFn
(
downstream
);
if
(
pBlock
==
NULL
)
{
if
(
!
hasValidBlock
)
{
createDataBlockForEmptyInput
(
pOperator
,
&
pBlock
);
if
(
pBlock
==
NULL
)
{
break
;
}
blockAllocated
=
true
;
}
else
{
break
;
}
}
hasValidBlock
=
true
;
int32_t
code
=
getTableScanInfo
(
pOperator
,
&
order
,
&
scanFlag
,
false
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
destroyDataBlockForEmptyInput
(
blockAllocated
,
&
pBlock
);
T_LONG_JMP
(
pTaskInfo
->
env
,
code
);
}
// there is an scalar expression that needs to be calculated before apply the group aggregation.
if
(
pAggInfo
->
scalarExprSup
.
pExprInfo
!=
NULL
&&
!
blockAllocated
)
{
SExprSupp
*
pSup1
=
&
pAggInfo
->
scalarExprSup
;
code
=
projectApplyFunctions
(
pSup1
->
pExprInfo
,
pBlock
,
pBlock
,
pSup1
->
pCtx
,
pSup1
->
numOfExprs
,
NULL
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
destroyDataBlockForEmptyInput
(
blockAllocated
,
&
pBlock
);
T_LONG_JMP
(
pTaskInfo
->
env
,
code
);
}
}
// the pDataBlock are always the same one, no need to call this again
setExecutionContext
(
pOperator
,
pOperator
->
exprSupp
.
numOfExprs
,
pBlock
->
info
.
id
.
groupId
);
setInputDataBlock
(
pSup
,
pBlock
,
order
,
scanFlag
,
true
);
code
=
doAggregateImpl
(
pOperator
,
pSup
->
pCtx
);
if
(
code
!=
0
)
{
destroyDataBlockForEmptyInput
(
blockAllocated
,
&
pBlock
);
T_LONG_JMP
(
pTaskInfo
->
env
,
code
);
}
destroyDataBlockForEmptyInput
(
blockAllocated
,
&
pBlock
);
}
// the downstream operator may return with error code, so let's check the code before generating results.
if
(
pTaskInfo
->
code
!=
TSDB_CODE_SUCCESS
)
{
T_LONG_JMP
(
pTaskInfo
->
env
,
pTaskInfo
->
code
);
}
initGroupedResultInfo
(
&
pAggInfo
->
groupResInfo
,
pAggInfo
->
aggSup
.
pResultRowHashTable
,
0
);
OPTR_SET_OPENED
(
pOperator
);
pOperator
->
cost
.
openCost
=
(
taosGetTimestampUs
()
-
st
)
/
1000
.
0
;
return
pTaskInfo
->
code
;
}
SSDataBlock
*
getAggregateResult
(
SOperatorInfo
*
pOperator
)
{
SAggOperatorInfo
*
pAggInfo
=
pOperator
->
info
;
SOptrBasicInfo
*
pInfo
=
&
pAggInfo
->
binfo
;
if
(
pOperator
->
status
==
OP_EXEC_DONE
)
{
return
NULL
;
}
SExecTaskInfo
*
pTaskInfo
=
pOperator
->
pTaskInfo
;
pTaskInfo
->
code
=
pOperator
->
fpSet
.
_openFn
(
pOperator
);
if
(
pTaskInfo
->
code
!=
TSDB_CODE_SUCCESS
)
{
setOperatorCompleted
(
pOperator
);
return
NULL
;
}
blockDataEnsureCapacity
(
pInfo
->
pRes
,
pOperator
->
resultInfo
.
capacity
);
while
(
1
)
{
doBuildResultDatablock
(
pOperator
,
pInfo
,
&
pAggInfo
->
groupResInfo
,
pAggInfo
->
aggSup
.
pResultBuf
);
doFilter
(
pInfo
->
pRes
,
pOperator
->
exprSupp
.
pFilterInfo
,
NULL
);
if
(
!
hasRemainResults
(
&
pAggInfo
->
groupResInfo
))
{
setOperatorCompleted
(
pOperator
);
break
;
}
if
(
pInfo
->
pRes
->
info
.
rows
>
0
)
{
break
;
}
}
size_t
rows
=
blockDataGetNumOfRows
(
pInfo
->
pRes
);
pOperator
->
resultInfo
.
totalRows
+=
rows
;
return
(
rows
==
0
)
?
NULL
:
pInfo
->
pRes
;
}
int32_t
doAggregateImpl
(
SOperatorInfo
*
pOperator
,
SqlFunctionCtx
*
pCtx
)
{
for
(
int32_t
k
=
0
;
k
<
pOperator
->
exprSupp
.
numOfExprs
;
++
k
)
{
if
(
functionNeedToExecute
(
&
pCtx
[
k
]))
{
// todo add a dummy funtion to avoid process check
if
(
pCtx
[
k
].
fpSet
.
process
==
NULL
)
{
continue
;
}
int32_t
code
=
pCtx
[
k
].
fpSet
.
process
(
&
pCtx
[
k
]);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
qError
(
"%s aggregate function error happens, code: %s"
,
GET_TASKID
(
pOperator
->
pTaskInfo
),
tstrerror
(
code
));
return
code
;
}
}
}
return
TSDB_CODE_SUCCESS
;
}
static
int32_t
createDataBlockForEmptyInput
(
SOperatorInfo
*
pOperator
,
SSDataBlock
**
ppBlock
)
{
if
(
!
tsCountAlwaysReturnValue
)
{
return
TSDB_CODE_SUCCESS
;
}
SAggOperatorInfo
*
pAggInfo
=
pOperator
->
info
;
if
(
pAggInfo
->
groupKeyOptimized
)
{
return
TSDB_CODE_SUCCESS
;
}
SOperatorInfo
*
downstream
=
pOperator
->
pDownstream
[
0
];
if
(
downstream
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_PARTITION
||
(
downstream
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN
&&
((
STableScanInfo
*
)
downstream
->
info
)
->
hasGroupByTag
==
true
))
{
return
TSDB_CODE_SUCCESS
;
}
SqlFunctionCtx
*
pCtx
=
pOperator
->
exprSupp
.
pCtx
;
bool
hasCountFunc
=
false
;
for
(
int32_t
i
=
0
;
i
<
pOperator
->
exprSupp
.
numOfExprs
;
++
i
)
{
const
char
*
pName
=
pCtx
[
i
].
pExpr
->
pExpr
->
_function
.
functionName
;
if
((
strcmp
(
pName
,
"count"
)
==
0
)
||
(
strcmp
(
pName
,
"hyperloglog"
)
==
0
)
||
(
strcmp
(
pName
,
"_hyperloglog_partial"
)
==
0
)
||
(
strcmp
(
pName
,
"_hyperloglog_merge"
)
==
0
))
{
hasCountFunc
=
true
;
break
;
}
}
if
(
!
hasCountFunc
)
{
return
TSDB_CODE_SUCCESS
;
}
SSDataBlock
*
pBlock
=
createDataBlock
();
pBlock
->
info
.
rows
=
1
;
pBlock
->
info
.
capacity
=
0
;
for
(
int32_t
i
=
0
;
i
<
pOperator
->
exprSupp
.
numOfExprs
;
++
i
)
{
SColumnInfoData
colInfo
=
{
0
};
colInfo
.
hasNull
=
true
;
colInfo
.
info
.
type
=
TSDB_DATA_TYPE_NULL
;
colInfo
.
info
.
bytes
=
1
;
SExprInfo
*
pOneExpr
=
&
pOperator
->
exprSupp
.
pExprInfo
[
i
];
for
(
int32_t
j
=
0
;
j
<
pOneExpr
->
base
.
numOfParams
;
++
j
)
{
SFunctParam
*
pFuncParam
=
&
pOneExpr
->
base
.
pParam
[
j
];
if
(
pFuncParam
->
type
==
FUNC_PARAM_TYPE_COLUMN
)
{
int32_t
slotId
=
pFuncParam
->
pCol
->
slotId
;
int32_t
numOfCols
=
taosArrayGetSize
(
pBlock
->
pDataBlock
);
if
(
slotId
>=
numOfCols
)
{
taosArrayEnsureCap
(
pBlock
->
pDataBlock
,
slotId
+
1
);
for
(
int32_t
k
=
numOfCols
;
k
<
slotId
+
1
;
++
k
)
{
taosArrayPush
(
pBlock
->
pDataBlock
,
&
colInfo
);
}
}
}
else
if
(
pFuncParam
->
type
==
FUNC_PARAM_TYPE_VALUE
)
{
// do nothing
}
}
}
blockDataEnsureCapacity
(
pBlock
,
pBlock
->
info
.
rows
);
*
ppBlock
=
pBlock
;
return
TSDB_CODE_SUCCESS
;
}
void
destroyDataBlockForEmptyInput
(
bool
blockAllocated
,
SSDataBlock
**
ppBlock
)
{
if
(
!
blockAllocated
)
{
return
;
}
blockDataDestroy
(
*
ppBlock
);
*
ppBlock
=
NULL
;
}
void
setExecutionContext
(
SOperatorInfo
*
pOperator
,
int32_t
numOfOutput
,
uint64_t
groupId
)
{
SAggOperatorInfo
*
pAggInfo
=
pOperator
->
info
;
if
(
pAggInfo
->
groupId
!=
UINT64_MAX
&&
pAggInfo
->
groupId
==
groupId
)
{
return
;
}
doSetTableGroupOutputBuf
(
pOperator
,
numOfOutput
,
groupId
);
// record the current active group id
pAggInfo
->
groupId
=
groupId
;
}
void
doSetTableGroupOutputBuf
(
SOperatorInfo
*
pOperator
,
int32_t
numOfOutput
,
uint64_t
groupId
)
{
// for simple group by query without interval, all the tables belong to one group result.
SExecTaskInfo
*
pTaskInfo
=
pOperator
->
pTaskInfo
;
SAggOperatorInfo
*
pAggInfo
=
pOperator
->
info
;
SResultRowInfo
*
pResultRowInfo
=
&
pAggInfo
->
binfo
.
resultRowInfo
;
SqlFunctionCtx
*
pCtx
=
pOperator
->
exprSupp
.
pCtx
;
int32_t
*
rowEntryInfoOffset
=
pOperator
->
exprSupp
.
rowEntryInfoOffset
;
SResultRow
*
pResultRow
=
doSetResultOutBufByKey
(
pAggInfo
->
aggSup
.
pResultBuf
,
pResultRowInfo
,
(
char
*
)
&
groupId
,
sizeof
(
groupId
),
true
,
groupId
,
pTaskInfo
,
false
,
&
pAggInfo
->
aggSup
,
true
);
/*
* not assign result buffer yet, add new result buffer
* all group belong to one result set, and each group result has different group id so set the id to be one
*/
if
(
pResultRow
->
pageId
==
-
1
)
{
int32_t
ret
=
addNewResultRowBuf
(
pResultRow
,
pAggInfo
->
aggSup
.
pResultBuf
,
pAggInfo
->
binfo
.
pRes
->
info
.
rowSize
);
if
(
ret
!=
TSDB_CODE_SUCCESS
)
{
T_LONG_JMP
(
pTaskInfo
->
env
,
terrno
);
}
}
setResultRowInitCtx
(
pResultRow
,
pCtx
,
numOfOutput
,
rowEntryInfoOffset
);
}
// a new buffer page for each table. Needs to opt this design
int32_t
addNewResultRowBuf
(
SResultRow
*
pWindowRes
,
SDiskbasedBuf
*
pResultBuf
,
uint32_t
size
)
{
if
(
pWindowRes
->
pageId
!=
-
1
)
{
return
0
;
}
SFilePage
*
pData
=
NULL
;
// in the first scan, new space needed for results
int32_t
pageId
=
-
1
;
SArray
*
list
=
getDataBufPagesIdList
(
pResultBuf
);
if
(
taosArrayGetSize
(
list
)
==
0
)
{
pData
=
getNewBufPage
(
pResultBuf
,
&
pageId
);
pData
->
num
=
sizeof
(
SFilePage
);
}
else
{
SPageInfo
*
pi
=
getLastPageInfo
(
list
);
pData
=
getBufPage
(
pResultBuf
,
getPageId
(
pi
));
if
(
pData
==
NULL
)
{
qError
(
"failed to get buffer, code:%s"
,
tstrerror
(
terrno
));
return
terrno
;
}
pageId
=
getPageId
(
pi
);
if
(
pData
->
num
+
size
>
getBufPageSize
(
pResultBuf
))
{
// release current page first, and prepare the next one
releaseBufPageInfo
(
pResultBuf
,
pi
);
pData
=
getNewBufPage
(
pResultBuf
,
&
pageId
);
if
(
pData
!=
NULL
)
{
pData
->
num
=
sizeof
(
SFilePage
);
}
}
}
if
(
pData
==
NULL
)
{
return
-
1
;
}
// set the number of rows in current disk page
if
(
pWindowRes
->
pageId
==
-
1
)
{
// not allocated yet, allocate new buffer
pWindowRes
->
pageId
=
pageId
;
pWindowRes
->
offset
=
(
int32_t
)
pData
->
num
;
pData
->
num
+=
size
;
}
return
0
;
}
int32_t
doInitAggInfoSup
(
SAggSupporter
*
pAggSup
,
SqlFunctionCtx
*
pCtx
,
int32_t
numOfOutput
,
size_t
keyBufSize
,
const
char
*
pKey
)
{
int32_t
code
=
0
;
// _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
pAggSup
->
currentPageId
=
-
1
;
pAggSup
->
resultRowSize
=
getResultRowSize
(
pCtx
,
numOfOutput
);
pAggSup
->
keyBuf
=
taosMemoryCalloc
(
1
,
keyBufSize
+
POINTER_BYTES
+
sizeof
(
int64_t
));
pAggSup
->
pResultRowHashTable
=
tSimpleHashInit
(
100
,
taosFastHash
);
if
(
pAggSup
->
keyBuf
==
NULL
||
pAggSup
->
pResultRowHashTable
==
NULL
)
{
return
TSDB_CODE_OUT_OF_MEMORY
;
}
uint32_t
defaultPgsz
=
0
;
uint32_t
defaultBufsz
=
0
;
getBufferPgSize
(
pAggSup
->
resultRowSize
,
&
defaultPgsz
,
&
defaultBufsz
);
if
(
!
osTempSpaceAvailable
())
{
code
=
TSDB_CODE_NO_AVAIL_DISK
;
qError
(
"Init stream agg supporter failed since %s, %s"
,
terrstr
(
code
),
pKey
);
return
code
;
}
code
=
createDiskbasedBuf
(
&
pAggSup
->
pResultBuf
,
defaultPgsz
,
defaultBufsz
,
pKey
,
tsTempDir
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
qError
(
"Create agg result buf failed since %s, %s"
,
tstrerror
(
code
),
pKey
);
return
code
;
}
return
code
;
}
void
cleanupAggSup
(
SAggSupporter
*
pAggSup
)
{
taosMemoryFreeClear
(
pAggSup
->
keyBuf
);
tSimpleHashCleanup
(
pAggSup
->
pResultRowHashTable
);
destroyDiskbasedBuf
(
pAggSup
->
pResultBuf
);
}
int32_t
initAggSup
(
SExprSupp
*
pSup
,
SAggSupporter
*
pAggSup
,
SExprInfo
*
pExprInfo
,
int32_t
numOfCols
,
size_t
keyBufSize
,
const
char
*
pkey
,
void
*
pState
)
{
int32_t
code
=
initExprSupp
(
pSup
,
pExprInfo
,
numOfCols
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
return
code
;
}
code
=
doInitAggInfoSup
(
pAggSup
,
pSup
->
pCtx
,
numOfCols
,
keyBufSize
,
pkey
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
return
code
;
}
for
(
int32_t
i
=
0
;
i
<
numOfCols
;
++
i
)
{
if
(
pState
)
{
pSup
->
pCtx
[
i
].
saveHandle
.
pBuf
=
NULL
;
pSup
->
pCtx
[
i
].
saveHandle
.
pState
=
pState
;
pSup
->
pCtx
[
i
].
exprIdx
=
i
;
}
else
{
pSup
->
pCtx
[
i
].
saveHandle
.
pBuf
=
pAggSup
->
pResultBuf
;
}
}
return
TSDB_CODE_SUCCESS
;
}
void
applyAggFunctionOnPartialTuples
(
SExecTaskInfo
*
taskInfo
,
SqlFunctionCtx
*
pCtx
,
SColumnInfoData
*
pTimeWindowData
,
int32_t
offset
,
int32_t
forwardStep
,
int32_t
numOfTotal
,
int32_t
numOfOutput
)
{
for
(
int32_t
k
=
0
;
k
<
numOfOutput
;
++
k
)
{
// keep it temporarily
SFunctionCtxStatus
status
=
{
0
};
functionCtxSave
(
&
pCtx
[
k
],
&
status
);
pCtx
[
k
].
input
.
startRowIndex
=
offset
;
pCtx
[
k
].
input
.
numOfRows
=
forwardStep
;
// not a whole block involved in query processing, statistics data can not be used
// NOTE: the original value of isSet have been changed here
if
(
pCtx
[
k
].
input
.
colDataSMAIsSet
&&
forwardStep
<
numOfTotal
)
{
pCtx
[
k
].
input
.
colDataSMAIsSet
=
false
;
}
if
(
pCtx
[
k
].
isPseudoFunc
)
{
SResultRowEntryInfo
*
pEntryInfo
=
GET_RES_INFO
(
&
pCtx
[
k
]);
char
*
p
=
GET_ROWCELL_INTERBUF
(
pEntryInfo
);
SColumnInfoData
idata
=
{
0
};
idata
.
info
.
type
=
TSDB_DATA_TYPE_BIGINT
;
idata
.
info
.
bytes
=
tDataTypes
[
TSDB_DATA_TYPE_BIGINT
].
bytes
;
idata
.
pData
=
p
;
SScalarParam
out
=
{.
columnData
=
&
idata
};
SScalarParam
tw
=
{.
numOfRows
=
5
,
.
columnData
=
pTimeWindowData
};
pCtx
[
k
].
sfp
.
process
(
&
tw
,
1
,
&
out
);
pEntryInfo
->
numOfRes
=
1
;
}
else
{
int32_t
code
=
TSDB_CODE_SUCCESS
;
if
(
functionNeedToExecute
(
&
pCtx
[
k
])
&&
pCtx
[
k
].
fpSet
.
process
!=
NULL
)
{
code
=
pCtx
[
k
].
fpSet
.
process
(
&
pCtx
[
k
]);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
qError
(
"%s apply functions error, code: %s"
,
GET_TASKID
(
taskInfo
),
tstrerror
(
code
));
taskInfo
->
code
=
code
;
T_LONG_JMP
(
taskInfo
->
env
,
code
);
}
}
// restore it
functionCtxRestore
(
&
pCtx
[
k
],
&
status
);
}
}
}
void
functionCtxSave
(
SqlFunctionCtx
*
pCtx
,
SFunctionCtxStatus
*
pStatus
)
{
pStatus
->
hasAgg
=
pCtx
->
input
.
colDataSMAIsSet
;
pStatus
->
numOfRows
=
pCtx
->
input
.
numOfRows
;
pStatus
->
startOffset
=
pCtx
->
input
.
startRowIndex
;
}
void
functionCtxRestore
(
SqlFunctionCtx
*
pCtx
,
SFunctionCtxStatus
*
pStatus
)
{
pCtx
->
input
.
colDataSMAIsSet
=
pStatus
->
hasAgg
;
pCtx
->
input
.
numOfRows
=
pStatus
->
numOfRows
;
pCtx
->
input
.
startRowIndex
=
pStatus
->
startOffset
;
}
\ No newline at end of file
source/libs/executor/src/cachescanoperator.c
浏览文件 @
1d7581f6
...
...
@@ -37,6 +37,7 @@ typedef struct SCacheRowsScanInfo {
SSDataBlock
*
pBufferredRes
;
SArray
*
pUidList
;
int32_t
indexOfBufferedRes
;
STableListInfo
*
pTableList
;
}
SCacheRowsScanInfo
;
static
SSDataBlock
*
doScanCache
(
SOperatorInfo
*
pOperator
);
...
...
@@ -47,15 +48,17 @@ static int32_t removeRedundantTsCol(SLastRowScanPhysiNode* pScanNode, SColM
#define SCAN_ROW_TYPE(_t) ((_t)? CACHESCAN_RETRIEVE_LAST : CACHESCAN_RETRIEVE_LAST_ROW)
SOperatorInfo
*
createCacherowsScanOperator
(
SLastRowScanPhysiNode
*
pScanNode
,
SReadHandle
*
readHandle
,
SExecTaskInfo
*
pTaskInfo
)
{
S
TableListInfo
*
pTableListInfo
,
S
ExecTaskInfo
*
pTaskInfo
)
{
int32_t
code
=
TSDB_CODE_SUCCESS
;
SCacheRowsScanInfo
*
pInfo
=
taosMemoryCalloc
(
1
,
sizeof
(
SCacheRowsScanInfo
));
SOperatorInfo
*
pOperator
=
taosMemoryCalloc
(
1
,
sizeof
(
SOperatorInfo
));
if
(
pInfo
==
NULL
||
pOperator
==
NULL
)
{
code
=
TSDB_CODE_OUT_OF_MEMORY
;
tableListDestroy
(
pTableListInfo
);
goto
_error
;
}
pInfo
->
pTableList
=
pTableListInfo
;
pInfo
->
readHandle
=
*
readHandle
;
SDataBlockDescNode
*
pDescNode
=
pScanNode
->
scan
.
node
.
pOutputDataBlockDesc
;
...
...
@@ -75,20 +78,18 @@ SOperatorInfo* createCacherowsScanOperator(SLastRowScanPhysiNode* pScanNode, SRe
goto
_error
;
}
STableListInfo
*
pTableList
=
pTaskInfo
->
pTableInfoList
;
int32_t
totalTables
=
tableListGetSize
(
pTableList
);
int32_t
totalTables
=
tableListGetSize
(
pTableListInfo
);
int32_t
capacity
=
0
;
pInfo
->
pUidList
=
taosArrayInit
(
4
,
sizeof
(
int64_t
));
// partition by tbname
if
(
oneTableForEachGroup
(
pTableList
)
||
(
totalTables
==
1
))
{
if
(
oneTableForEachGroup
(
pTableList
Info
)
||
(
totalTables
==
1
))
{
pInfo
->
retrieveType
=
CACHESCAN_RETRIEVE_TYPE_ALL
|
SCAN_ROW_TYPE
(
pScanNode
->
ignoreNull
);
STableKeyInfo
*
pList
=
tableListGetInfo
(
pTableList
,
0
);
STableKeyInfo
*
pList
=
tableListGetInfo
(
pTableList
Info
,
0
);
uint64_t
suid
=
tableListGetSuid
(
pTableList
);
uint64_t
suid
=
tableListGetSuid
(
pTableList
Info
);
code
=
tsdbCacherowsReaderOpen
(
pInfo
->
readHandle
.
vnode
,
pInfo
->
retrieveType
,
pList
,
totalTables
,
taosArrayGetSize
(
pInfo
->
matchInfo
.
pList
),
suid
,
&
pInfo
->
pLastrowReader
,
pTaskInfo
->
id
.
str
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
...
...
@@ -136,7 +137,7 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
SCacheRowsScanInfo
*
pInfo
=
pOperator
->
info
;
SExecTaskInfo
*
pTaskInfo
=
pOperator
->
pTaskInfo
;
STableListInfo
*
pTableList
=
p
TaskInfo
->
pTableInfo
List
;
STableListInfo
*
pTableList
=
p
Info
->
pTable
List
;
uint64_t
suid
=
tableListGetSuid
(
pTableList
);
int32_t
size
=
tableListGetSize
(
pTableList
);
...
...
@@ -281,6 +282,7 @@ void destroyCacheScanOperator(void* param) {
taosMemoryFree
(
pInfo
->
pSlotIds
);
taosArrayDestroy
(
pInfo
->
pUidList
);
taosArrayDestroy
(
pInfo
->
matchInfo
.
pList
);
tableListDestroy
(
pInfo
->
pTableList
);
if
(
pInfo
->
pLastrowReader
!=
NULL
)
{
pInfo
->
pLastrowReader
=
tsdbCacherowsReaderClose
(
pInfo
->
pLastrowReader
);
...
...
source/libs/executor/src/exchangeoperator.c
浏览文件 @
1d7581f6
...
...
@@ -70,7 +70,7 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn
tsem_wait
(
&
pExchangeInfo
->
ready
);
if
(
isTaskKilled
(
pTaskInfo
))
{
longjmp
(
pTaskInfo
->
env
,
pTaskInfo
->
code
);
T_LONG_JMP
(
pTaskInfo
->
env
,
pTaskInfo
->
code
);
}
for
(
int32_t
i
=
0
;
i
<
totalSources
;
++
i
)
{
...
...
@@ -576,7 +576,7 @@ int32_t prepareConcurrentlyLoad(SOperatorInfo* pOperator) {
pOperator
->
status
=
OP_RES_TO_RETURN
;
pOperator
->
cost
.
openCost
=
taosGetTimestampUs
()
-
startTs
;
if
(
isTaskKilled
(
pTaskInfo
))
{
longjmp
(
pTaskInfo
->
env
,
pTaskInfo
->
code
);
T_LONG_JMP
(
pTaskInfo
->
env
,
pTaskInfo
->
code
);
}
return
TSDB_CODE_SUCCESS
;
...
...
@@ -629,7 +629,7 @@ int32_t seqLoadRemoteData(SOperatorInfo* pOperator) {
doSendFetchDataRequest
(
pExchangeInfo
,
pTaskInfo
,
pExchangeInfo
->
current
);
tsem_wait
(
&
pExchangeInfo
->
ready
);
if
(
isTaskKilled
(
pTaskInfo
))
{
longjmp
(
pTaskInfo
->
env
,
pTaskInfo
->
code
);
T_LONG_JMP
(
pTaskInfo
->
env
,
pTaskInfo
->
code
);
}
SDownstreamSourceNode
*
pSource
=
taosArrayGet
(
pExchangeInfo
->
pSources
,
pExchangeInfo
->
current
);
...
...
source/libs/executor/src/executor.c
浏览文件 @
1d7581f6
...
...
@@ -404,7 +404,7 @@ int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, const SArray* tableIdList, bo
}
}
STableListInfo
*
pTableListInfo
=
pTaskInfo
->
pTableInfoList
;
STableListInfo
*
pTableListInfo
=
((
STableScanInfo
*
)
pScanInfo
->
pTableScanOp
->
info
)
->
base
.
pTableListInfo
;
taosWLockLatch
(
&
pTaskInfo
->
lock
);
for
(
int32_t
i
=
0
;
i
<
numOfQualifiedTables
;
++
i
)
{
...
...
@@ -485,9 +485,7 @@ int32_t qGetQueryTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, char* table
int32_t
qCreateExecTask
(
SReadHandle
*
readHandle
,
int32_t
vgId
,
uint64_t
taskId
,
SSubplan
*
pSubplan
,
qTaskInfo_t
*
pTaskInfo
,
DataSinkHandle
*
handle
,
char
*
sql
,
EOPTR_EXEC_MODEL
model
)
{
assert
(
pSubplan
!=
NULL
);
SExecTaskInfo
**
pTask
=
(
SExecTaskInfo
**
)
pTaskInfo
;
taosThreadOnce
(
&
initPoolOnce
,
initRefPool
);
qDebug
(
"start to create task, TID:0x%"
PRIx64
" QID:0x%"
PRIx64
", vgId:%d"
,
taskId
,
pSubplan
->
id
.
queryId
,
vgId
);
...
...
@@ -507,7 +505,12 @@ int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId,
if
(
handle
)
{
void
*
pSinkParam
=
NULL
;
code
=
createDataSinkParam
(
pSubplan
->
pDataSink
,
&
pSinkParam
,
pTaskInfo
,
readHandle
);
SArray
*
pInfoList
=
getTableListInfo
(
*
pTask
);
STableListInfo
*
pTableListInfo
=
taosArrayGetP
(
pInfoList
,
0
);
taosArrayDestroy
(
pInfoList
);
code
=
createDataSinkParam
(
pSubplan
->
pDataSink
,
&
pSinkParam
,
pTableListInfo
,
readHandle
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
qError
(
"failed to createDataSinkParam, vgId:%d, code:%s, %s"
,
vgId
,
tstrerror
(
code
),
(
*
pTask
)
->
id
.
str
);
goto
_error
;
...
...
@@ -1083,7 +1086,6 @@ void qStreamSetOpen(qTaskInfo_t tinfo) {
int32_t
qStreamPrepareScan
(
qTaskInfo_t
tinfo
,
STqOffsetVal
*
pOffset
,
int8_t
subType
)
{
SExecTaskInfo
*
pTaskInfo
=
(
SExecTaskInfo
*
)
tinfo
;
SOperatorInfo
*
pOperator
=
pTaskInfo
->
pRoot
;
STableListInfo
*
pTableListInfo
=
pTaskInfo
->
pTableInfoList
;
const
char
*
id
=
GET_TASKID
(
pTaskInfo
);
// if pOffset equal to current offset, means continue consume
...
...
@@ -1092,17 +1094,14 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
}
if
(
subType
==
TOPIC_SUB_TYPE__COLUMN
)
{
if
(
pOperator
->
operatorType
!=
QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN
)
{
if
(
pOperator
->
numOfDownstream
!=
1
)
{
qError
(
"invalid operator, number of downstream:%d, %s"
,
pOperator
->
numOfDownstream
,
id
);
return
-
1
;
}
pOperator
=
pOperator
->
pDownstream
[
0
];
pOperator
=
extractOperatorInTree
(
pOperator
,
QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN
,
id
);
if
(
pOperator
==
NULL
)
{
return
-
1
;
}
SStreamScanInfo
*
pInfo
=
pOperator
->
info
;
STableScanInfo
*
pScanInfo
=
pInfo
->
pTableScanOp
->
info
;
STableScanBase
*
pScanBaseInfo
=
&
pScanInfo
->
base
;
STableListInfo
*
pTableListInfo
=
pScanBaseInfo
->
pTableListInfo
;
if
(
pOffset
->
type
==
TMQ_OFFSET__LOG
)
{
tsdbReaderClose
(
pScanBaseInfo
->
dataReader
);
...
...
@@ -1187,9 +1186,14 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
}
}
else
{
// subType == TOPIC_SUB_TYPE__TABLE/TOPIC_SUB_TYPE__DB
if
(
pOffset
->
type
==
TMQ_OFFSET__SNAPSHOT_DATA
)
{
SStreamRawScanInfo
*
pInfo
=
pOperator
->
info
;
SSnapContext
*
sContext
=
pInfo
->
sContext
;
SOperatorInfo
*
p
=
extractOperatorInTree
(
pOperator
,
QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN
,
id
);
STableListInfo
*
pTableListInfo
=
((
SStreamRawScanInfo
*
)(
p
->
info
))
->
pTableListInfo
;
if
(
setForSnapShot
(
sContext
,
pOffset
->
uid
)
!=
0
)
{
qError
(
"setDataForSnapShot error. uid:%"
PRId64
" , %s"
,
pOffset
->
uid
,
id
);
return
-
1
;
...
...
source/libs/executor/src/executorimpl.c
浏览文件 @
1d7581f6
此差异已折叠。
点击以展开。
source/libs/executor/src/scanoperator.c
浏览文件 @
1d7581f6
...
...
@@ -676,7 +676,7 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
}
if
(
pBlock
->
info
.
id
.
uid
)
{
pBlock
->
info
.
id
.
groupId
=
getTableGroupId
(
pTa
skInfo
->
pTableInfoList
,
pBlock
->
info
.
id
.
uid
);
pBlock
->
info
.
id
.
groupId
=
getTableGroupId
(
pTa
bleScanInfo
->
base
.
pTableListInfo
,
pBlock
->
info
.
id
.
uid
);
}
uint32_t
status
=
0
;
...
...
@@ -771,7 +771,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
// scan table one by one sequentially
if
(
pInfo
->
scanMode
==
TABLE_SCAN__TABLE_ORDER
)
{
int32_t
numOfTables
=
0
;
//tableListGetSize(pTaskInfo->pTable
InfoList
);
int32_t
numOfTables
=
0
;
//tableListGetSize(pTaskInfo->pTable
ListInfo
);
STableKeyInfo
tInfo
=
{
0
};
while
(
1
)
{
...
...
@@ -784,7 +784,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
pInfo
->
currentTable
++
;
taosRLockLatch
(
&
pTaskInfo
->
lock
);
numOfTables
=
tableListGetSize
(
p
TaskInfo
->
pTableInfoList
);
numOfTables
=
tableListGetSize
(
p
Info
->
base
.
pTableListInfo
);
if
(
pInfo
->
currentTable
>=
numOfTables
)
{
qDebug
(
"all table checked in table list, total:%d, return NULL, %s"
,
numOfTables
,
GET_TASKID
(
pTaskInfo
));
...
...
@@ -792,7 +792,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
return
NULL
;
}
tInfo
=
*
(
STableKeyInfo
*
)
tableListGetInfo
(
p
TaskInfo
->
pTableInfoList
,
pInfo
->
currentTable
);
tInfo
=
*
(
STableKeyInfo
*
)
tableListGetInfo
(
p
Info
->
base
.
pTableListInfo
,
pInfo
->
currentTable
);
taosRUnLockLatch
(
&
pTaskInfo
->
lock
);
tsdbSetTableList
(
pInfo
->
base
.
dataReader
,
&
tInfo
,
1
);
...
...
@@ -804,14 +804,14 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
}
}
else
{
// scan table group by group sequentially
if
(
pInfo
->
currentGroupId
==
-
1
)
{
if
((
++
pInfo
->
currentGroupId
)
>=
tableListGetOutputGroups
(
p
TaskInfo
->
pTableInfoList
))
{
if
((
++
pInfo
->
currentGroupId
)
>=
tableListGetOutputGroups
(
p
Info
->
base
.
pTableListInfo
))
{
setOperatorCompleted
(
pOperator
);
return
NULL
;
}
int32_t
num
=
0
;
STableKeyInfo
*
pList
=
NULL
;
tableListGetGroupList
(
p
TaskInfo
->
pTableInfoList
,
pInfo
->
currentGroupId
,
&
pList
,
&
num
);
tableListGetGroupList
(
p
Info
->
base
.
pTableListInfo
,
pInfo
->
currentGroupId
,
&
pList
,
&
num
);
ASSERT
(
pInfo
->
base
.
dataReader
==
NULL
);
int32_t
code
=
tsdbReaderOpen
(
pInfo
->
base
.
readHandle
.
vnode
,
&
pInfo
->
base
.
cond
,
pList
,
num
,
pInfo
->
pResBlock
,
...
...
@@ -830,7 +830,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
return
result
;
}
if
((
++
pInfo
->
currentGroupId
)
>=
tableListGetOutputGroups
(
p
TaskInfo
->
pTableInfoList
))
{
if
((
++
pInfo
->
currentGroupId
)
>=
tableListGetOutputGroups
(
p
Info
->
base
.
pTableListInfo
))
{
setOperatorCompleted
(
pOperator
);
return
NULL
;
}
...
...
@@ -841,7 +841,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
int32_t
num
=
0
;
STableKeyInfo
*
pList
=
NULL
;
tableListGetGroupList
(
p
TaskInfo
->
pTableInfoList
,
pInfo
->
currentGroupId
,
&
pList
,
&
num
);
tableListGetGroupList
(
p
Info
->
base
.
pTableListInfo
,
pInfo
->
currentGroupId
,
&
pList
,
&
num
);
tsdbSetTableList
(
pInfo
->
base
.
dataReader
,
pList
,
num
);
tsdbReaderReset
(
pInfo
->
base
.
dataReader
,
&
pInfo
->
base
.
cond
);
...
...
@@ -866,25 +866,30 @@ static int32_t getTableScannerExecInfo(struct SOperatorInfo* pOptr, void** pOptr
return
0
;
}
static
void
destroyTableScanOperatorInfo
(
void
*
param
)
{
STableScanInfo
*
pTableScanInfo
=
(
STableScanInfo
*
)
param
;
blockDataDestroy
(
pTableScanInfo
->
pResBlock
);
cleanupQueryTableDataCond
(
&
pTableScanInfo
->
base
.
cond
);
static
void
destroyTableScanBase
(
STableScanBase
*
pBase
)
{
cleanupQueryTableDataCond
(
&
pBase
->
cond
);
tsdbReaderClose
(
p
TableScanInfo
->
base
.
dataReader
);
p
TableScanInfo
->
base
.
dataReader
=
NULL
;
tsdbReaderClose
(
p
Base
->
dataReader
);
p
Base
->
dataReader
=
NULL
;
if
(
p
TableScanInfo
->
base
.
matchInfo
.
pList
!=
NULL
)
{
taosArrayDestroy
(
p
TableScanInfo
->
base
.
matchInfo
.
pList
);
if
(
p
Base
->
matchInfo
.
pList
!=
NULL
)
{
taosArrayDestroy
(
p
Base
->
matchInfo
.
pList
);
}
taosLRUCacheCleanup
(
pTableScanInfo
->
base
.
metaCache
.
pTableMetaEntryCache
);
cleanupExprSupp
(
&
pTableScanInfo
->
base
.
pseudoSup
);
tableListDestroy
(
pBase
->
pTableListInfo
);
taosLRUCacheCleanup
(
pBase
->
metaCache
.
pTableMetaEntryCache
);
cleanupExprSupp
(
&
pBase
->
pseudoSup
);
}
static
void
destroyTableScanOperatorInfo
(
void
*
param
)
{
STableScanInfo
*
pTableScanInfo
=
(
STableScanInfo
*
)
param
;
blockDataDestroy
(
pTableScanInfo
->
pResBlock
);
destroyTableScanBase
(
&
pTableScanInfo
->
base
);
taosMemoryFreeClear
(
param
);
}
SOperatorInfo
*
createTableScanOperatorInfo
(
STableScanPhysiNode
*
pTableScanNode
,
SReadHandle
*
readHandle
,
SExecTaskInfo
*
pTaskInfo
)
{
S
TableListInfo
*
pTableListInfo
,
S
ExecTaskInfo
*
pTaskInfo
)
{
int32_t
code
=
0
;
STableScanInfo
*
pInfo
=
taosMemoryCalloc
(
1
,
sizeof
(
STableScanInfo
));
SOperatorInfo
*
pOperator
=
taosMemoryCalloc
(
1
,
sizeof
(
SOperatorInfo
));
...
...
@@ -941,6 +946,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
pTaskInfo
);
pOperator
->
exprSupp
.
numOfExprs
=
numOfCols
;
pInfo
->
base
.
pTableListInfo
=
pTableListInfo
;
pInfo
->
base
.
metaCache
.
pTableMetaEntryCache
=
taosLRUCacheInit
(
1024
*
128
,
-
1
,
.
5
);
if
(
pInfo
->
base
.
metaCache
.
pTableMetaEntryCache
==
NULL
)
{
code
=
terrno
;
...
...
@@ -1060,7 +1066,7 @@ static SSDataBlock* readPreVersionData(SOperatorInfo* pTableScanOp, uint64_t tbU
if
(
hasNext
)
{
/*SSDataBlock* p = */
tsdbRetrieveDataBlock
(
pReader
,
NULL
);
doSetTagColumnData
(
&
pTableScanInfo
->
base
,
pBlock
,
pTaskInfo
,
pBlock
->
info
.
rows
);
pBlock
->
info
.
id
.
groupId
=
getTableGroupId
(
pTa
skInfo
->
pTableInfoList
,
pBlock
->
info
.
id
.
uid
);
pBlock
->
info
.
id
.
groupId
=
getTableGroupId
(
pTa
bleScanInfo
->
base
.
pTableListInfo
,
pBlock
->
info
.
id
.
uid
);
}
tsdbReaderClose
(
pReader
);
...
...
@@ -1081,7 +1087,8 @@ static uint64_t getGroupIdByCol(SStreamScanInfo* pInfo, uint64_t uid, TSKEY ts,
}
static
uint64_t
getGroupIdByUid
(
SStreamScanInfo
*
pInfo
,
uint64_t
uid
)
{
return
getTableGroupId
(
pInfo
->
pTableScanOp
->
pTaskInfo
->
pTableInfoList
,
uid
);
STableScanInfo
*
pTableScanInfo
=
pInfo
->
pTableScanOp
->
info
;
return
getTableGroupId
(
pTableScanInfo
->
base
.
pTableListInfo
,
uid
);
}
static
uint64_t
getGroupIdByData
(
SStreamScanInfo
*
pInfo
,
uint64_t
uid
,
TSKEY
ts
,
int64_t
maxVersion
)
{
...
...
@@ -1555,7 +1562,8 @@ static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock
pInfo
->
pRes
->
info
.
type
=
STREAM_NORMAL
;
pInfo
->
pRes
->
info
.
version
=
pBlock
->
info
.
version
;
pInfo
->
pRes
->
info
.
id
.
groupId
=
getTableGroupId
(
pTaskInfo
->
pTableInfoList
,
pBlock
->
info
.
id
.
uid
);
STableScanInfo
*
pTableScanInfo
=
pInfo
->
pTableScanOp
->
info
;
pInfo
->
pRes
->
info
.
id
.
groupId
=
getTableGroupId
(
pTableScanInfo
->
base
.
pTableListInfo
,
pBlock
->
info
.
id
.
uid
);
// todo extract method
for
(
int32_t
i
=
0
;
i
<
taosArrayGetSize
(
pInfo
->
matchInfo
.
pList
);
++
i
)
{
...
...
@@ -2137,19 +2145,19 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
code
=
tsdbNextDataBlock
(
pInfo
->
dataReader
,
&
hasNext
);
if
(
code
)
{
tsdbReleaseDataBlock
(
pInfo
->
dataReader
);
longjmp
(
pTaskInfo
->
env
,
code
);
T_LONG_JMP
(
pTaskInfo
->
env
,
code
);
}
}
if
(
pInfo
->
dataReader
&&
hasNext
)
{
if
(
isTaskKilled
(
pTaskInfo
))
{
tsdbReleaseDataBlock
(
pInfo
->
dataReader
);
longjmp
(
pTaskInfo
->
env
,
pTaskInfo
->
code
);
T_LONG_JMP
(
pTaskInfo
->
env
,
pTaskInfo
->
code
);
}
SSDataBlock
*
pBlock
=
tsdbRetrieveDataBlock
(
pInfo
->
dataReader
,
NULL
);
if
(
pBlock
==
NULL
)
{
longjmp
(
pTaskInfo
->
env
,
terrno
);
T_LONG_JMP
(
pTaskInfo
->
env
,
terrno
);
}
qDebug
(
"tmqsnap doRawScan get data uid:%"
PRId64
""
,
pBlock
->
info
.
id
.
uid
);
...
...
@@ -2239,6 +2247,7 @@ static void destroyRawScanOperatorInfo(void* param) {
SStreamRawScanInfo
*
pRawScan
=
(
SStreamRawScanInfo
*
)
param
;
tsdbReaderClose
(
pRawScan
->
dataReader
);
destroySnapContext
(
pRawScan
->
sContext
);
tableListDestroy
(
pRawScan
->
pTableListInfo
);
taosMemoryFree
(
pRawScan
);
}
...
...
@@ -2260,6 +2269,7 @@ SOperatorInfo* createRawScanOperatorInfo(SReadHandle* pHandle, SExecTaskInfo* pT
goto
_end
;
}
pInfo
->
pTableListInfo
=
tableListCreate
();
pInfo
->
vnode
=
pHandle
->
vnode
;
pInfo
->
sContext
=
pHandle
->
sContext
;
...
...
@@ -2278,9 +2288,11 @@ _end:
static
void
destroyStreamScanOperatorInfo
(
void
*
param
)
{
SStreamScanInfo
*
pStreamScan
=
(
SStreamScanInfo
*
)
param
;
if
(
pStreamScan
->
pTableScanOp
&&
pStreamScan
->
pTableScanOp
->
info
)
{
destroyOperatorInfo
(
pStreamScan
->
pTableScanOp
);
}
if
(
pStreamScan
->
tqReader
)
{
tqCloseReader
(
pStreamScan
->
tqReader
);
}
...
...
@@ -2307,13 +2319,14 @@ static void destroyStreamScanOperatorInfo(void* param) {
}
SOperatorInfo
*
createStreamScanOperatorInfo
(
SReadHandle
*
pHandle
,
STableScanPhysiNode
*
pTableScanNode
,
SNode
*
pTagCond
,
SExecTaskInfo
*
pTaskInfo
)
{
S
TableListInfo
*
pTableListInfo
,
S
ExecTaskInfo
*
pTaskInfo
)
{
SArray
*
pColIds
=
NULL
;
SStreamScanInfo
*
pInfo
=
taosMemoryCalloc
(
1
,
sizeof
(
SStreamScanInfo
));
SOperatorInfo
*
pOperator
=
taosMemoryCalloc
(
1
,
sizeof
(
SOperatorInfo
));
if
(
pInfo
==
NULL
||
pOperator
==
NULL
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
tableListDestroy
(
pTableListInfo
);
goto
_error
;
}
...
...
@@ -2327,6 +2340,7 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
int32_t
code
=
extractColMatchInfo
(
pScanPhyNode
->
pScanCols
,
pDescNode
,
&
numOfCols
,
COL_MATCH_FROM_COL_ID
,
&
pInfo
->
matchInfo
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
tableListDestroy
(
pTableListInfo
);
goto
_error
;
}
...
...
@@ -2346,11 +2360,14 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
SExprInfo
*
pSubTableExpr
=
taosMemoryCalloc
(
1
,
sizeof
(
SExprInfo
));
if
(
pSubTableExpr
==
NULL
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
tableListDestroy
(
pTableListInfo
);
goto
_error
;
}
pInfo
->
tbnameCalSup
.
pExprInfo
=
pSubTableExpr
;
createExprFromOneNode
(
pSubTableExpr
,
pTableScanNode
->
pSubtable
,
0
);
if
(
initExprSupp
(
&
pInfo
->
tbnameCalSup
,
pSubTableExpr
,
1
)
!=
0
)
{
tableListDestroy
(
pTableListInfo
);
goto
_error
;
}
}
...
...
@@ -2360,10 +2377,12 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
SExprInfo
*
pTagExpr
=
createExpr
(
pTableScanNode
->
pTags
,
&
numOfTags
);
if
(
pTagExpr
==
NULL
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
tableListDestroy
(
pTableListInfo
);
goto
_error
;
}
if
(
initExprSupp
(
&
pInfo
->
tagCalSup
,
pTagExpr
,
numOfTags
)
!=
0
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
tableListDestroy
(
pTableListInfo
);
goto
_error
;
}
}
...
...
@@ -2371,11 +2390,12 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
pInfo
->
pBlockLists
=
taosArrayInit
(
4
,
sizeof
(
SPackedData
));
if
(
pInfo
->
pBlockLists
==
NULL
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
tableListDestroy
(
pTableListInfo
);
goto
_error
;
}
if
(
pHandle
->
vnode
)
{
SOperatorInfo
*
pTableScanOp
=
createTableScanOperatorInfo
(
pTableScanNode
,
pHandle
,
pTaskInfo
);
SOperatorInfo
*
pTableScanOp
=
createTableScanOperatorInfo
(
pTableScanNode
,
pHandle
,
pTa
bleListInfo
,
pTa
skInfo
);
STableScanInfo
*
pTSInfo
=
(
STableScanInfo
*
)
pTableScanOp
->
info
;
if
(
pHandle
->
version
>
0
)
{
pTSInfo
->
base
.
cond
.
endVersion
=
pHandle
->
version
;
...
...
@@ -2383,7 +2403,7 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
STableKeyInfo
*
pList
=
NULL
;
int32_t
num
=
0
;
tableListGetGroupList
(
pTa
skInfo
->
pTableInfoList
,
0
,
&
pList
,
&
num
);
tableListGetGroupList
(
pTa
bleListInfo
,
0
,
&
pList
,
&
num
);
if
(
pHandle
->
initTableReader
)
{
pTSInfo
->
scanMode
=
TABLE_SCAN__TABLE_ORDER
;
...
...
@@ -2413,16 +2433,18 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
// set the extract column id to streamHandle
tqReaderSetColIdList
(
pInfo
->
tqReader
,
pColIds
);
SArray
*
tableIdList
=
extractTableIdList
(
pTaskInfo
->
pTableInfoList
);
SArray
*
tableIdList
=
extractTableIdList
(
((
STableScanInfo
*
)(
pInfo
->
pTableScanOp
->
info
))
->
base
.
pTableListInfo
);
code
=
tqReaderSetTbUidList
(
pInfo
->
tqReader
,
tableIdList
);
if
(
code
!=
0
)
{
taosArrayDestroy
(
tableIdList
);
goto
_error
;
}
taosArrayDestroy
(
tableIdList
);
memcpy
(
&
pTaskInfo
->
streamInfo
.
tableCond
,
&
pTSInfo
->
base
.
cond
,
sizeof
(
SQueryTableDataCond
));
}
else
{
taosArrayDestroy
(
pColIds
);
tableListDestroy
(
pTableListInfo
);
pColIds
=
NULL
;
}
...
...
@@ -2457,7 +2479,7 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
pTaskInfo
);
pOperator
->
exprSupp
.
numOfExprs
=
taosArrayGetSize
(
pInfo
->
pRes
->
pDataBlock
);
__optr_fn_t
nextFn
=
pTaskInfo
->
execModel
==
OPTR_EXEC_MODEL_STREAM
?
doStreamScan
:
doQueueScan
;
__optr_fn_t
nextFn
=
(
pTaskInfo
->
execModel
==
OPTR_EXEC_MODEL_STREAM
)
?
doStreamScan
:
doQueueScan
;
pOperator
->
fpSet
=
createOperatorFpSet
(
optrDummyOpenFn
,
nextFn
,
NULL
,
destroyStreamScanOperatorInfo
,
optrDefaultBufFn
,
NULL
);
...
...
@@ -2488,7 +2510,7 @@ static SSDataBlock* doTagScan(SOperatorInfo* pOperator) {
SSDataBlock
*
pRes
=
pInfo
->
pRes
;
blockDataCleanup
(
pRes
);
int32_t
size
=
tableListGetSize
(
p
TaskInfo
->
pTableInfoList
);
int32_t
size
=
tableListGetSize
(
p
Info
->
pTableListInfo
);
if
(
size
==
0
)
{
setTaskStatus
(
pTaskInfo
,
TASK_COMPLETED
);
return
NULL
;
...
...
@@ -2500,7 +2522,7 @@ static SSDataBlock* doTagScan(SOperatorInfo* pOperator) {
metaReaderInit
(
&
mr
,
pInfo
->
readHandle
.
meta
,
0
);
while
(
pInfo
->
curPos
<
size
&&
count
<
pOperator
->
resultInfo
.
capacity
)
{
STableKeyInfo
*
item
=
tableListGetInfo
(
p
TaskInfo
->
pTableInfoList
,
pInfo
->
curPos
);
STableKeyInfo
*
item
=
tableListGetInfo
(
p
Info
->
pTableListInfo
,
pInfo
->
curPos
);
int32_t
code
=
metaGetTableEntryByUid
(
&
mr
,
item
->
uid
);
tDecoderClear
(
&
mr
.
coder
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
...
...
@@ -2561,10 +2583,11 @@ static void destroyTagScanOperatorInfo(void* param) {
STagScanInfo
*
pInfo
=
(
STagScanInfo
*
)
param
;
pInfo
->
pRes
=
blockDataDestroy
(
pInfo
->
pRes
);
taosArrayDestroy
(
pInfo
->
matchInfo
.
pList
);
pInfo
->
pTableListInfo
=
tableListDestroy
(
pInfo
->
pTableListInfo
);
taosMemoryFreeClear
(
param
);
}
SOperatorInfo
*
createTagScanOperatorInfo
(
SReadHandle
*
pReadHandle
,
STagScanPhysiNode
*
pPhyNode
,
SOperatorInfo
*
createTagScanOperatorInfo
(
SReadHandle
*
pReadHandle
,
STagScanPhysiNode
*
pPhyNode
,
STableListInfo
*
pTableListInfo
,
SExecTaskInfo
*
pTaskInfo
)
{
STagScanInfo
*
pInfo
=
taosMemoryCalloc
(
1
,
sizeof
(
STagScanInfo
));
SOperatorInfo
*
pOperator
=
taosMemoryCalloc
(
1
,
sizeof
(
SOperatorInfo
));
...
...
@@ -2587,6 +2610,7 @@ SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysi
goto
_error
;
}
pInfo
->
pTableListInfo
=
pTableListInfo
;
pInfo
->
pRes
=
createDataBlockFromDescNode
(
pDescNode
);
pInfo
->
readHandle
=
*
pReadHandle
;
pInfo
->
curPos
=
0
;
...
...
@@ -2620,7 +2644,7 @@ static SSDataBlock* getTableDataBlockImpl(void* param) {
SQueryTableDataCond
*
pQueryCond
=
taosArrayGet
(
pInfo
->
queryConds
,
readIdx
);
int64_t
st
=
taosGetTimestampUs
();
void
*
p
=
tableListGetInfo
(
p
TaskInfo
->
pTableInfoList
,
readIdx
+
pInfo
->
tableStartIndex
);
void
*
p
=
tableListGetInfo
(
p
Info
->
base
.
pTableListInfo
,
readIdx
+
pInfo
->
tableStartIndex
);
SReadHandle
*
pHandle
=
&
pInfo
->
base
.
readHandle
;
if
(
NULL
==
source
->
dataReader
||
!
source
->
multiReader
)
{
...
...
@@ -2677,7 +2701,7 @@ static SSDataBlock* getTableDataBlockImpl(void* param) {
continue
;
}
pBlock
->
info
.
id
.
groupId
=
getTableGroupId
(
p
TaskInfo
->
pTableInfoList
,
pBlock
->
info
.
id
.
uid
);
pBlock
->
info
.
id
.
groupId
=
getTableGroupId
(
p
Info
->
base
.
pTableListInfo
,
pBlock
->
info
.
id
.
uid
);
pOperator
->
resultInfo
.
totalRows
+=
pBlock
->
info
.
rows
;
pInfo
->
base
.
readRecorder
.
elapsedTime
+=
(
taosGetTimestampUs
()
-
st
)
/
1000
.
0
;
...
...
@@ -2733,10 +2757,10 @@ int32_t startGroupTableMergeScan(SOperatorInfo* pOperator) {
SExecTaskInfo
*
pTaskInfo
=
pOperator
->
pTaskInfo
;
{
size_t
numOfTables
=
tableListGetSize
(
p
TaskInfo
->
pTableInfoList
);
size_t
numOfTables
=
tableListGetSize
(
p
Info
->
base
.
pTableListInfo
);
int32_t
i
=
pInfo
->
tableStartIndex
+
1
;
for
(;
i
<
numOfTables
;
++
i
)
{
STableKeyInfo
*
tableKeyInfo
=
tableListGetInfo
(
p
TaskInfo
->
pTableInfoList
,
i
);
STableKeyInfo
*
tableKeyInfo
=
tableListGetInfo
(
p
Info
->
base
.
pTableListInfo
,
i
);
if
(
tableKeyInfo
->
groupId
!=
pInfo
->
groupId
)
{
break
;
}
...
...
@@ -2870,7 +2894,7 @@ SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) {
T_LONG_JMP
(
pTaskInfo
->
env
,
code
);
}
size_t
tableListSize
=
tableListGetSize
(
p
TaskInfo
->
pTableInfoList
);
size_t
tableListSize
=
tableListGetSize
(
p
Info
->
base
.
pTableListInfo
);
if
(
!
pInfo
->
hasGroupId
)
{
pInfo
->
hasGroupId
=
true
;
...
...
@@ -2879,7 +2903,7 @@ SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) {
return
NULL
;
}
pInfo
->
tableStartIndex
=
0
;
pInfo
->
groupId
=
((
STableKeyInfo
*
)
tableListGetInfo
(
p
TaskInfo
->
pTableInfoList
,
pInfo
->
tableStartIndex
))
->
groupId
;
pInfo
->
groupId
=
((
STableKeyInfo
*
)
tableListGetInfo
(
p
Info
->
base
.
pTableListInfo
,
pInfo
->
tableStartIndex
))
->
groupId
;
startGroupTableMergeScan
(
pOperator
);
}
...
...
@@ -2904,7 +2928,7 @@ SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) {
}
pInfo
->
tableStartIndex
=
pInfo
->
tableEndIndex
+
1
;
pInfo
->
groupId
=
tableListGetInfo
(
p
TaskInfo
->
pTableInfoList
,
pInfo
->
tableStartIndex
)
->
groupId
;
pInfo
->
groupId
=
tableListGetInfo
(
p
Info
->
base
.
pTableListInfo
,
pInfo
->
tableStartIndex
)
->
groupId
;
startGroupTableMergeScan
(
pOperator
);
resetLimitInfoForNextGroup
(
&
pInfo
->
limitInfo
);
}
...
...
@@ -2926,9 +2950,6 @@ void destroyTableMergeScanOperatorInfo(void* param) {
p
->
dataReader
=
NULL
;
}
tsdbReaderClose
(
pTableScanInfo
->
base
.
dataReader
);
pTableScanInfo
->
base
.
dataReader
=
NULL
;
taosArrayDestroy
(
pTableScanInfo
->
sortSourceParams
);
tsortDestroySortHandle
(
pTableScanInfo
->
pSortHandle
);
pTableScanInfo
->
pSortHandle
=
NULL
;
...
...
@@ -2937,20 +2958,14 @@ void destroyTableMergeScanOperatorInfo(void* param) {
SQueryTableDataCond
*
pCond
=
taosArrayGet
(
pTableScanInfo
->
queryConds
,
i
);
taosMemoryFree
(
pCond
->
colList
);
}
taosArrayDestroy
(
pTableScanInfo
->
queryConds
);
if
(
pTableScanInfo
->
base
.
matchInfo
.
pList
!=
NULL
)
{
taosArrayDestroy
(
pTableScanInfo
->
base
.
matchInfo
.
pList
);
}
taosArrayDestroy
(
pTableScanInfo
->
queryConds
);
destroyTableScanBase
(
&
pTableScanInfo
->
base
);
pTableScanInfo
->
pResBlock
=
blockDataDestroy
(
pTableScanInfo
->
pResBlock
);
pTableScanInfo
->
pSortInputBlock
=
blockDataDestroy
(
pTableScanInfo
->
pSortInputBlock
);
taosArrayDestroy
(
pTableScanInfo
->
pSortInfo
);
cleanupExprSupp
(
&
pTableScanInfo
->
base
.
pseudoSup
);
taosLRUCacheCleanup
(
pTableScanInfo
->
base
.
metaCache
.
pTableMetaEntryCache
);
taosMemoryFreeClear
(
param
);
}
...
...
@@ -2969,7 +2984,7 @@ int32_t getTableMergeScanExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExpla
}
SOperatorInfo
*
createTableMergeScanOperatorInfo
(
STableScanPhysiNode
*
pTableScanNode
,
SReadHandle
*
readHandle
,
SExecTaskInfo
*
pTaskInfo
)
{
S
TableListInfo
*
pTableListInfo
,
S
ExecTaskInfo
*
pTaskInfo
)
{
STableMergeScanInfo
*
pInfo
=
taosMemoryCalloc
(
1
,
sizeof
(
STableMergeScanInfo
));
SOperatorInfo
*
pOperator
=
taosMemoryCalloc
(
1
,
sizeof
(
SOperatorInfo
));
if
(
pInfo
==
NULL
||
pOperator
==
NULL
)
{
...
...
@@ -3011,6 +3026,7 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN
pInfo
->
base
.
limitInfo
.
limit
.
limit
=
-
1
;
pInfo
->
base
.
limitInfo
.
slimit
.
limit
=
-
1
;
pInfo
->
base
.
pTableListInfo
=
pTableListInfo
;
pInfo
->
sample
.
sampleRatio
=
pTableScanNode
->
ratio
;
pInfo
->
sample
.
seed
=
taosGetTimestampSec
();
...
...
source/libs/executor/src/sysscanoperator.c
浏览文件 @
1d7581f6
...
...
@@ -83,10 +83,11 @@ typedef struct MergeIndex {
}
MergeIndex
;
typedef
struct
SBlockDistInfo
{
SSDataBlock
*
pResBlock
;
STsdbReader
*
pHandle
;
SReadHandle
readHandle
;
uint64_t
uid
;
// table uid
SSDataBlock
*
pResBlock
;
STsdbReader
*
pHandle
;
SReadHandle
readHandle
;
STableListInfo
*
pTableListInfo
;
uint64_t
uid
;
// table uid
}
SBlockDistInfo
;
static
int32_t
sysChkFilter__Comm
(
SNode
*
pNode
);
...
...
@@ -2214,6 +2215,7 @@ static void destroyBlockDistScanOperatorInfo(void* param) {
SBlockDistInfo
*
pDistInfo
=
(
SBlockDistInfo
*
)
param
;
blockDataDestroy
(
pDistInfo
->
pResBlock
);
tsdbReaderClose
(
pDistInfo
->
pHandle
);
tableListDestroy
(
pDistInfo
->
pTableListInfo
);
taosMemoryFreeClear
(
param
);
}
...
...
@@ -2245,7 +2247,7 @@ static int32_t initTableblockDistQueryCond(uint64_t uid, SQueryTableDataCond* pC
}
SOperatorInfo
*
createDataBlockInfoScanOperator
(
SReadHandle
*
readHandle
,
SBlockDistScanPhysiNode
*
pBlockScanNode
,
SExecTaskInfo
*
pTaskInfo
)
{
S
TableListInfo
*
pTableListInfo
,
S
ExecTaskInfo
*
pTaskInfo
)
{
SBlockDistInfo
*
pInfo
=
taosMemoryCalloc
(
1
,
sizeof
(
SBlockDistInfo
));
SOperatorInfo
*
pOperator
=
taosMemoryCalloc
(
1
,
sizeof
(
SOperatorInfo
));
if
(
pInfo
==
NULL
||
pOperator
==
NULL
)
{
...
...
@@ -2263,9 +2265,9 @@ SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDi
goto
_error
;
}
STableListInfo
*
pTableListInfo
=
pTaskInfo
->
pTableInfoList
;
size_t
num
=
tableListGetSize
(
pTableListInfo
);
void
*
pList
=
tableListGetInfo
(
pTableListInfo
,
0
);
pInfo
->
pTableListInfo
=
pTableListInfo
;
size_t
num
=
tableListGetSize
(
pTableListInfo
);
void
*
pList
=
tableListGetInfo
(
pTableListInfo
,
0
);
code
=
tsdbReaderOpen
(
readHandle
->
vnode
,
&
cond
,
pList
,
num
,
pInfo
->
pResBlock
,
&
pInfo
->
pHandle
,
pTaskInfo
->
id
.
str
,
false
);
cleanupQueryTableDataCond
(
&
cond
);
...
...
source/libs/executor/src/timewindowoperator.c
浏览文件 @
1d7581f6
...
...
@@ -938,6 +938,7 @@ static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResul
if
(
ret
!=
TSDB_CODE_SUCCESS
||
pResult
==
NULL
)
{
T_LONG_JMP
(
pTaskInfo
->
env
,
TSDB_CODE_OUT_OF_MEMORY
);
}
TSKEY
ekey
=
ascScan
?
win
.
ekey
:
win
.
skey
;
int32_t
forwardRows
=
getNumOfRowsInTimeWindow
(
&
pBlock
->
info
,
tsCols
,
startPos
,
ekey
,
binarySearchForKey
,
NULL
,
pInfo
->
inputOrder
);
...
...
@@ -2175,13 +2176,6 @@ static void rebuildIntervalWindow(SOperatorInfo* pOperator, SArray* pWinArray, S
}
}
bool
isDeletedWindow
(
STimeWindow
*
pWin
,
uint64_t
groupId
,
SAggSupporter
*
pSup
)
{
SET_RES_WINDOW_KEY
(
pSup
->
keyBuf
,
&
pWin
->
skey
,
sizeof
(
int64_t
),
groupId
);
SResultRowPosition
*
p1
=
(
SResultRowPosition
*
)
tSimpleHashGet
(
pSup
->
pResultRowHashTable
,
pSup
->
keyBuf
,
GET_RES_WINDOW_KEY_LEN
(
sizeof
(
int64_t
)));
return
p1
==
NULL
;
}
bool
isDeletedStreamWindow
(
STimeWindow
*
pWin
,
uint64_t
groupId
,
SStreamState
*
pState
,
STimeWindowAggSupp
*
pTwSup
)
{
if
(
pWin
->
ekey
<
pTwSup
->
maxTs
-
pTwSup
->
deleteMark
)
{
SWinKey
key
=
{.
ts
=
pWin
->
skey
,
.
groupId
=
groupId
};
...
...
source/libs/parser/src/parTranslater.c
浏览文件 @
1d7581f6
...
...
@@ -757,7 +757,8 @@ static bool isPrimaryKeyImpl(SNode* pExpr) {
if
(
FUNCTION_TYPE_SELECT_VALUE
==
pFunc
->
funcType
||
FUNCTION_TYPE_GROUP_KEY
==
pFunc
->
funcType
||
FUNCTION_TYPE_FIRST
==
pFunc
->
funcType
||
FUNCTION_TYPE_LAST
==
pFunc
->
funcType
)
{
return
isPrimaryKeyImpl
(
nodesListGetNode
(
pFunc
->
pParameterList
,
0
));
}
else
if
(
FUNCTION_TYPE_WSTART
==
pFunc
->
funcType
||
FUNCTION_TYPE_WEND
==
pFunc
->
funcType
||
FUNCTION_TYPE_IROWTS
==
pFunc
->
funcType
)
{
}
else
if
(
FUNCTION_TYPE_WSTART
==
pFunc
->
funcType
||
FUNCTION_TYPE_WEND
==
pFunc
->
funcType
||
FUNCTION_TYPE_IROWTS
==
pFunc
->
funcType
)
{
return
true
;
}
}
...
...
@@ -3119,6 +3120,19 @@ static const char* getPrecisionStr(uint8_t precision) {
return
"unknown"
;
}
static
void
convertVarDuration
(
SValueNode
*
pOffset
,
uint8_t
precision
)
{
const
int64_t
factors
[
3
]
=
{
NANOSECOND_PER_MSEC
,
NANOSECOND_PER_USEC
,
1
};
const
int8_t
units
[
3
]
=
{
TIME_UNIT_MILLISECOND
,
TIME_UNIT_MICROSECOND
,
TIME_UNIT_NANOSECOND
};
if
(
pOffset
->
unit
==
'n'
)
{
pOffset
->
datum
.
i
=
pOffset
->
datum
.
i
*
31
*
(
NANOSECOND_PER_DAY
/
factors
[
precision
]);
}
else
{
pOffset
->
datum
.
i
=
pOffset
->
datum
.
i
*
365
*
(
NANOSECOND_PER_DAY
/
factors
[
precision
]);
}
pOffset
->
unit
=
units
[
precision
];
}
static
int32_t
checkIntervalWindow
(
STranslateContext
*
pCxt
,
SIntervalWindowNode
*
pInterval
)
{
uint8_t
precision
=
((
SColumnNode
*
)
pInterval
->
pCol
)
->
node
.
resType
.
precision
;
...
...
@@ -3143,6 +3157,10 @@ static int32_t checkIntervalWindow(STranslateContext* pCxt, SIntervalWindowNode*
getMonthsFromTimeVal
(
pInter
->
datum
.
i
,
precision
,
pInter
->
unit
)))
{
return
generateSyntaxErrMsg
(
&
pCxt
->
msgBuf
,
TSDB_CODE_PAR_INTER_OFFSET_TOO_BIG
);
}
if
(
pOffset
->
unit
==
'n'
||
pOffset
->
unit
==
'y'
)
{
convertVarDuration
(
pOffset
,
precision
);
}
}
if
(
NULL
!=
pInterval
->
pSliding
)
{
...
...
tests/script/tsim/query/interval-offset.sim
浏览文件 @
1d7581f6
...
...
@@ -212,10 +212,10 @@ print ===> rows2: $data20 $data21 $data22 $data23 $data24
if $rows != 3 then
return -1
endi
if $data01 !=
2
then
if $data01 !=
4
then
return -1
endi
if $data04 !=
2
then
if $data04 !=
4
then
return -1
endi
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录