Skip to content

T
TDengine

taosdata / TDengine
接近 2 年 前同步成功

通知 1191
Star 22018
Fork 4786
T
TDengine
提交 6d2ce5cd 编写于 作者: H Hongze Cheng
浏览文件
操作

Merge branch '3.0' of https://github.com/taosdata/TDengine into enh/tsdb_optimize

上级 3074ab65 1ca24fea
展开全部 隐藏空白更改
内联 并排
Showing with 3130 addition and 212 deletion
docs/en/28-releases/01-tdengine.md
浏览文件 @ 6d2ce5cd
......@@ -10,6 +10,10 @@ For TDengine 2.x installation packages by version, please visit [here](https://w
import Release from "/components/ReleaseV3";
## 3.0.5.1
<Release type="tdengine" version="3.0.5.1" />
## 3.0.5.0
<Release type="tdengine" version="3.0.5.0" />
......
docs/en/28-releases/02-tools.md
浏览文件 @ 6d2ce5cd
......@@ -10,6 +10,10 @@ For other historical version installers, please visit [here](https://www.taosdat
import Release from "/components/ReleaseV3";
## 2.5.2
<Release type="tools" version="2.5.2" />
## 2.5.1
<Release type="tools" version="2.5.1" />
......
docs/zh/28-releases/01-tdengine.md
浏览文件 @ 6d2ce5cd
......@@ -10,6 +10,10 @@ TDengine 2.x 各版本安装包请访问[这里](https://www.taosdata.com/all-do
import Release from "/components/ReleaseV3";
## 3.0.5.1
<Release type="tdengine" version="3.0.5.1" />
## 3.0.5.0
<Release type="tdengine" version="3.0.5.0" />
......
docs/zh/28-releases/02-tools.md
浏览文件 @ 6d2ce5cd
......@@ -10,6 +10,10 @@ taosTools 各版本安装包下载链接如下：
import Release from "/components/ReleaseV3";
## 2.5.2
<Release type="tools" version="2.5.2" />
## 2.5.1
<Release type="tools" version="2.5.1" />
......
include/common/tcommon.h
浏览文件 @ 6d2ce5cd
......@@ -198,6 +198,7 @@ typedef struct SDataBlockInfo {
SBlockID id;
int16_t hasVarCol;
int16_t dataLoad; // denote if the data is loaded or not
uint8_t scanFlag;
// TODO: optimize and remove following
int64_t version; // used for stream, and need serialization
......
include/libs/nodes/plannodes.h
浏览文件 @ 6d2ce5cd
......@@ -53,6 +53,8 @@ typedef struct SLogicNode {
EDataOrderLevel requireDataOrder; // requirements for input data
EDataOrderLevel resultDataOrder; // properties of the output data
EGroupAction groupAction;
EOrder inputTsOrder;
EOrder outputTsOrder;
} SLogicNode;
typedef enum EScanType {
......@@ -111,7 +113,6 @@ typedef struct SJoinLogicNode {
SNode* pMergeCondition;
SNode* pOnConditions;
bool isSingleTableJoin;
EOrder inputTsOrder;
SNode* pColEqualOnConditions;
} SJoinLogicNode;
......@@ -229,8 +230,6 @@ typedef struct SWindowLogicNode {
int8_t igExpired;
int8_t igCheckUpdate;
EWindowAlgorithm windowAlgo;
EOrder inputTsOrder;
EOrder outputTsOrder;
} SWindowLogicNode;
typedef struct SFillLogicNode {
......@@ -241,7 +240,6 @@ typedef struct SFillLogicNode {
SNode* pWStartTs;
SNode* pValues; // SNodeListNode
STimeWindow timeRange;
EOrder inputTsOrder;
} SFillLogicNode;
typedef struct SSortLogicNode {
......@@ -310,6 +308,8 @@ typedef struct SDataBlockDescNode {
typedef struct SPhysiNode {
ENodeType type;
EOrder inputTsOrder;
EOrder outputTsOrder;
SDataBlockDescNode* pOutputDataBlockDesc;
SNode* pConditions;
SNodeList* pChildren;
......@@ -406,7 +406,6 @@ typedef struct SSortMergeJoinPhysiNode {
SNode* pMergeCondition;
SNode* pOnConditions;
SNodeList* pTargets;
EOrder inputTsOrder;
SNode* pColEqualOnConditions;
} SSortMergeJoinPhysiNode;
......@@ -460,8 +459,6 @@ typedef struct SWindowPhysiNode {
int64_t watermark;
int64_t deleteMark;
int8_t igExpired;
EOrder inputTsOrder;
EOrder outputTsOrder;
bool mergeDataBlock;
} SWindowPhysiNode;
......@@ -488,7 +485,6 @@ typedef struct SFillPhysiNode {
SNode* pWStartTs; // SColumnNode
SNode* pValues; // SNodeListNode
STimeWindow timeRange;
EOrder inputTsOrder;
} SFillPhysiNode;
typedef SFillPhysiNode SStreamFillPhysiNode;
......
include/libs/nodes/querynodes.h
浏览文件 @ 6d2ce5cd
......@@ -69,6 +69,7 @@ typedef struct SColumnNode {
uint64_t tableId;
int8_t tableType;
col_id_t colId;
uint16_t projIdx; // the idx in project list, start from 1
EColumnType colType; // column or tag
bool hasIndex;
char dbName[TSDB_DB_NAME_LEN];
......
include/libs/qcom/query.h
浏览文件 @ 6d2ce5cd
......@@ -281,7 +281,8 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t
(_code) == TSDB_CODE_PAR_INVALID_DROP_COL || ((_code) == TSDB_CODE_TDB_INVALID_TABLE_ID))
#define NEED_CLIENT_REFRESH_VG_ERROR(_code) \
((_code) == TSDB_CODE_VND_HASH_MISMATCH || (_code) == TSDB_CODE_VND_INVALID_VGROUP_ID)
#define NEED_CLIENT_REFRESH_TBLMETA_ERROR(_code) ((_code) == TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER)
#define NEED_CLIENT_REFRESH_TBLMETA_ERROR(_code) \
((_code) == TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER || (_code) == TSDB_CODE_MND_INVALID_SCHEMA_VER)
#define NEED_CLIENT_HANDLE_ERROR(_code) \
(NEED_CLIENT_RM_TBLMETA_ERROR(_code) || NEED_CLIENT_REFRESH_VG_ERROR(_code) || \
NEED_CLIENT_REFRESH_TBLMETA_ERROR(_code))
......
source/client/src/clientMain.c
浏览文件 @ 6d2ce5cd
......@@ -1120,6 +1120,7 @@ void doAsyncQuery(SRequestObj *pRequest, bool updateMetaForce) {
if (NEED_CLIENT_HANDLE_ERROR(code)) {
tscDebug("0x%" PRIx64 " client retry to handle the error, code:%d - %s, tryCount:%d, reqId:0x%" PRIx64,
pRequest->self, code, tstrerror(code), pRequest->retry, pRequest->requestId);
refreshMeta(pRequest->pTscObj, pRequest);
pRequest->prevCode = code;
doAsyncQuery(pRequest, true);
return;
......
source/common/src/tdatablock.c
浏览文件 @ 6d2ce5cd
......@@ -791,8 +791,8 @@ size_t blockDataGetRowSize(SSDataBlock* pBlock) {
* @return
*/
size_t blockDataGetSerialMetaSize(uint32_t numOfCols) {
// | version | total length | total rows | total columns | flag seg| block group id | column schema | each column
// length |
// | version | total length | total rows | total columns | flag seg| block group id | column schema
// | each column length |
return sizeof(int32_t) + sizeof(int32_t) + sizeof(int32_t) + sizeof(int32_t) + sizeof(int32_t) + sizeof(uint64_t) +
numOfCols * (sizeof(int8_t) + sizeof(int32_t)) + numOfCols * sizeof(int32_t);
}
......
source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
浏览文件 @ 6d2ce5cd
......@@ -218,6 +218,7 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp
if (pMsg->msgType != TDMT_VND_ALTER_CONFIRM && pVnode->disable) {
dDebug("vgId:%d, msg:%p put into vnode-write queue failed since its disable", pVnode->vgId, pMsg);
terrno = TSDB_CODE_VND_STOPPED;
code = terrno;
break;
}
dGTrace("vgId:%d, msg:%p put into vnode-write queue", pVnode->vgId, pMsg);
......
source/dnode/vnode/src/sma/smaTimeRange.c
浏览文件 @ 6d2ce5cd
......@@ -29,7 +29,7 @@ int32_t tdProcessTSmaInsert(SSma *pSma, int64_t indexUid, const char *msg) {
int32_t code = TSDB_CODE_SUCCESS;
if ((code = tdProcessTSmaInsertImpl(pSma, indexUid, msg)) < 0) {
smaWarn("vgId:%d, insert tsma data failed since %s", SMA_VID(pSma), tstrerror(terrno));
smaError("vgId:%d, insert tsma data failed since %s", SMA_VID(pSma), tstrerror(terrno));
}
return code;
......@@ -346,6 +346,43 @@ _end:
return TSDB_CODE_SUCCESS;
}
static int32_t tsmaProcessDelReq(SSma *pSma, int64_t indexUid, SBatchDeleteReq *pDelReq) {
int32_t code = 0;
int32_t lino = 0;
if (taosArrayGetSize(pDelReq->deleteReqs) > 0) {
int32_t len = 0;
tEncodeSize(tEncodeSBatchDeleteReq, pDelReq, len, code);
TSDB_CHECK_CODE(code, lino, _exit);
void *pBuf = rpcMallocCont(len + sizeof(SMsgHead));
if (!pBuf) {
code = terrno;
TSDB_CHECK_CODE(code, lino, _exit);
}
SEncoder encoder;
tEncoderInit(&encoder, POINTER_SHIFT(pBuf, sizeof(SMsgHead)), len);
tEncodeSBatchDeleteReq(&encoder, pDelReq);
tEncoderClear(&encoder);
((SMsgHead *)pBuf)->vgId = TD_VID(pSma->pVnode);
SRpcMsg delMsg = {.msgType = TDMT_VND_BATCH_DEL, .pCont = pBuf, .contLen = len + sizeof(SMsgHead)};
code = tmsgPutToQueue(&pSma->pVnode->msgCb, WRITE_QUEUE, &delMsg);
TSDB_CHECK_CODE(code, lino, _exit);
}
_exit:
taosArrayDestroy(pDelReq->deleteReqs);
if (code) {
smaError("vgId:%d, failed at line %d to process delete req for smaIndex %" PRIi64 " since %s", SMA_VID(pSma), lino,
indexUid, tstrerror(code));
}
return code;
}
/**
* @brief Insert/Update Time-range-wise SMA data.
*
......@@ -355,7 +392,6 @@ _end:
*/
static int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char *msg) {
const SArray *pDataBlocks = (const SArray *)msg;
// TODO: destroy SSDataBlocks(msg)
if (!pDataBlocks) {
terrno = TSDB_CODE_TSMA_INVALID_PTR;
smaWarn("vgId:%d, insert tsma data failed since pDataBlocks is NULL", SMA_VID(pSma));
......@@ -419,8 +455,10 @@ static int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char
goto _err;
}
// TODO deleteReq
taosArrayDestroy(deleteReq.deleteReqs);
if ((terrno = tsmaProcessDelReq(pSma, indexUid, &deleteReq)) != 0) {
goto _err;
}
#if 0
if (!strncasecmp("td.tsma.rst.tb", pTsmaStat->pTSma->dstTbName, 14)) {
terrno = TSDB_CODE_APP_ERROR;
......
source/dnode/vnode/src/vnd/vnodeSvr.c
浏览文件 @ 6d2ce5cd
......@@ -593,9 +593,7 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
}
}
// TODO: remove the function
void smaHandleRes(void *pVnode, int64_t smaId, const SArray *data) {
// TODO
// blockDebugShowDataBlocks(data, __func__);
tdProcessTSmaInsert(((SVnode *)pVnode)->pSma, smaId, (const char *)data);
}
......
source/libs/command/inc/commandInt.h
浏览文件 @ 6d2ce5cd
......@@ -145,7 +145,7 @@ typedef struct SExplainCtx {
SHashObj *groupHash; // Hash<SExplainGroup>
} SExplainCtx;
#define EXPLAIN_ORDER_STRING(_order) ((ORDER_ASC == _order) ? "asc" : "desc")
#define EXPLAIN_ORDER_STRING(_order) ((ORDER_ASC == _order) ? "asc" : ORDER_DESC == _order ? "desc" : "unknown")
#define EXPLAIN_JOIN_STRING(_type) ((JOIN_TYPE_INNER == _type) ? "Inner join" : "Join")
#define INVERAL_TIME_FROM_PRECISION_TO_UNIT(_t, _u, _p) (((_u) == 'n' || (_u) == 'y') ? (_t) : (convertTimeFromPrecisionToUnit(_t, _p, _u)))
......
source/libs/command/src/explain.c
浏览文件 @ 6d2ce5cd
......@@ -499,6 +499,9 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
EXPLAIN_ROW_APPEND(EXPLAIN_COLUMNS_FORMAT, pPrjNode->pProjections->length);
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_WIDTH_FORMAT, pPrjNode->node.pOutputDataBlockDesc->totalRowSize);
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_INPUT_ORDER_FORMAT, EXPLAIN_ORDER_STRING(pPrjNode->node.inputTsOrder));
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_RIGHT_PARENTHESIS_FORMAT);
EXPLAIN_ROW_END();
QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level));
......@@ -544,6 +547,9 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
EXPLAIN_ROW_APPEND(EXPLAIN_COLUMNS_FORMAT, pJoinNode->pTargets->length);
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_WIDTH_FORMAT, pJoinNode->node.pOutputDataBlockDesc->totalRowSize);
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_INPUT_ORDER_FORMAT, EXPLAIN_ORDER_STRING(pJoinNode->node.inputTsOrder));
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_RIGHT_PARENTHESIS_FORMAT);
EXPLAIN_ROW_END();
QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level));
......@@ -597,6 +603,9 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_GROUPS_FORMAT, pAggNode->pGroupKeys->length);
}
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_INPUT_ORDER_FORMAT, EXPLAIN_ORDER_STRING(pAggNode->node.inputTsOrder));
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_RIGHT_PARENTHESIS_FORMAT);
EXPLAIN_ROW_END();
QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level));
......@@ -716,6 +725,11 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
case QUERY_NODE_PHYSICAL_PLAN_SORT: {
SSortPhysiNode *pSortNode = (SSortPhysiNode *)pNode;
EXPLAIN_ROW_NEW(level, EXPLAIN_SORT_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_INPUT_ORDER_FORMAT, EXPLAIN_ORDER_STRING(pSortNode->node.inputTsOrder));
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_OUTPUT_ORDER_TYPE_FORMAT, EXPLAIN_ORDER_STRING(pSortNode->node.outputTsOrder));
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_LEFT_PARENTHESIS_FORMAT);
if (pResNode->pExecInfo) {
QRY_ERR_RET(qExplainBufAppendExecInfo(pResNode->pExecInfo, tbuf, &tlen));
......@@ -796,9 +810,10 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_WIDTH_FORMAT, pIntNode->window.node.pOutputDataBlockDesc->totalRowSize);
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_INPUT_ORDER_FORMAT, EXPLAIN_ORDER_STRING(pIntNode->window.inputTsOrder));
EXPLAIN_ROW_APPEND(EXPLAIN_INPUT_ORDER_FORMAT, EXPLAIN_ORDER_STRING(pIntNode->window.node.inputTsOrder));
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_OUTPUT_ORDER_TYPE_FORMAT, EXPLAIN_ORDER_STRING(pIntNode->window.node.outputTsOrder));
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_OUTPUT_ORDER_TYPE_FORMAT, EXPLAIN_ORDER_STRING(pIntNode->window.outputTsOrder));
EXPLAIN_ROW_APPEND(EXPLAIN_RIGHT_PARENTHESIS_FORMAT);
EXPLAIN_ROW_END();
QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level));
......@@ -847,6 +862,10 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
EXPLAIN_ROW_APPEND(EXPLAIN_FUNCTIONS_FORMAT, pIntNode->window.pFuncs->length);
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_WIDTH_FORMAT, pIntNode->window.node.pOutputDataBlockDesc->totalRowSize);
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_INPUT_ORDER_FORMAT, EXPLAIN_ORDER_STRING(pIntNode->window.node.inputTsOrder));
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_OUTPUT_ORDER_TYPE_FORMAT, EXPLAIN_ORDER_STRING(pIntNode->window.node.outputTsOrder));
EXPLAIN_ROW_APPEND(EXPLAIN_RIGHT_PARENTHESIS_FORMAT);
EXPLAIN_ROW_END();
QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level));
......@@ -895,6 +914,9 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
EXPLAIN_ROW_APPEND(EXPLAIN_MODE_FORMAT, nodesGetFillModeString(pFillNode->mode));
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_WIDTH_FORMAT, pFillNode->node.pOutputDataBlockDesc->totalRowSize);
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_INPUT_ORDER_FORMAT, EXPLAIN_ORDER_STRING(pFillNode->node.inputTsOrder));
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_RIGHT_PARENTHESIS_FORMAT);
EXPLAIN_ROW_END();
QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level));
......@@ -1080,6 +1102,10 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
EXPLAIN_ROW_APPEND(EXPLAIN_COLUMNS_FORMAT, nodesGetOutputNumFromSlotList(pDescNode->pSlots));
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_WIDTH_FORMAT, pDescNode->totalRowSize);
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_INPUT_ORDER_FORMAT, EXPLAIN_ORDER_STRING(pMergeNode->node.inputTsOrder));
EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
EXPLAIN_ROW_APPEND(EXPLAIN_OUTPUT_ORDER_TYPE_FORMAT, EXPLAIN_ORDER_STRING(pMergeNode->node.outputTsOrder));
EXPLAIN_ROW_APPEND(EXPLAIN_RIGHT_PARENTHESIS_FORMAT);
EXPLAIN_ROW_END();
QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level));
......
source/libs/executor/inc/executorInt.h
浏览文件 @ 6d2ce5cd
......@@ -399,6 +399,8 @@ typedef struct SOptrBasicInfo {
SResultRowInfo resultRowInfo;
SSDataBlock* pRes;
bool mergeResultBlock;
int32_t inputTsOrder;
int32_t outputTsOrder;
} SOptrBasicInfo;
typedef struct SIntervalAggOperatorInfo {
......@@ -411,8 +413,6 @@ typedef struct SIntervalAggOperatorInfo {
STimeWindow win; // query time range
bool timeWindowInterpo; // interpolation needed or not
SArray* pInterpCols; // interpolation columns
int32_t resultTsOrder; // result timestamp order
int32_t inputOrder; // input data ts order
EOPTR_EXEC_MODEL execModel; // operator execution model [batch model|stream model]
STimeWindowAggSupp twAggSup;
SArray* pPrevValues; // SArray<SGroupKeys> used to keep the previous not null value for interpolation.
......
source/libs/executor/inc/tsort.h
浏览文件 @ 6d2ce5cd
......@@ -146,7 +146,7 @@ void* tsortGetValue(STupleHandle* pVHandle, int32_t colId);
* @return
*/
uint64_t tsortGetGroupId(STupleHandle* pVHandle);
void* tsortGetBlockInfo(STupleHandle* pVHandle);
/**
*
* @param pSortHandle
......
source/libs/executor/src/aggregateoperator.c
浏览文件 @ 6d2ce5cd
......@@ -108,6 +108,8 @@ SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SAggPhysiN
pInfo->binfo.mergeResultBlock = pAggNode->mergeDataBlock;
pInfo->groupKeyOptimized = pAggNode->groupKeyOptimized;
pInfo->groupId = UINT64_MAX;
pInfo->binfo.inputTsOrder = pAggNode->node.inputTsOrder;
pInfo->binfo.outputTsOrder = pAggNode->node.outputTsOrder;
setOperatorInfo(pOperator, "TableAggregate", QUERY_NODE_PHYSICAL_PLAN_HASH_AGG, true, OP_NOT_OPENED, pInfo,
pTaskInfo);
......@@ -164,10 +166,8 @@ int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) {
SOperatorInfo* downstream = pOperator->pDownstream[0];
int64_t st = taosGetTimestampUs();
int32_t order = TSDB_ORDER_ASC;
int32_t scanFlag = MAIN_SCAN;
int32_t code = TSDB_CODE_SUCCESS;
int32_t order = pAggInfo->binfo.inputTsOrder;
bool hasValidBlock = false;
while (1) {
......@@ -185,12 +185,7 @@ int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) {
}
}
hasValidBlock = true;
int32_t code = getTableScanInfo(pOperator, &order, &scanFlag, false);
if (code != TSDB_CODE_SUCCESS) {
destroyDataBlockForEmptyInput(blockAllocated, &pBlock);
T_LONG_JMP(pTaskInfo->env, code);
}
pAggInfo->binfo.pRes->info.scanFlag = pBlock->info.scanFlag;
// there is an scalar expression that needs to be calculated before apply the group aggregation.
if (pAggInfo->scalarExprSup.pExprInfo != NULL && !blockAllocated) {
......@@ -204,7 +199,7 @@ int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) {
// 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);
setInputDataBlock(pSup, pBlock, order, pBlock->info.scanFlag, true);
code = doAggregateImpl(pOperator, pSup->pCtx);
if (code != 0) {
destroyDataBlockForEmptyInput(blockAllocated, &pBlock);
......
source/libs/executor/src/cachescanoperator.c
浏览文件 @ 6d2ce5cd
......@@ -208,6 +208,7 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
pRes->info.id.uid = *(tb_uid_t*)taosArrayGet(pInfo->pUidList, pInfo->indexOfBufferedRes);
pRes->info.rows = 1;
pRes->info.scanFlag = MAIN_SCAN;
SExprSupp* pSup = &pInfo->pseudoExprSup;
int32_t code = addTagPseudoColumnData(&pInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pRes,
......
source/libs/executor/src/eventwindowoperator.c
浏览文件 @ 6d2ce5cd
......@@ -120,6 +120,8 @@ SOperatorInfo* createEventwindowOperatorInfo(SOperatorInfo* downstream, SPhysiNo
initBasicInfo(&pInfo->binfo, pResBlock);
initResultRowInfo(&pInfo->binfo.resultRowInfo);
pInfo->binfo.inputTsOrder = physiNode->inputTsOrder;
pInfo->binfo.outputTsOrder = physiNode->outputTsOrder;
pInfo->twAggSup = (STimeWindowAggSupp){.waterMark = pEventWindowNode->window.watermark,
.calTrigger = pEventWindowNode->window.triggerType};
......@@ -183,7 +185,7 @@ static SSDataBlock* eventWindowAggregate(SOperatorInfo* pOperator) {
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SExprSupp* pSup = &pOperator->exprSupp;
int32_t order = TSDB_ORDER_ASC;
int32_t order = pInfo->binfo.inputTsOrder;
SSDataBlock* pRes = pInfo->binfo.pRes;
......@@ -196,6 +198,7 @@ static SSDataBlock* eventWindowAggregate(SOperatorInfo* pOperator) {
break;
}
pRes->info.scanFlag = pBlock->info.scanFlag;
setInputDataBlock(pSup, pBlock, order, MAIN_SCAN, true);
blockDataUpdateTsWindow(pBlock, pInfo->tsSlotId);
......
source/libs/executor/src/exchangeoperator.c
浏览文件 @ 6d2ce5cd
......@@ -520,6 +520,7 @@ int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, char* pData, SArray* pCo
// data from mnode
pRes->info.dataLoad = 1;
pRes->info.rows = pBlock->info.rows;
pRes->info.scanFlag = MAIN_SCAN;
relocateColumnData(pRes, pColList, pBlock->pDataBlock, false);
blockDataDestroy(pBlock);
}
......
source/libs/executor/src/filloperator.c
浏览文件 @ 6d2ce5cd
......@@ -178,16 +178,15 @@ static SSDataBlock* doFillImpl(SOperatorInfo* pOperator) {
blockDataCleanup(pResBlock);
int32_t order = TSDB_ORDER_ASC;
int32_t scanFlag = MAIN_SCAN;
getTableScanInfo(pOperator, &order, &scanFlag, false);
int32_t order = pInfo->pFillInfo->order;
SOperatorInfo* pDownstream = pOperator->pDownstream[0];
#if 0
// the scan order may be different from the output result order for agg interval operator.
if (pDownstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL) {
order = ((SIntervalAggOperatorInfo*) pDownstream->info)->resultTsOrder;
}
#endif
doHandleRemainBlockFromNewGroup(pOperator, pInfo, pResultInfo, order);
if (pResBlock->info.rows > 0) {
......@@ -206,13 +205,14 @@ static SSDataBlock* doFillImpl(SOperatorInfo* pOperator) {
taosFillSetStartInfo(pInfo->pFillInfo, 0, pInfo->win.ekey);
} else {
pResBlock->info.scanFlag = pBlock->info.scanFlag;
pBlock->info.dataLoad = 1;
blockDataUpdateTsWindow(pBlock, pInfo->primarySrcSlotId);
blockDataCleanup(pInfo->pRes);
blockDataEnsureCapacity(pInfo->pRes, pBlock->info.rows);
blockDataEnsureCapacity(pInfo->pFinalRes, pBlock->info.rows);
doApplyScalarCalculation(pOperator, pBlock, order, scanFlag);
doApplyScalarCalculation(pOperator, pBlock, order, pBlock->info.scanFlag);
if (pInfo->curGroupId == 0 || (pInfo->curGroupId == pInfo->pRes->info.id.groupId)) {
if (pInfo->curGroupId == 0 && taosFillNotStarted(pInfo->pFillInfo)) {
......@@ -405,7 +405,7 @@ SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode*
? &((SMergeAlignedIntervalAggOperatorInfo*)downstream->info)->intervalAggOperatorInfo->interval
: &((SIntervalAggOperatorInfo*)downstream->info)->interval;
int32_t order = (pPhyFillNode->inputTsOrder == ORDER_ASC) ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
int32_t order = (pPhyFillNode->node.inputTsOrder == ORDER_ASC) ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
int32_t type = convertFillType(pPhyFillNode->mode);
SResultInfo* pResultInfo = &pOperator->resultInfo;
......
source/libs/executor/src/groupoperator.c
浏览文件 @ 6d2ce5cd
......@@ -378,9 +378,7 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator) {
return buildGroupResultDataBlock(pOperator);
}
int32_t order = TSDB_ORDER_ASC;
int32_t scanFlag = MAIN_SCAN;
int32_t order = pInfo->binfo.inputTsOrder;
int64_t st = taosGetTimestampUs();
SOperatorInfo* downstream = pOperator->pDownstream[0];
......@@ -390,13 +388,10 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator) {
break;
}
int32_t code = getTableScanInfo(pOperator, &order, &scanFlag, false);
if (code != TSDB_CODE_SUCCESS) {
T_LONG_JMP(pTaskInfo->env, code);
}
pInfo->binfo.pRes->info.scanFlag = pBlock->info.scanFlag;
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(&pOperator->exprSupp, pBlock, order, scanFlag, true);
setInputDataBlock(&pOperator->exprSupp, pBlock, order, pBlock->info.scanFlag, true);
// there is an scalar expression that needs to be calculated right before apply the group aggregation.
if (pInfo->scalarSup.pExprInfo != NULL) {
......@@ -481,6 +476,8 @@ SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SAggPhysiNode*
setOperatorInfo(pOperator, "GroupbyAggOperator", 0, true, OP_NOT_OPENED, pInfo, pTaskInfo);
pInfo->binfo.mergeResultBlock = pAggNode->mergeDataBlock;
pInfo->binfo.inputTsOrder = pAggNode->node.inputTsOrder;
pInfo->binfo.outputTsOrder = pAggNode->node.outputTsOrder;
pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, hashGroupbyAggregate, NULL, destroyGroupOperatorInfo,
optrDefaultBufFn, NULL);
......@@ -762,6 +759,7 @@ static SSDataBlock* hashPartition(SOperatorInfo* pOperator) {
break;
}
pInfo->binfo.pRes->info.scanFlag = pBlock->info.scanFlag;
// there is an scalar expression that needs to be calculated right before apply the group aggregation.
if (pInfo->scalarSup.pExprInfo != NULL) {
pTaskInfo->code = projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock, pInfo->scalarSup.pCtx,
......
source/libs/executor/src/joinoperator.c
浏览文件 @ 6d2ce5cd
......@@ -253,9 +253,9 @@ SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t
}
pInfo->inputOrder = TSDB_ORDER_ASC;
if (pJoinNode->inputTsOrder == ORDER_ASC) {
if (pJoinNode->node.inputTsOrder == ORDER_ASC) {
pInfo->inputOrder = TSDB_ORDER_ASC;
} else if (pJoinNode->inputTsOrder == ORDER_DESC) {
} else if (pJoinNode->node.inputTsOrder == ORDER_DESC) {
pInfo->inputOrder = TSDB_ORDER_DESC;
}
......@@ -684,6 +684,7 @@ static void doMergeJoinImpl(struct SOperatorInfo* pOperator, SSDataBlock* pRes)
// the pDataBlock are always the same one, no need to call this again
pRes->info.rows = nrows;
pRes->info.dataLoad = 1;
pRes->info.scanFlag = MAIN_SCAN;
if (pRes->info.rows >= pOperator->resultInfo.threshold) {
break;
}
......
source/libs/executor/src/projectoperator.c
浏览文件 @ 6d2ce5cd
......@@ -93,6 +93,8 @@ SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhys
pInfo->binfo.pRes = pResBlock;
pInfo->pFinalRes = createOneDataBlock(pResBlock, false);
pInfo->binfo.inputTsOrder = pProjPhyNode->node.inputTsOrder;
pInfo->binfo.outputTsOrder = pProjPhyNode->node.outputTsOrder;
if (pTaskInfo->execModel == OPTR_EXEC_MODEL_STREAM) {
pInfo->mergeDataBlocks = false;
......@@ -238,8 +240,9 @@ SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
}
int64_t st = 0;
int32_t order = 0;
int32_t order = pInfo->inputTsOrder;
int32_t scanFlag = 0;
int32_t code = TSDB_CODE_SUCCESS;
if (pOperator->cost.openCost == 0) {
st = taosGetTimestampUs();
......@@ -284,10 +287,10 @@ SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
break;
}
// the pDataBlock are always the same one, no need to call this again
int32_t code = getTableScanInfo(downstream, &order, &scanFlag, false);
if (code != TSDB_CODE_SUCCESS) {
T_LONG_JMP(pTaskInfo->env, code);
if (pProjectInfo->mergeDataBlocks) {
pFinalRes->info.scanFlag = scanFlag = pBlock->info.scanFlag;
} else {
pRes->info.scanFlag = scanFlag = pBlock->info.scanFlag;
}
setInputDataBlock(pSup, pBlock, order, scanFlag, false);
......@@ -406,6 +409,8 @@ SOperatorInfo* createIndefinitOutputOperatorInfo(SOperatorInfo* downstream, SPhy
}
pInfo->binfo.pRes = pResBlock;
pInfo->binfo.inputTsOrder = pNode->inputTsOrder;
pInfo->binfo.outputTsOrder = pNode->outputTsOrder;
pInfo->pPseudoColInfo = setRowTsColumnOutputInfo(pSup->pCtx, numOfExpr);
setOperatorInfo(pOperator, "IndefinitOperator", QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC, false, OP_NOT_OPENED, pInfo,
......@@ -429,18 +434,13 @@ _error:
static void doHandleDataBlock(SOperatorInfo* pOperator, SSDataBlock* pBlock, SOperatorInfo* downstream,
SExecTaskInfo* pTaskInfo) {
int32_t order = 0;
int32_t scanFlag = 0;
SIndefOperatorInfo* pIndefInfo = pOperator->info;
SOptrBasicInfo* pInfo = &pIndefInfo->binfo;
SExprSupp* pSup = &pOperator->exprSupp;
// the pDataBlock are always the same one, no need to call this again
int32_t code = getTableScanInfo(downstream, &order, &scanFlag, false);
if (code != TSDB_CODE_SUCCESS) {
T_LONG_JMP(pTaskInfo->env, code);
}
int32_t order = pInfo->inputTsOrder;
int32_t scanFlag = pBlock->info.scanFlag;
int32_t code = TSDB_CODE_SUCCESS;
// there is an scalar expression that needs to be calculated before apply the group aggregation.
SExprSupp* pScalarSup = &pIndefInfo->scalarSup;
......@@ -506,6 +506,7 @@ SSDataBlock* doApplyIndefinitFunction(SOperatorInfo* pOperator) {
setOperatorCompleted(pOperator);
break;
}
pInfo->pRes->info.scanFlag = pBlock->info.scanFlag;
if (pIndefInfo->groupId == 0 && pBlock->info.id.groupId != 0) {
pIndefInfo->groupId = pBlock->info.id.groupId; // this is the initial group result
......
source/libs/executor/src/scanoperator.c
浏览文件 @ 6d2ce5cd
......@@ -712,6 +712,7 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
pTableScanInfo->base.readRecorder.elapsedTime += (taosGetTimestampUs() - st) / 1000.0;
pOperator->cost.totalCost = pTableScanInfo->base.readRecorder.elapsedTime;
pBlock->info.scanFlag = pTableScanInfo->base.scanFlag;
return pBlock;
}
return NULL;
......
source/libs/executor/src/sortoperator.c
浏览文件 @ 6d2ce5cd
......@@ -69,6 +69,8 @@ SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode*
pInfo->binfo.pRes = createDataBlockFromDescNode(pDescNode);
pInfo->pSortInfo = createSortInfo(pSortNode->pSortKeys);
pInfo->binfo.inputTsOrder = pSortNode->node.inputTsOrder;
pInfo->binfo.outputTsOrder = pSortNode->node.outputTsOrder;
initLimitInfo(pSortNode->node.pLimit, pSortNode->node.pSlimit, &pInfo->limitInfo);
setOperatorInfo(pOperator, "SortOperator", QUERY_NODE_PHYSICAL_PLAN_SORT, true, OP_NOT_OPENED, pInfo, pTaskInfo);
......@@ -114,6 +116,7 @@ void appendOneRowToDataBlock(SSDataBlock* pBlock, STupleHandle* pTupleHandle) {
}
pBlock->info.dataLoad = 1;
pBlock->info.scanFlag = ((SDataBlockInfo*)tsortGetBlockInfo(pTupleHandle))->scanFlag;
pBlock->info.rows += 1;
}
......@@ -155,6 +158,7 @@ SSDataBlock* getSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlock, i
pDataBlock->info.dataLoad = 1;
pDataBlock->info.rows = p->info.rows;
pDataBlock->info.scanFlag = p->info.scanFlag;
}
blockDataDestroy(p);
......@@ -331,6 +335,7 @@ SSDataBlock* getGroupSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlo
pDataBlock->info.rows = p->info.rows;
pDataBlock->info.capacity = p->info.rows;
pDataBlock->info.scanFlag = p->info.scanFlag;
}
blockDataDestroy(p);
......@@ -505,6 +510,8 @@ SOperatorInfo* createGroupSortOperatorInfo(SOperatorInfo* downstream, SGroupSort
pInfo->binfo.pRes = createDataBlockFromDescNode(pDescNode);
blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
pInfo->binfo.inputTsOrder = pSortPhyNode->node.inputTsOrder;
pInfo->binfo.outputTsOrder = pSortPhyNode->node.outputTsOrder;
int32_t numOfOutputCols = 0;
int32_t code = extractColMatchInfo(pSortPhyNode->pTargets, pDescNode, &numOfOutputCols, COL_MATCH_FROM_SLOT_ID,
......@@ -698,6 +705,7 @@ SSDataBlock* getMultiwaySortedBlockData(SSortHandle* pHandle, SSDataBlock* pData
}
pDataBlock->info.rows = p->info.rows;
pDataBlock->info.scanFlag = p->info.scanFlag;
if (pInfo->ignoreGroupId) {
pDataBlock->info.id.groupId = 0;
} else {
......@@ -799,6 +807,8 @@ SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** downStreams, size
size_t numOfCols = taosArrayGetSize(pInfo->binfo.pRes->pDataBlock);
pInfo->bufPageSize = getProperSortPageSize(rowSize, numOfCols);
pInfo->sortBufSize = pInfo->bufPageSize * (numStreams + 1); // one additional is reserved for merged result.
pInfo->binfo.inputTsOrder = pMergePhyNode->node.inputTsOrder;
pInfo->binfo.outputTsOrder = pMergePhyNode->node.outputTsOrder;
setOperatorInfo(pOperator, "MultiwayMergeOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE, false, OP_NOT_OPENED, pInfo, pTaskInfo);
pOperator->fpSet = createOperatorFpSet(openMultiwayMergeOperator, doMultiwayMerge, NULL,
......
source/libs/executor/src/timesliceoperator.c
浏览文件 @ 6d2ce5cd
......@@ -44,6 +44,8 @@ typedef struct STimeSliceOperatorInfo {
uint64_t groupId;
SGroupKeys* pPrevGroupKey;
SSDataBlock* pNextGroupRes;
SSDataBlock* pRemainRes; // save block unfinished processing
int32_t remainIndex; // the remaining index in the block to be processed
} STimeSliceOperatorInfo;
static void destroyTimeSliceOperatorInfo(void* param);
......@@ -644,13 +646,47 @@ static int32_t resetKeeperInfo(STimeSliceOperatorInfo* pInfo) {
return TSDB_CODE_SUCCESS;
}
static bool checkThresholdReached(STimeSliceOperatorInfo* pSliceInfo, int32_t threshold) {
SSDataBlock* pResBlock = pSliceInfo->pRes;
if (pResBlock->info.rows > threshold) {
return true;
}
return false;
}
static bool checkWindowBoundReached(STimeSliceOperatorInfo* pSliceInfo) {
if (pSliceInfo->current > pSliceInfo->win.ekey) {
return true;
}
return false;
}
static void saveBlockStatus(STimeSliceOperatorInfo* pSliceInfo, SSDataBlock* pBlock, int32_t curIndex) {
SSDataBlock* pResBlock = pSliceInfo->pRes;
SColumnInfoData* pTsCol = taosArrayGet(pBlock->pDataBlock, pSliceInfo->tsCol.slotId);
if (curIndex < pBlock->info.rows - 1) {
pSliceInfo->pRemainRes = pBlock;
pSliceInfo->remainIndex = curIndex + 1;
return;
}
// all data in remaining block processed
pSliceInfo->pRemainRes = NULL;
}
static void doTimesliceImpl(SOperatorInfo* pOperator, STimeSliceOperatorInfo* pSliceInfo, SSDataBlock* pBlock,
SExecTaskInfo* pTaskInfo, bool ignoreNull) {
SSDataBlock* pResBlock = pSliceInfo->pRes;
SInterval* pInterval = &pSliceInfo->interval;
SColumnInfoData* pTsCol = taosArrayGet(pBlock->pDataBlock, pSliceInfo->tsCol.slotId);
for (int32_t i = 0; i < pBlock->info.rows; ++i) {
int32_t i = (pSliceInfo->pRemainRes == NULL) ? 0 : pSliceInfo->remainIndex;
for (; i < pBlock->info.rows; ++i) {
int64_t ts = *(int64_t*)colDataGetData(pTsCol, i);
// check for duplicate timestamps
......@@ -662,10 +698,6 @@ static void doTimesliceImpl(SOperatorInfo* pOperator, STimeSliceOperatorInfo* pS
continue;
}
if (pSliceInfo->current > pSliceInfo->win.ekey) {
break;
}
if (ts == pSliceInfo->current) {
addCurrentRowToResult(pSliceInfo, &pOperator->exprSupp, pResBlock, pBlock, i);
......@@ -674,9 +706,14 @@ static void doTimesliceImpl(SOperatorInfo* pOperator, STimeSliceOperatorInfo* pS
pSliceInfo->current =
taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
if (pSliceInfo->current > pSliceInfo->win.ekey) {
if (checkWindowBoundReached(pSliceInfo)) {
break;
}
if (checkThresholdReached(pSliceInfo, pOperator->resultInfo.threshold)) {
saveBlockStatus(pSliceInfo, pBlock, i);
return;
}
} else if (ts < pSliceInfo->current) {
// in case of interpolation window starts and ends between two datapoints, fill(prev) need to interpolate
doKeepPrevRows(pSliceInfo, pBlock, i);
......@@ -697,9 +734,13 @@ static void doTimesliceImpl(SOperatorInfo* pOperator, STimeSliceOperatorInfo* pS
}
}
if (pSliceInfo->current > pSliceInfo->win.ekey) {
if (checkWindowBoundReached(pSliceInfo)) {
break;
}
if (checkThresholdReached(pSliceInfo, pOperator->resultInfo.threshold)) {
saveBlockStatus(pSliceInfo, pBlock, i);
return;
}
} else {
// ignore current row, and do nothing
}
......@@ -730,11 +771,20 @@ static void doTimesliceImpl(SOperatorInfo* pOperator, STimeSliceOperatorInfo* pS
}
doKeepPrevRows(pSliceInfo, pBlock, i);
if (pSliceInfo->current > pSliceInfo->win.ekey) {
if (checkWindowBoundReached(pSliceInfo)) {
break;
}
if (checkThresholdReached(pSliceInfo, pOperator->resultInfo.threshold)) {
saveBlockStatus(pSliceInfo, pBlock, i);
return;
}
}
}
// if reached here, meaning block processing finished naturally,
// or interpolation reach window upper bound
pSliceInfo->pRemainRes = NULL;
}
static void genInterpAfterDataBlock(STimeSliceOperatorInfo* pSliceInfo, SOperatorInfo* pOperator, int32_t index) {
......@@ -781,39 +831,69 @@ static void resetTimesliceInfo(STimeSliceOperatorInfo* pSliceInfo) {
resetKeeperInfo(pSliceInfo);
}
static void doHandleTimeslice(SOperatorInfo* pOperator, SSDataBlock* pBlock) {
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
STimeSliceOperatorInfo* pSliceInfo = pOperator->info;
SExprSupp* pSup = &pOperator->exprSupp;
bool ignoreNull = getIgoreNullRes(pSup);
int32_t order = TSDB_ORDER_ASC;
int32_t code = initKeeperInfo(pSliceInfo, pBlock, &pOperator->exprSupp);
if (code != TSDB_CODE_SUCCESS) {
T_LONG_JMP(pTaskInfo->env, code);
}
if (pSliceInfo->scalarSup.pExprInfo != NULL) {
SExprSupp* pExprSup = &pSliceInfo->scalarSup;
projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
}
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pSup, pBlock, order, MAIN_SCAN, true);
doTimesliceImpl(pOperator, pSliceInfo, pBlock, pTaskInfo, ignoreNull);
copyPrevGroupKey(&pOperator->exprSupp, pSliceInfo->pPrevGroupKey, pBlock);
}
static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
}
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
STimeSliceOperatorInfo* pSliceInfo = pOperator->info;
SSDataBlock* pResBlock = pSliceInfo->pRes;
SExprSupp* pSup = &pOperator->exprSupp;
bool ignoreNull = getIgoreNullRes(pSup);
int32_t order = TSDB_ORDER_ASC;
SInterval* pInterval = &pSliceInfo->interval;
SOperatorInfo* downstream = pOperator->pDownstream[0];
blockDataCleanup(pResBlock);
while (1) {
if (pSliceInfo->pNextGroupRes != NULL) {
setInputDataBlock(pSup, pSliceInfo->pNextGroupRes, order, MAIN_SCAN, true);
doTimesliceImpl(pOperator, pSliceInfo, pSliceInfo->pNextGroupRes, pTaskInfo, ignoreNull);
copyPrevGroupKey(&pOperator->exprSupp, pSliceInfo->pPrevGroupKey, pSliceInfo->pNextGroupRes);
doHandleTimeslice(pOperator, pSliceInfo->pNextGroupRes);
if (checkWindowBoundReached(pSliceInfo) || checkThresholdReached(pSliceInfo, pOperator->resultInfo.threshold)) {
doFilter(pResBlock, pOperator->exprSupp.pFilterInfo, NULL);
if (pSliceInfo->pRemainRes == NULL) {
pSliceInfo->pNextGroupRes = NULL;
}
if (pResBlock->info.rows != 0) {
goto _finished;
} else {
// after fillter if result block has 0 rows, go back to
// process pNextGroupRes again for unfinished data
continue;
}
}
pSliceInfo->pNextGroupRes = NULL;
}
while (1) {
SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
SSDataBlock* pBlock = pSliceInfo->pRemainRes ? pSliceInfo->pRemainRes : downstream->fpSet.getNextFn(downstream);
if (pBlock == NULL) {
setOperatorCompleted(pOperator);
break;
}
pResBlock->info.scanFlag = pBlock->info.scanFlag;
if (pSliceInfo->groupId == 0 && pBlock->info.id.groupId != 0) {
pSliceInfo->groupId = pBlock->info.id.groupId;
} else {
......@@ -824,21 +904,15 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
}
}
if (pSliceInfo->scalarSup.pExprInfo != NULL) {
SExprSupp* pExprSup = &pSliceInfo->scalarSup;
projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
}
int32_t code = initKeeperInfo(pSliceInfo, pBlock, &pOperator->exprSupp);
if (code != TSDB_CODE_SUCCESS) {
T_LONG_JMP(pTaskInfo->env, code);
doHandleTimeslice(pOperator, pBlock);
if (checkWindowBoundReached(pSliceInfo) || checkThresholdReached(pSliceInfo, pOperator->resultInfo.threshold)) {
doFilter(pResBlock, pOperator->exprSupp.pFilterInfo, NULL);
if (pResBlock->info.rows != 0) {
goto _finished;
}
}
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pSup, pBlock, order, MAIN_SCAN, true);
doTimesliceImpl(pOperator, pSliceInfo, pBlock, pTaskInfo, ignoreNull);
copyPrevGroupKey(&pOperator->exprSupp, pSliceInfo->pPrevGroupKey, pBlock);
}
// post work for a specific group
// check if need to interpolate after last datablock
// except for fill(next), fill(linear)
......@@ -851,11 +925,12 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
// restore initial value for next group
resetTimesliceInfo(pSliceInfo);
if (pResBlock->info.rows >= 4096) {
if (pResBlock->info.rows != 0) {
break;
}
}
_finished:
// restore the value
setTaskStatus(pOperator->pTaskInfo, TASK_COMPLETED);
if (pResBlock->info.rows == 0) {
......@@ -911,6 +986,8 @@ SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SPhysiNode
pInfo->groupId = 0;
pInfo->pPrevGroupKey = NULL;
pInfo->pNextGroupRes = NULL;
pInfo->pRemainRes = NULL;
pInfo->remainIndex = 0;
if (downstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {
STableScanInfo* pScanInfo = (STableScanInfo*)downstream->info;
......
source/libs/executor/src/timewindowoperator.c
浏览文件 @ 6d2ce5cd
......@@ -355,7 +355,7 @@ static void setNotInterpoWindowKey(SqlFunctionCtx* pCtx, int32_t numOfOutput, in
static bool setTimeWindowInterpolationStartTs(SIntervalAggOperatorInfo* pInfo, int32_t pos, SSDataBlock* pBlock,
const TSKEY* tsCols, STimeWindow* win, SExprSupp* pSup) {
bool ascQuery = (pInfo->inputOrder == TSDB_ORDER_ASC);
bool ascQuery = (pInfo->binfo.inputTsOrder == TSDB_ORDER_ASC);
TSKEY curTs = tsCols[pos];
......@@ -385,7 +385,7 @@ static bool setTimeWindowInterpolationStartTs(SIntervalAggOperatorInfo* pInfo, i
static bool setTimeWindowInterpolationEndTs(SIntervalAggOperatorInfo* pInfo, SExprSupp* pSup, int32_t endRowIndex,
SArray* pDataBlock, const TSKEY* tsCols, TSKEY blockEkey,
STimeWindow* win) {
int32_t order = pInfo->inputOrder;
int32_t order = pInfo->binfo.inputTsOrder;
TSKEY actualEndKey = tsCols[endRowIndex];
TSKEY key = (order == TSDB_ORDER_ASC) ? win->ekey : win->skey;
......@@ -547,7 +547,7 @@ static void doWindowBorderInterpolation(SIntervalAggOperatorInfo* pInfo, SSDataB
if (!done) {
int32_t endRowIndex = startPos + forwardRows - 1;
TSKEY endKey = (pInfo->inputOrder == TSDB_ORDER_ASC) ? pBlock->info.window.ekey : pBlock->info.window.skey;
TSKEY endKey = (pInfo->binfo.inputTsOrder == TSDB_ORDER_ASC) ? pBlock->info.window.ekey : pBlock->info.window.skey;
bool interp = setTimeWindowInterpolationEndTs(pInfo, pSup, endRowIndex, pBlock->pDataBlock, tsCols, endKey, win);
if (interp) {
setResultRowInterpo(pResult, RESULT_ROW_END_INTERP);
......@@ -885,12 +885,12 @@ static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResul
int32_t numOfOutput = pSup->numOfExprs;
int64_t* tsCols = extractTsCol(pBlock, pInfo);
uint64_t tableGroupId = pBlock->info.id.groupId;
bool ascScan = (pInfo->inputOrder == TSDB_ORDER_ASC);
bool ascScan = (pInfo->binfo.inputTsOrder == TSDB_ORDER_ASC);
TSKEY ts = getStartTsKey(&pBlock->info.window, tsCols);
SResultRow* pResult = NULL;
STimeWindow win =
getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts, &pInfo->interval, pInfo->inputOrder);
getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts, &pInfo->interval, pInfo->binfo.inputTsOrder);
int32_t ret = setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId,
pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
if (ret != TSDB_CODE_SUCCESS || pResult == NULL) {
......@@ -899,7 +899,7 @@ static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResul
TSKEY ekey = ascScan ? win.ekey : win.skey;
int32_t forwardRows =
getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, pInfo->inputOrder);
getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, pInfo->binfo.inputTsOrder);
// prev time window not interpolation yet.
if (pInfo->timeWindowInterpo) {
......@@ -926,7 +926,7 @@ static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResul
STimeWindow nextWin = win;
while (1) {
int32_t prevEndPos = forwardRows - 1 + startPos;
startPos = getNextQualifiedWindow(&pInfo->interval, &nextWin, &pBlock->info, tsCols, prevEndPos, pInfo->inputOrder);
startPos = getNextQualifiedWindow(&pInfo->interval, &nextWin, &pBlock->info, tsCols, prevEndPos, pInfo->binfo.inputTsOrder);
if (startPos < 0) {
break;
}
......@@ -939,7 +939,7 @@ static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResul
ekey = ascScan ? nextWin.ekey : nextWin.skey;
forwardRows =
getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, pInfo->inputOrder);
getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, pInfo->binfo.inputTsOrder);
// window start(end) key interpolation
doWindowBorderInterpolation(pInfo, pBlock, pResult, &nextWin, startPos, forwardRows, pSup);
// TODO: add to open window? how to close the open windows after input blocks exhausted?
......@@ -1032,7 +1032,7 @@ static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) {
break;
}
getTableScanInfo(pOperator, &pInfo->inputOrder, &scanFlag, true);
pInfo->binfo.pRes->info.scanFlag = scanFlag = pBlock->info.scanFlag;
if (pInfo->scalarSupp.pExprInfo != NULL) {
SExprSupp* pExprSup = &pInfo->scalarSupp;
......@@ -1040,11 +1040,11 @@ static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) {
}
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pSup, pBlock, pInfo->inputOrder, scanFlag, true);
setInputDataBlock(pSup, pBlock, pInfo->binfo.inputTsOrder, scanFlag, true);
hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, scanFlag);
}
initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, pInfo->resultTsOrder);
initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, pInfo->binfo.outputTsOrder);
OPTR_SET_OPENED(pOperator);
pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;
......@@ -1158,7 +1158,7 @@ static int32_t openStateWindowAggOptr(SOperatorInfo* pOperator) {
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SExprSupp* pSup = &pOperator->exprSupp;
int32_t order = TSDB_ORDER_ASC;
int32_t order = pInfo->binfo.inputTsOrder;
int64_t st = taosGetTimestampUs();
SOperatorInfo* downstream = pOperator->pDownstream[0];
......@@ -1168,6 +1168,7 @@ static int32_t openStateWindowAggOptr(SOperatorInfo* pOperator) {
break;
}
pInfo->binfo.pRes->info.scanFlag = pBlock->info.scanFlag;
setInputDataBlock(pSup, pBlock, order, MAIN_SCAN, true);
blockDataUpdateTsWindow(pBlock, pInfo->tsSlotId);
......@@ -1650,8 +1651,8 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SIntervalPh
};
pInfo->win = pTaskInfo->window;
pInfo->inputOrder = (pPhyNode->window.inputTsOrder == ORDER_ASC) ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
pInfo->resultTsOrder = (pPhyNode->window.outputTsOrder == ORDER_ASC) ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
pInfo->binfo.inputTsOrder = pPhyNode->window.node.inputTsOrder;
pInfo->binfo.outputTsOrder = pPhyNode->window.node.outputTsOrder;
pInfo->interval = interval;
pInfo->twAggSup = as;
pInfo->binfo.mergeResultBlock = pPhyNode->window.mergeDataBlock;
......@@ -1802,7 +1803,7 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) {
}
int64_t st = taosGetTimestampUs();
int32_t order = TSDB_ORDER_ASC;
int32_t order = pInfo->binfo.inputTsOrder;
SOperatorInfo* downstream = pOperator->pDownstream[0];
......@@ -1812,6 +1813,7 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) {
break;
}
pBInfo->pRes->info.scanFlag = pBlock->info.scanFlag;
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pSup, pBlock, order, MAIN_SCAN, true);
blockDataUpdateTsWindow(pBlock, pInfo->tsSlotId);
......@@ -1872,6 +1874,8 @@ SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SStateWi
if (pInfo->stateKey.pData == NULL) {
goto _error;
}
pInfo->binfo.inputTsOrder = pStateNode->window.node.inputTsOrder;
pInfo->binfo.outputTsOrder = pStateNode->window.node.outputTsOrder;
int32_t code = filterInitFromNode((SNode*)pStateNode->window.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);
if (code != TSDB_CODE_SUCCESS) {
......@@ -1970,6 +1974,8 @@ SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SSessionW
pInfo->binfo.pRes = pResBlock;
pInfo->winSup.prevTs = INT64_MIN;
pInfo->reptScan = false;
pInfo->binfo.inputTsOrder = pSessionNode->window.node.inputTsOrder;
pInfo->binfo.outputTsOrder = pSessionNode->window.node.outputTsOrder;
code = filterInitFromNode((SNode*)pSessionNode->window.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
......@@ -4318,7 +4324,6 @@ static void doMergeAlignedIntervalAgg(SOperatorInfo* pOperator) {
SSDataBlock* pRes = pIaInfo->binfo.pRes;
SResultRowInfo* pResultRowInfo = &pIaInfo->binfo.resultRowInfo;
SOperatorInfo* downstream = pOperator->pDownstream[0];
int32_t scanFlag = MAIN_SCAN;
while (1) {
SSDataBlock* pBlock = NULL;
......@@ -4365,8 +4370,8 @@ static void doMergeAlignedIntervalAgg(SOperatorInfo* pOperator) {
}
}
getTableScanInfo(pOperator, &pIaInfo->inputOrder, &scanFlag, false);
setInputDataBlock(pSup, pBlock, pIaInfo->inputOrder, scanFlag, true);
pRes->info.scanFlag = pBlock->info.scanFlag;
setInputDataBlock(pSup, pBlock, pIaInfo->binfo.inputTsOrder, pBlock->info.scanFlag, true);
doMergeAlignedIntervalAggImpl(pOperator, &pIaInfo->binfo.resultRowInfo, pBlock, pRes);
doFilter(pRes, pOperator->exprSupp.pFilterInfo, NULL);
......@@ -4439,7 +4444,8 @@ SOperatorInfo* createMergeAlignedIntervalOperatorInfo(SOperatorInfo* downstream,
miaInfo->curTs = INT64_MIN;
iaInfo->win = pTaskInfo->window;
iaInfo->inputOrder = TSDB_ORDER_ASC;
iaInfo->binfo.inputTsOrder = pNode->window.node.inputTsOrder;
iaInfo->binfo.outputTsOrder = pNode->window.node.outputTsOrder;
iaInfo->interval = interval;
iaInfo->primaryTsIndex = ((SColumnNode*)pNode->window.pTspk)->slotId;
iaInfo->binfo.mergeResultBlock = pNode->window.mergeDataBlock;
......@@ -4516,7 +4522,7 @@ static int32_t outputPrevIntervalResult(SOperatorInfo* pOperatorInfo, uint64_t t
STimeWindow* newWin) {
SMergeIntervalAggOperatorInfo* miaInfo = pOperatorInfo->info;
SIntervalAggOperatorInfo* iaInfo = &miaInfo->intervalAggOperatorInfo;
bool ascScan = (iaInfo->inputOrder == TSDB_ORDER_ASC);
bool ascScan = (iaInfo->binfo.inputTsOrder == TSDB_ORDER_ASC);
SGroupTimeWindow groupTimeWindow = {.groupId = tableGroupId, .window = *newWin};
tdListAppend(miaInfo->groupIntervals, &groupTimeWindow);
......@@ -4551,12 +4557,12 @@ static void doMergeIntervalAggImpl(SOperatorInfo* pOperatorInfo, SResultRowInfo*
int32_t numOfOutput = pExprSup->numOfExprs;
int64_t* tsCols = extractTsCol(pBlock, iaInfo);
uint64_t tableGroupId = pBlock->info.id.groupId;
bool ascScan = (iaInfo->inputOrder == TSDB_ORDER_ASC);
bool ascScan = (iaInfo->binfo.inputTsOrder == TSDB_ORDER_ASC);
TSKEY blockStartTs = getStartTsKey(&pBlock->info.window, tsCols);
SResultRow* pResult = NULL;
STimeWindow win = getActiveTimeWindow(iaInfo->aggSup.pResultBuf, pResultRowInfo, blockStartTs, &iaInfo->interval,
iaInfo->inputOrder);
iaInfo->binfo.inputTsOrder);
int32_t ret =
setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, pExprSup->pCtx,
......@@ -4567,7 +4573,7 @@ static void doMergeIntervalAggImpl(SOperatorInfo* pOperatorInfo, SResultRowInfo*
TSKEY ekey = ascScan ? win.ekey : win.skey;
int32_t forwardRows =
getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, iaInfo->inputOrder);
getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, iaInfo->binfo.inputTsOrder);
ASSERT(forwardRows > 0);
// prev time window not interpolation yet.
......@@ -4598,7 +4604,7 @@ static void doMergeIntervalAggImpl(SOperatorInfo* pOperatorInfo, SResultRowInfo*
while (1) {
int32_t prevEndPos = forwardRows - 1 + startPos;
startPos =
getNextQualifiedWindow(&iaInfo->interval, &nextWin, &pBlock->info, tsCols, prevEndPos, iaInfo->inputOrder);
getNextQualifiedWindow(&iaInfo->interval, &nextWin, &pBlock->info, tsCols, prevEndPos, iaInfo->binfo.inputTsOrder);
if (startPos < 0) {
break;
}
......@@ -4613,7 +4619,7 @@ static void doMergeIntervalAggImpl(SOperatorInfo* pOperatorInfo, SResultRowInfo*
ekey = ascScan ? nextWin.ekey : nextWin.skey;
forwardRows =
getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, iaInfo->inputOrder);
getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, iaInfo->binfo.inputTsOrder);
// window start(end) key interpolation
doWindowBorderInterpolation(iaInfo, pBlock, pResult, &nextWin, startPos, forwardRows, pExprSup);
......@@ -4649,7 +4655,6 @@ static SSDataBlock* doMergeIntervalAgg(SOperatorInfo* pOperator) {
if (!miaInfo->inputBlocksFinished) {
SOperatorInfo* downstream = pOperator->pDownstream[0];
int32_t scanFlag = MAIN_SCAN;
while (1) {
SSDataBlock* pBlock = NULL;
if (miaInfo->prefetchedBlock == NULL) {
......@@ -4674,9 +4679,9 @@ static SSDataBlock* doMergeIntervalAgg(SOperatorInfo* pOperator) {
break;
}
getTableScanInfo(pOperator, &iaInfo->inputOrder, &scanFlag, false);
setInputDataBlock(pExpSupp, pBlock, iaInfo->inputOrder, scanFlag, true);
doMergeIntervalAggImpl(pOperator, &iaInfo->binfo.resultRowInfo, pBlock, scanFlag, pRes);
pRes->info.scanFlag = pBlock->info.scanFlag;
setInputDataBlock(pExpSupp, pBlock, iaInfo->binfo.inputTsOrder, pBlock->info.scanFlag, true);
doMergeIntervalAggImpl(pOperator, &iaInfo->binfo.resultRowInfo, pBlock, pBlock->info.scanFlag, pRes);
if (pRes->info.rows >= pOperator->resultInfo.threshold) {
break;
......@@ -4726,10 +4731,11 @@ SOperatorInfo* createMergeIntervalOperatorInfo(SOperatorInfo* downstream, SMerge
SIntervalAggOperatorInfo* pIntervalInfo = &pMergeIntervalInfo->intervalAggOperatorInfo;
pIntervalInfo->win = pTaskInfo->window;
pIntervalInfo->inputOrder = TSDB_ORDER_ASC;
pIntervalInfo->binfo.inputTsOrder = pIntervalPhyNode->window.node.inputTsOrder;
pIntervalInfo->interval = interval;
pIntervalInfo->binfo.mergeResultBlock = pIntervalPhyNode->window.mergeDataBlock;
pIntervalInfo->primaryTsIndex = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId;
pIntervalInfo->binfo.outputTsOrder = pIntervalPhyNode->window.node.outputTsOrder;
SExprSupp* pExprSupp = &pOperator->exprSupp;
......
source/libs/executor/src/tsort.c
浏览文件 @ 6d2ce5cd
......@@ -887,6 +887,7 @@ void* tsortGetValue(STupleHandle* pVHandle, int32_t colIndex) {
}
uint64_t tsortGetGroupId(STupleHandle* pVHandle) { return pVHandle->pBlock->info.id.groupId; }
void* tsortGetBlockInfo(STupleHandle* pVHandle) { return &pVHandle->pBlock->info; }
SSortExecInfo tsortGetSortExecInfo(SSortHandle* pHandle) {
SSortExecInfo info = {0};
......
source/libs/nodes/src/nodesCloneFuncs.c
浏览文件 @ 6d2ce5cd
......@@ -110,6 +110,7 @@ static int32_t columnNodeCopy(const SColumnNode* pSrc, SColumnNode* pDst) {
COPY_SCALAR_FIELD(tableId);
COPY_SCALAR_FIELD(tableType);
COPY_SCALAR_FIELD(colId);
COPY_SCALAR_FIELD(projIdx);
COPY_SCALAR_FIELD(colType);
COPY_SCALAR_FIELD(hasIndex);
COPY_CHAR_ARRAY_FIELD(dbName);
......@@ -358,6 +359,8 @@ static int32_t logicNodeCopy(const SLogicNode* pSrc, SLogicNode* pDst) {
COPY_SCALAR_FIELD(requireDataOrder);
COPY_SCALAR_FIELD(resultDataOrder);
COPY_SCALAR_FIELD(groupAction);
COPY_SCALAR_FIELD(inputTsOrder);
COPY_SCALAR_FIELD(outputTsOrder);
return TSDB_CODE_SUCCESS;
}
......@@ -404,7 +407,6 @@ static int32_t logicJoinCopy(const SJoinLogicNode* pSrc, SJoinLogicNode* pDst) {
CLONE_NODE_FIELD(pOnConditions);
CLONE_NODE_FIELD(pColEqualOnConditions);
COPY_SCALAR_FIELD(isSingleTableJoin);
COPY_SCALAR_FIELD(inputTsOrder);
return TSDB_CODE_SUCCESS;
}
......@@ -482,8 +484,6 @@ static int32_t logicWindowCopy(const SWindowLogicNode* pSrc, SWindowLogicNode* p
COPY_SCALAR_FIELD(igExpired);
COPY_SCALAR_FIELD(igCheckUpdate);
COPY_SCALAR_FIELD(windowAlgo);
COPY_SCALAR_FIELD(inputTsOrder);
COPY_SCALAR_FIELD(outputTsOrder);
return TSDB_CODE_SUCCESS;
}
......@@ -495,7 +495,6 @@ static int32_t logicFillCopy(const SFillLogicNode* pSrc, SFillLogicNode* pDst) {
CLONE_NODE_FIELD(pWStartTs);
CLONE_NODE_FIELD(pValues);
COPY_OBJECT_FIELD(timeRange, sizeof(STimeWindow));
COPY_SCALAR_FIELD(inputTsOrder);
return TSDB_CODE_SUCCESS;
}
......@@ -544,6 +543,8 @@ static int32_t physiNodeCopy(const SPhysiNode* pSrc, SPhysiNode* pDst) {
CLONE_NODE_FIELD_EX(pOutputDataBlockDesc, SDataBlockDescNode*);
CLONE_NODE_FIELD(pConditions);
CLONE_NODE_LIST_FIELD(pChildren);
COPY_SCALAR_FIELD(inputTsOrder);
COPY_SCALAR_FIELD(outputTsOrder);
return TSDB_CODE_SUCCESS;
}
......
source/libs/nodes/src/nodesCodeFuncs.c
浏览文件 @ 6d2ce5cd
......@@ -1904,9 +1904,6 @@ static int32_t physiJoinNodeToJson(const void* pObj, SJson* pJson) {
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkJoinPhysiPlanJoinType, pNode->joinType);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkJoinPhysiPlanInputTsOrder, pNode->inputTsOrder);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkJoinPhysiPlanMergeCondition, nodeToJson, pNode->pMergeCondition);
}
......@@ -1929,9 +1926,6 @@ static int32_t jsonToPhysiJoinNode(const SJson* pJson, void* pObj) {
if (TSDB_CODE_SUCCESS == code) {
tjsonGetNumberValue(pJson, jkJoinPhysiPlanJoinType, pNode->joinType, code);
}
if (TSDB_CODE_SUCCESS == code) {
tjsonGetNumberValue(pJson, jkJoinPhysiPlanInputTsOrder, pNode->inputTsOrder, code);
}
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeObject(pJson, jkJoinPhysiPlanOnConditions, &pNode->pOnConditions);
}
......@@ -2150,7 +2144,6 @@ static const char* jkWindowPhysiPlanWatermark = "Watermark";
static const char* jkWindowPhysiPlanDeleteMark = "DeleteMark";
static const char* jkWindowPhysiPlanIgnoreExpired = "IgnoreExpired";
static const char* jkWindowPhysiPlanInputTsOrder = "InputTsOrder";
static const char* jkWindowPhysiPlanOutputTsOrder = "outputTsOrder";
static const char* jkWindowPhysiPlanMergeDataBlock = "MergeDataBlock";
static int32_t physiWindowNodeToJson(const void* pObj, SJson* pJson) {
......@@ -2181,12 +2174,6 @@ static int32_t physiWindowNodeToJson(const void* pObj, SJson* pJson) {
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkWindowPhysiPlanIgnoreExpired, pNode->igExpired);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkWindowPhysiPlanInputTsOrder, pNode->inputTsOrder);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkWindowPhysiPlanOutputTsOrder, pNode->outputTsOrder);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddBoolToObject(pJson, jkWindowPhysiPlanMergeDataBlock, pNode->mergeDataBlock);
}
......@@ -2222,12 +2209,6 @@ static int32_t jsonToPhysiWindowNode(const SJson* pJson, void* pObj) {
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetTinyIntValue(pJson, jkWindowPhysiPlanIgnoreExpired, &pNode->igExpired);
}
if (TSDB_CODE_SUCCESS == code) {
tjsonGetNumberValue(pJson, jkWindowPhysiPlanInputTsOrder, pNode->inputTsOrder, code);
}
if (TSDB_CODE_SUCCESS == code) {
tjsonGetNumberValue(pJson, jkWindowPhysiPlanOutputTsOrder, pNode->outputTsOrder, code);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBoolValue(pJson, jkWindowPhysiPlanMergeDataBlock, &pNode->mergeDataBlock);
}
......@@ -2294,7 +2275,6 @@ static const char* jkFillPhysiPlanWStartTs = "WStartTs";
static const char* jkFillPhysiPlanValues = "Values";
static const char* jkFillPhysiPlanStartTime = "StartTime";
static const char* jkFillPhysiPlanEndTime = "EndTime";
static const char* jkFillPhysiPlanInputTsOrder = "inputTsOrder";
static int32_t physiFillNodeToJson(const void* pObj, SJson* pJson) {
const SFillPhysiNode* pNode = (const SFillPhysiNode*)pObj;
......@@ -2321,9 +2301,6 @@ static int32_t physiFillNodeToJson(const void* pObj, SJson* pJson) {
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkFillPhysiPlanEndTime, pNode->timeRange.ekey);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkFillPhysiPlanInputTsOrder, pNode->inputTsOrder);
}
return code;
}
......@@ -2353,9 +2330,6 @@ static int32_t jsonToPhysiFillNode(const SJson* pJson, void* pObj) {
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBigIntValue(pJson, jkFillPhysiPlanEndTime, &pNode->timeRange.ekey);
}
if (TSDB_CODE_SUCCESS == code) {
tjsonGetNumberValue(pJson, jkFillPhysiPlanInputTsOrder, pNode->inputTsOrder, code);
}
return code;
}
......@@ -3053,6 +3027,7 @@ static const char* jkColumnTableId = "TableId";
static const char* jkColumnTableType = "TableType";
static const char* jkColumnColId = "ColId";
static const char* jkColumnColType = "ColType";
static const char* jkColumnProjId = "ProjId";
static const char* jkColumnDbName = "DbName";
static const char* jkColumnTableName = "TableName";
static const char* jkColumnTableAlias = "TableAlias";
......@@ -3073,6 +3048,9 @@ static int32_t columnNodeToJson(const void* pObj, SJson* pJson) {
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkColumnColId, pNode->colId);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkColumnProjId, pNode->projIdx);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkColumnColType, pNode->colType);
}
......@@ -3111,6 +3089,9 @@ static int32_t jsonToColumnNode(const SJson* pJson, void* pObj) {
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetSmallIntValue(pJson, jkColumnColId, &pNode->colId);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetSmallIntValue(pJson, jkColumnProjId, &pNode->projIdx);
}
if (TSDB_CODE_SUCCESS == code) {
tjsonGetNumberValue(pJson, jkColumnColType, pNode->colType, code);
}
......
source/libs/nodes/src/nodesMsgFuncs.c
浏览文件 @ 6d2ce5cd
......@@ -1851,7 +1851,9 @@ enum {
PHY_NODE_CODE_CONDITIONS,
PHY_NODE_CODE_CHILDREN,
PHY_NODE_CODE_LIMIT,
PHY_NODE_CODE_SLIMIT
PHY_NODE_CODE_SLIMIT,
PHY_NODE_CODE_INPUT_TS_ORDER,
PHY_NODE_CODE_OUTPUT_TS_ORDER
};
static int32_t physiNodeToMsg(const void* pObj, STlvEncoder* pEncoder) {
......@@ -1870,6 +1872,12 @@ static int32_t physiNodeToMsg(const void* pObj, STlvEncoder* pEncoder) {
if (TSDB_CODE_SUCCESS == code) {
code = tlvEncodeObj(pEncoder, PHY_NODE_CODE_SLIMIT, nodeToMsg, pNode->pSlimit);
}
if (TSDB_CODE_SUCCESS == code) {
code = tlvEncodeEnum(pEncoder, PHY_NODE_CODE_INPUT_TS_ORDER, pNode->inputTsOrder);
}
if (TSDB_CODE_SUCCESS == code) {
code = tlvEncodeEnum(pEncoder, PHY_NODE_CODE_OUTPUT_TS_ORDER, pNode->outputTsOrder);
}
return code;
}
......@@ -1896,6 +1904,12 @@ static int32_t msgToPhysiNode(STlvDecoder* pDecoder, void* pObj) {
case PHY_NODE_CODE_SLIMIT:
code = msgToNodeFromTlv(pTlv, (void**)&pNode->pSlimit);
break;
case PHY_NODE_CODE_INPUT_TS_ORDER:
code = tlvDecodeEnum(pTlv, &pNode->inputTsOrder, sizeof(pNode->inputTsOrder));
break;
case PHY_NODE_CODE_OUTPUT_TS_ORDER:
code = tlvDecodeEnum(pTlv, &pNode->outputTsOrder, sizeof(pNode->outputTsOrder));
break;
default:
break;
}
......@@ -2339,9 +2353,6 @@ static int32_t physiJoinNodeToMsg(const void* pObj, STlvEncoder* pEncoder) {
if (TSDB_CODE_SUCCESS == code) {
code = tlvEncodeObj(pEncoder, PHY_SORT_MERGE_JOIN_CODE_TARGETS, nodeListToMsg, pNode->pTargets);
}
if (TSDB_CODE_SUCCESS == code) {
code = tlvEncodeEnum(pEncoder, PHY_SORT_MERGE_JOIN_CODE_INPUT_TS_ORDER, pNode->inputTsOrder);
}
if (TSDB_CODE_SUCCESS == code) {
code = tlvEncodeObj(pEncoder, PHY_SORT_MERGE_JOIN_CODE_TAG_EQUAL_CONDITIONS, nodeToMsg, pNode->pColEqualOnConditions);
}
......@@ -2370,9 +2381,6 @@ static int32_t msgToPhysiJoinNode(STlvDecoder* pDecoder, void* pObj) {
case PHY_SORT_MERGE_JOIN_CODE_TARGETS:
code = msgToNodeListFromTlv(pTlv, (void**)&pNode->pTargets);
break;
case PHY_SORT_MERGE_JOIN_CODE_INPUT_TS_ORDER:
code = tlvDecodeEnum(pTlv, &pNode->inputTsOrder, sizeof(pNode->inputTsOrder));
break;
case PHY_SORT_MERGE_JOIN_CODE_TAG_EQUAL_CONDITIONS:
code = msgToNodeFromTlv(pTlv, (void**)&pNode->pColEqualOnConditions);
break;
......@@ -2675,12 +2683,6 @@ static int32_t physiWindowNodeToMsg(const void* pObj, STlvEncoder* pEncoder) {
if (TSDB_CODE_SUCCESS == code) {
code = tlvEncodeI8(pEncoder, PHY_WINDOW_CODE_IG_EXPIRED, pNode->igExpired);
}
if (TSDB_CODE_SUCCESS == code) {
code = tlvEncodeEnum(pEncoder, PHY_WINDOW_CODE_INPUT_TS_ORDER, pNode->inputTsOrder);
}
if (TSDB_CODE_SUCCESS == code) {
code = tlvEncodeEnum(pEncoder, PHY_WINDOW_CODE_OUTPUT_TS_ORDER, pNode->outputTsOrder);
}
if (TSDB_CODE_SUCCESS == code) {
code = tlvEncodeBool(pEncoder, PHY_WINDOW_CODE_MERGE_DATA_BLOCK, pNode->mergeDataBlock);
}
......@@ -2722,12 +2724,6 @@ static int32_t msgToPhysiWindowNode(STlvDecoder* pDecoder, void* pObj) {
case PHY_WINDOW_CODE_IG_EXPIRED:
code = tlvDecodeI8(pTlv, &pNode->igExpired);
break;
case PHY_WINDOW_CODE_INPUT_TS_ORDER:
code = tlvDecodeEnum(pTlv, &pNode->inputTsOrder, sizeof(pNode->inputTsOrder));
break;
case PHY_WINDOW_CODE_OUTPUT_TS_ORDER:
code = tlvDecodeEnum(pTlv, &pNode->outputTsOrder, sizeof(pNode->outputTsOrder));
break;
case PHY_WINDOW_CODE_MERGE_DATA_BLOCK:
code = tlvDecodeBool(pTlv, &pNode->mergeDataBlock);
break;
......@@ -2846,9 +2842,6 @@ static int32_t physiFillNodeToMsg(const void* pObj, STlvEncoder* pEncoder) {
if (TSDB_CODE_SUCCESS == code) {
code = tlvEncodeObj(pEncoder, PHY_FILL_CODE_TIME_RANGE, timeWindowToMsg, &pNode->timeRange);
}
if (TSDB_CODE_SUCCESS == code) {
code = tlvEncodeEnum(pEncoder, PHY_FILL_CODE_INPUT_TS_ORDER, pNode->inputTsOrder);
}
return code;
}
......@@ -2881,9 +2874,6 @@ static int32_t msgToPhysiFillNode(STlvDecoder* pDecoder, void* pObj) {
case PHY_FILL_CODE_TIME_RANGE:
code = tlvDecodeObjFromTlv(pTlv, msgToTimeWindow, (void**)&pNode->timeRange);
break;
case PHY_FILL_CODE_INPUT_TS_ORDER:
code = tlvDecodeEnum(pTlv, &pNode->inputTsOrder, sizeof(pNode->inputTsOrder));
break;
default:
break;
}
......
source/libs/planner/src/planLogicCreater.c
浏览文件 @ 6d2ce5cd
......@@ -436,7 +436,7 @@ static int32_t createJoinLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
pJoin->joinType = pJoinTable->joinType;
pJoin->isSingleTableJoin = pJoinTable->table.singleTable;
pJoin->inputTsOrder = ORDER_ASC;
pJoin->node.inputTsOrder = ORDER_ASC;
pJoin->node.groupAction = GROUP_ACTION_CLEAR;
pJoin->node.requireDataOrder = DATA_ORDER_LEVEL_GLOBAL;
pJoin->node.resultDataOrder = DATA_ORDER_LEVEL_GLOBAL;
......@@ -741,8 +741,8 @@ static int32_t createWindowLogicNodeFinalize(SLogicPlanContext* pCxt, SSelectStm
pWindow->igExpired = pCxt->pPlanCxt->igExpired;
pWindow->igCheckUpdate = pCxt->pPlanCxt->igCheckUpdate;
}
pWindow->inputTsOrder = ORDER_ASC;
pWindow->outputTsOrder = ORDER_ASC;
pWindow->node.inputTsOrder = ORDER_ASC;
pWindow->node.outputTsOrder = ORDER_ASC;
int32_t code = nodesCollectFuncs(pSelect, SQL_CLAUSE_WINDOW, fmIsWindowClauseFunc, &pWindow->pFuncs);
if (TSDB_CODE_SUCCESS == code) {
......@@ -972,7 +972,7 @@ static int32_t createFillLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
pFill->node.groupAction = getGroupAction(pCxt, pSelect);
pFill->node.requireDataOrder = getRequireDataOrder(true, pSelect);
pFill->node.resultDataOrder = pFill->node.requireDataOrder;
pFill->inputTsOrder = ORDER_ASC;
pFill->node.inputTsOrder = 0;
int32_t code = partFillExprs(pSelect, &pFill->pFillExprs, &pFill->pNotFillExprs);
if (TSDB_CODE_SUCCESS == code) {
......@@ -1045,6 +1045,19 @@ static int32_t createSortLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
if (NULL == pSort->pSortKeys) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
SNode* pNode = NULL;
SOrderByExprNode* firstSortKey = (SOrderByExprNode*)nodesListGetNode(pSort->pSortKeys, 0);
if (firstSortKey->pExpr->type == QUERY_NODE_COLUMN) {
SColumnNode* pCol = (SColumnNode*)firstSortKey->pExpr;
int16_t projIdx = 1;
FOREACH(pNode, pSelect->pProjectionList) {
SExprNode* pExpr = (SExprNode*)pNode;
if (0 == strcmp(pCol->node.aliasName, pExpr->aliasName)) {
pCol->projIdx = projIdx; break;
}
projIdx++;
}
}
}
if (TSDB_CODE_SUCCESS == code) {
......
source/libs/planner/src/planOptimizer.c
浏览文件 @ 6d2ce5cd
......@@ -116,25 +116,33 @@ static EDealRes optRebuildTbanme(SNode** pNode, void* pContext) {
return DEAL_RES_CONTINUE;
}
static void optSetParentOrder(SLogicNode* pNode, EOrder order) {
static void optSetParentOrder(SLogicNode* pNode, EOrder order, SLogicNode* pNodeForcePropagate) {
if (NULL == pNode) {
return;
}
pNode->inputTsOrder = order;
switch (nodeType(pNode)) {
case QUERY_NODE_LOGIC_PLAN_WINDOW:
((SWindowLogicNode*)pNode)->inputTsOrder = order;
// window has a sorting function, and the operator behind it uses its output order
return;
// for those nodes that will change the order, stop propagating
//case QUERY_NODE_LOGIC_PLAN_WINDOW:
case QUERY_NODE_LOGIC_PLAN_JOIN:
((SJoinLogicNode*)pNode)->inputTsOrder = order;
break;
case QUERY_NODE_LOGIC_PLAN_FILL:
((SFillLogicNode*)pNode)->inputTsOrder = order;
case QUERY_NODE_LOGIC_PLAN_AGG:
case QUERY_NODE_LOGIC_PLAN_SORT:
if (pNode == pNodeForcePropagate) {
pNode->outputTsOrder = order;
break;
} else
return;
case QUERY_NODE_LOGIC_PLAN_WINDOW:
// Window output ts order default to be asc, and changed when doing sort by primary key optimization.
// We stop propagate the original order to parents.
// Use window output ts order instead.
order = pNode->outputTsOrder;
break;
default:
pNode->outputTsOrder = order;
break;
}
optSetParentOrder(pNode->pParent, order);
optSetParentOrder(pNode->pParent, order, pNodeForcePropagate);
}
EDealRes scanPathOptHaveNormalColImpl(SNode* pNode, void* pContext) {
......@@ -339,12 +347,12 @@ static void scanPathOptSetScanOrder(EScanOrder scanOrder, SScanLogicNode* pScan)
case SCAN_ORDER_ASC:
pScan->scanSeq[0] = 1;
pScan->scanSeq[1] = 0;
optSetParentOrder(pScan->node.pParent, ORDER_ASC);
optSetParentOrder(pScan->node.pParent, ORDER_ASC, NULL);
break;
case SCAN_ORDER_DESC:
pScan->scanSeq[0] = 0;
pScan->scanSeq[1] = 1;
optSetParentOrder(pScan->node.pParent, ORDER_DESC);
optSetParentOrder(pScan->node.pParent, ORDER_DESC, NULL);
break;
case SCAN_ORDER_BOTH:
pScan->scanSeq[0] = 1;
......@@ -1239,6 +1247,7 @@ static int32_t sortPriKeyOptApply(SOptimizeContext* pCxt, SLogicSubplan* pLogicS
if ((ORDER_DESC == order && pScan->scanSeq[0] > 0) || (ORDER_ASC == order && pScan->scanSeq[1] > 0)) {
TSWAP(pScan->scanSeq[0], pScan->scanSeq[1]);
}
pScan->node.outputTsOrder = order;
if (TSDB_SUPER_TABLE == pScan->tableType) {
pScan->scanType = SCAN_TYPE_TABLE_MERGE;
pScan->node.resultDataOrder = DATA_ORDER_LEVEL_GLOBAL;
......@@ -1246,9 +1255,9 @@ static int32_t sortPriKeyOptApply(SOptimizeContext* pCxt, SLogicSubplan* pLogicS
}
pScan->sortPrimaryKey = true;
} else if (QUERY_NODE_LOGIC_PLAN_WINDOW == nodeType(pSequencingNode)) {
((SWindowLogicNode*)pSequencingNode)->outputTsOrder = order;
((SLogicNode*)pSequencingNode)->outputTsOrder = order;
}
optSetParentOrder(((SLogicNode*)pSequencingNode)->pParent, order);
optSetParentOrder(((SLogicNode*)pSequencingNode)->pParent, order, (SLogicNode*)pSort);
}
SLogicNode* pChild = (SLogicNode*)nodesListGetNode(pSort->node.pChildren, 0);
......@@ -2881,10 +2890,62 @@ static int32_t tableCountScanOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLo
return code;
}
static SSortLogicNode* sortNonPriKeySatisfied(SLogicNode* pNode) {
if (QUERY_NODE_LOGIC_PLAN_SORT != nodeType(pNode)) {
return NULL;
}
SSortLogicNode* pSort = (SSortLogicNode*)pNode;
if (sortPriKeyOptIsPriKeyOrderBy(pSort->pSortKeys)) {
return NULL;
}
SNode* pSortKeyNode = NULL, *pSortKeyExpr = NULL;
FOREACH(pSortKeyNode, pSort->pSortKeys) {
pSortKeyExpr = ((SOrderByExprNode*)pSortKeyNode)->pExpr;
switch (nodeType(pSortKeyExpr)) {
case QUERY_NODE_COLUMN:
break;
case QUERY_NODE_VALUE:
continue;
default:
return NULL;
}
}
if (!pSortKeyExpr || ((SColumnNode*)pSortKeyExpr)->projIdx != 1 ||
((SColumnNode*)pSortKeyExpr)->node.resType.type != TSDB_DATA_TYPE_TIMESTAMP) {
return NULL;
}
return pSort;
}
static bool sortNonPriKeyShouldOptimize(SLogicNode* pNode, void* pInfo) {
SSortLogicNode* pSort = sortNonPriKeySatisfied(pNode);
if (!pSort) return false;
SNodeList* pSortNodeList = pInfo;
nodesListAppend(pSortNodeList, (SNode*)pSort);
return false;
}
static int32_t sortNonPriKeyOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLogicSubplan) {
SNodeList* pNodeList = nodesMakeList();
optFindEligibleNode(pLogicSubplan->pNode, sortNonPriKeyShouldOptimize, pNodeList);
SNode* pNode = NULL;
FOREACH(pNode, pNodeList) {
SSortLogicNode* pSort = (SSortLogicNode*)pNode;
SOrderByExprNode* pOrderByExpr = (SOrderByExprNode*)nodesListGetNode(pSort->pSortKeys, 0);
pSort->node.outputTsOrder = pOrderByExpr->order;
optSetParentOrder(pSort->node.pParent, pOrderByExpr->order, NULL);
}
pCxt->optimized = false;
nodesClearList(pNodeList);
return TSDB_CODE_SUCCESS;
}
// clang-format off
static const SOptimizeRule optimizeRuleSet[] = {
{.pName = "ScanPath", .optimizeFunc = scanPathOptimize},
{.pName = "PushDownCondition", .optimizeFunc = pushDownCondOptimize},
{.pName = "sortNonPriKeyOptimize", .optimizeFunc = sortNonPriKeyOptimize},
{.pName = "SortPrimaryKey", .optimizeFunc = sortPrimaryKeyOptimize},
{.pName = "SmaIndex", .optimizeFunc = smaIndexOptimize},
{.pName = "PartitionTags", .optimizeFunc = partTagsOptimize},
......
source/libs/planner/src/planPhysiCreater.c
浏览文件 @ 6d2ce5cd
......@@ -366,6 +366,8 @@ static SPhysiNode* makePhysiNode(SPhysiPlanContext* pCxt, SLogicNode* pLogicNode
TSWAP(pPhysiNode->pLimit, pLogicNode->pLimit);
TSWAP(pPhysiNode->pSlimit, pLogicNode->pSlimit);
pPhysiNode->inputTsOrder = pLogicNode->inputTsOrder;
pPhysiNode->outputTsOrder = pLogicNode->outputTsOrder;
int32_t code = createDataBlockDesc(pCxt, pLogicNode->pTargets, &pPhysiNode->pOutputDataBlockDesc);
if (TSDB_CODE_SUCCESS != code) {
......@@ -676,7 +678,7 @@ static int32_t createJoinPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren
int32_t code = TSDB_CODE_SUCCESS;
pJoin->joinType = pJoinLogicNode->joinType;
pJoin->inputTsOrder = pJoinLogicNode->inputTsOrder;
pJoin->node.inputTsOrder = pJoinLogicNode->node.inputTsOrder;
setNodeSlotId(pCxt, pLeftDesc->dataBlockId, pRightDesc->dataBlockId, pJoinLogicNode->pMergeCondition,
&pJoin->pMergeCondition);
if (TSDB_CODE_SUCCESS == code) {
......@@ -939,6 +941,11 @@ static int32_t createIndefRowsFuncPhysiNode(SPhysiPlanContext* pCxt, SNodeList*
SNodeList* pFuncs = NULL;
int32_t code = rewritePrecalcExprs(pCxt, pFuncLogicNode->pFuncs, &pPrecalcExprs, &pFuncs);
if (pIdfRowsFunc->node.inputTsOrder == 0) {
// default to asc
pIdfRowsFunc->node.inputTsOrder = TSDB_ORDER_ASC;
}
SDataBlockDescNode* pChildTupe = (((SPhysiNode*)nodesListGetNode(pChildren, 0))->pOutputDataBlockDesc);
// push down expression to pOutputDataBlockDesc of child node
if (TSDB_CODE_SUCCESS == code && NULL != pPrecalcExprs) {
......@@ -1156,9 +1163,12 @@ static int32_t createWindowPhysiNodeFinalize(SPhysiPlanContext* pCxt, SNodeList*
pWindow->watermark = pWindowLogicNode->watermark;
pWindow->deleteMark = pWindowLogicNode->deleteMark;
pWindow->igExpired = pWindowLogicNode->igExpired;
pWindow->inputTsOrder = pWindowLogicNode->inputTsOrder;
pWindow->outputTsOrder = pWindowLogicNode->outputTsOrder;
pWindow->mergeDataBlock = (GROUP_ACTION_KEEP == pWindowLogicNode->node.groupAction ? false : true);
pWindow->node.inputTsOrder = pWindowLogicNode->node.inputTsOrder;
pWindow->node.outputTsOrder = pWindowLogicNode->node.outputTsOrder;
if (nodeType(pWindow) == QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL) {
pWindow->node.inputTsOrder = pWindowLogicNode->node.outputTsOrder;
}
SNodeList* pPrecalcExprs = NULL;
SNodeList* pFuncs = NULL;
......@@ -1492,7 +1502,7 @@ static int32_t createFillPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren
pFill->mode = pFillNode->mode;
pFill->timeRange = pFillNode->timeRange;
pFill->inputTsOrder = pFillNode->inputTsOrder;
pFill->node.inputTsOrder = pFillNode->node.inputTsOrder;
SDataBlockDescNode* pChildTupe = (((SPhysiNode*)nodesListGetNode(pChildren, 0))->pOutputDataBlockDesc);
int32_t code = setListSlotId(pCxt, pChildTupe->dataBlockId, -1, pFillNode->pFillExprs, &pFill->pFillExprs);
......
source/libs/planner/src/planSpliter.c
浏览文件 @ 6d2ce5cd
......@@ -534,7 +534,9 @@ static int32_t stbSplGetNumOfVgroups(SLogicNode* pNode) {
static int32_t stbSplRewriteFromMergeNode(SMergeLogicNode* pMerge, SLogicNode* pNode) {
int32_t code = TSDB_CODE_SUCCESS;
pMerge->node.inputTsOrder = pNode->outputTsOrder;
pMerge->node.outputTsOrder = pNode->outputTsOrder;
switch (nodeType(pNode)) {
case QUERY_NODE_LOGIC_PLAN_PROJECT: {
SProjectLogicNode *pLogicNode = (SProjectLogicNode*)pNode;
......@@ -631,7 +633,7 @@ static int32_t stbSplSplitIntervalForBatch(SSplitContext* pCxt, SStableSplitInfo
((SWindowLogicNode*)pInfo->pSplitNode)->windowAlgo = INTERVAL_ALGO_MERGE;
SNodeList* pMergeKeys = NULL;
code = stbSplCreateMergeKeysByPrimaryKey(((SWindowLogicNode*)pInfo->pSplitNode)->pTspk,
((SWindowLogicNode*)pInfo->pSplitNode)->outputTsOrder, &pMergeKeys);
((SWindowLogicNode*)pInfo->pSplitNode)->node.outputTsOrder, &pMergeKeys);
if (TSDB_CODE_SUCCESS == code) {
code = stbSplCreateMergeNode(pCxt, NULL, pInfo->pSplitNode, pMergeKeys, pPartWindow, true);
}
......@@ -721,7 +723,7 @@ static int32_t stbSplSplitSessionOrStateForBatch(SSplitContext* pCxt, SStableSpl
SNodeList* pMergeKeys = NULL;
int32_t code = stbSplCreateMergeKeysByPrimaryKey(((SWindowLogicNode*)pWindow)->pTspk,
((SWindowLogicNode*)pWindow)->inputTsOrder, &pMergeKeys);
((SWindowLogicNode*)pWindow)->node.inputTsOrder, &pMergeKeys);
if (TSDB_CODE_SUCCESS == code) {
code = stbSplCreateMergeNode(pCxt, pInfo->pSubplan, pChild, pMergeKeys, (SLogicNode*)pChild, true);
......
source/libs/scheduler/src/schTask.c
浏览文件 @ 6d2ce5cd
......@@ -765,7 +765,7 @@ int32_t schSetTaskCandidateAddrs(SSchJob *pJob, SSchTask *pTask) {
if (SCH_IS_DATA_BIND_TASK(pTask)) {
SCH_TASK_ELOG("no execNode specifed for data src task, numOfEps:%d", pTask->plan->execNode.epSet.numOfEps);
SCH_ERR_RET(TSDB_CODE_APP_ERROR);
SCH_ERR_RET(TSDB_CODE_MND_INVALID_SCHEMA_VER);
}
SCH_ERR_RET(schSetAddrsFromNodeList(pJob, pTask));
......
source/util/src/terror.c
浏览文件 @ 6d2ce5cd
......@@ -261,8 +261,8 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_STB_ALTER_OPTION, "Invalid stable alter
TAOS_DEFINE_ERROR(TSDB_CODE_MND_STB_OPTION_UNCHNAGED, "STable option unchanged")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_FIELD_CONFLICT_WITH_TOPIC,"Field used by topic")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_SINGLE_STB_MODE_DB, "Database is single stable mode")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_SCHEMA_VER, "Invalid schema version while alter stb")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_STABLE_UID_NOT_MATCH, "Invalid stable uid while alter stb")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_SCHEMA_VER, "Invalid schema version")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_STABLE_UID_NOT_MATCH, "Invalid stable uid")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_FIELD_CONFLICT_WITH_TSMA, "Field used by tsma")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_DNODE_IN_CREATING, "Dnode in creating status")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_DNODE_IN_DROPPING, "Dnode in dropping status")
......
tests/parallel_test/cases.task
浏览文件 @ 6d2ce5cd
......@@ -184,6 +184,7 @@
,,y,script,./test.sh -f tsim/query/forceFill.sim
,,y,script,./test.sh -f tsim/query/emptyTsRange.sim
,,y,script,./test.sh -f tsim/query/partitionby.sim
,,y,script,./test.sh -f tsim/query/explain_tsorder.sim
,,y,script,./test.sh -f tsim/qnode/basic1.sim
,,y,script,./test.sh -f tsim/snode/basic1.sim
,,y,script,./test.sh -f tsim/mnode/basic1.sim
......
tests/script/tsim/query/explain_tsorder.sim 0 → 100644
浏览文件 @ 6d2ce5cd
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/exec.sh -n dnode1 -s start
sql connect
sql create database test
sql use test
sql CREATE STABLE `meters` (`ts` TIMESTAMP, `c2` INT) TAGS (`cc` VARCHAR(3))
sql insert into d1 using meters tags("MY") values("2022-05-15 00:01:08.000 ",234)
sql insert into d1 using meters tags("MY") values("2022-05-16 00:01:08.000 ",136)
sql insert into d1 using meters tags("MY") values("2022-05-17 00:01:08.000 ", 59)
sql insert into d1 using meters tags("MY") values("2022-05-18 00:01:08.000 ", 58)
sql insert into d1 using meters tags("MY") values("2022-05-19 00:01:08.000 ",243)
sql insert into d1 using meters tags("MY") values("2022-05-20 00:01:08.000 ",120)
sql insert into d1 using meters tags("MY") values("2022-05-21 00:01:08.000 ", 11)
sql insert into d1 using meters tags("MY") values("2022-05-22 00:01:08.000 ",196)
sql insert into d1 using meters tags("MY") values("2022-05-23 00:01:08.000 ",116)
sql insert into d1 using meters tags("MY") values("2022-05-24 00:01:08.000 ",210)
sql insert into d2 using meters tags("HT") values("2022-05-15 00:01:08.000", 234)
sql insert into d2 using meters tags("HT") values("2022-05-16 00:01:08.000", 136)
sql insert into d2 using meters tags("HT") values("2022-05-17 00:01:08.000", 59)
sql insert into d2 using meters tags("HT") values("2022-05-18 00:01:08.000", 58)
sql insert into d2 using meters tags("HT") values("2022-05-19 00:01:08.000", 243)
sql insert into d2 using meters tags("HT") values("2022-05-20 00:01:08.000", 120)
sql insert into d2 using meters tags("HT") values("2022-05-21 00:01:08.000", 11)
sql insert into d2 using meters tags("HT") values("2022-05-22 00:01:08.000", 196)
sql insert into d2 using meters tags("HT") values("2022-05-23 00:01:08.000", 116)
sql insert into d2 using meters tags("HT") values("2022-05-24 00:01:08.000", 210)
#sleep 10000000
system taos -P7100 -s 'source tsim/query/t/explain_tsorder.sql' | grep -v 'Query OK' | grep -v 'Client Version' > /tmp/explain_tsorder.result
system echo ----------------------diff start-----------------------
system git diff --exit-code --color tsim/query/r/explain_tsorder.result /tmp/explain_tsorder.result
system echo ----------------------diff succeed-----------------------
tests/script/tsim/query/r/explain_tsorder.result 0 → 100644
浏览文件 @ 6d2ce5cd
此差异已折叠。
tests/script/tsim/query/t/explain_tsorder.sql 0 → 100644
浏览文件 @ 6d2ce5cd
use test;
explain verbose true select _wstart, last(ts), avg(c2) from meters interval(10s) order by _wstart desc\G;
explain verbose true select _wstart, last(ts), avg(c2) from meters interval(10s) order by _wstart asc\G;
explain verbose true select _wstart, first(ts), avg(c2) from meters interval(10s) order by _wstart asc\G;
explain verbose true select _wstart, first(ts), avg(c2) from meters interval(10s) order by _wstart desc\G;
explain verbose true select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s)) order by d\G;
explain verbose true select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s)) order by d desc\G;
explain verbose true select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by a) order by d\G;
explain verbose true select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by a desc) order by d\G;
explain verbose true select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by a) order by d desc\G;
explain verbose true select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by a desc) order by d desc\G;
explain verbose true select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by b) order by d\G;
explain verbose true select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by b desc) order by d\G;
explain verbose true select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by b) order by d desc\G;
explain verbose true select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by b desc) order by d desc\G;
explain verbose true select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by b) group by c order by d\G;
explain verbose true select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by b desc) group by c order by d\G;
explain verbose true select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by b) group by c order by d desc\G;
explain verbose true select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by b desc) group by c order by d desc\G;
explain verbose true select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by b) where a > 10000 and a < 20000 interval(10s) fill(NULL) order by d\G;
explain verbose true select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by a desc) where a > 10000 and a < 20000 interval(10s) fill(NULL) order by d\G;
explain verbose true select last(b) as d from (select last(ts) as b, avg(c2) as c from meters interval(10s) order by b desc) where b > 10000 and b < 20000 interval(10s) fill(NULL) order by d\G;
explain verbose true select last(b) as d from (select last(ts) as b, avg(c2) as c from meters interval(10s) order by b desc) where b > 10000 and b < 20000 interval(10s) fill(NULL) order by d desc\G;
select _wstart, last(ts), avg(c2) from meters interval(10s) order by _wstart desc;
select _wstart, last(ts), avg(c2) from meters interval(10s) order by _wstart asc;
select _wstart, first(ts), avg(c2) from meters interval(10s) order by _wstart asc;
select _wstart, first(ts), avg(c2) from meters interval(10s) order by _wstart desc;
select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s)) order by d;
select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s)) order by d desc;
select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by a) order by d;
select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by a desc) order by d;
select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by a) order by d desc;
select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by a desc) order by d desc;
select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by b) order by d;
select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by b desc) order by d;
select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by b) order by d desc;
select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by b desc) order by d desc;
select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by b) group by c order by d
select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by b desc) group by c order by d;
select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by b) group by c order by d desc;
select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by b desc) group by c order by d desc;
select last(a) as d from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by a desc) where a > '2022-05-15 00:01:00.000' and a < '2022-05-19 00:01:08.000' interval(10s) order by d;
select last(b) as d from (select last(ts) as b, avg(c2) as c from meters interval(10s) order by b desc) where b > '2022-05-15 00:01:00.000' and b < '2022-05-19 00:01:08.000' interval(10s) order by d;
select last(b) as d from (select last(ts) as b, avg(c2) as c from meters interval(10s) order by b desc) where b > '2022-05-15 00:01:00.000' and b < '2022-05-19 00:01:08.000' interval(10s) order by d desc;
select _wstart, first(a) as d, avg(c) from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by a desc) where a > '2022-05-15 00:01:00.000' and a < '2022-05-21 00:01:08.000' interval(5h) fill(linear) order by d desc;
explain verbose true select _wstart, first(a) as d, avg(c) from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by a desc) where a > '2022-05-15 00:01:00.000' and a < '2022-05-21 00:01:08.000' interval(5h) fill(linear) order by d desc\G;
explain verbose true select _wstart, first(a) as d, avg(c) from (select _wstart as a, last(ts) as b, avg(c2) as c from meters interval(10s) order by a asc) where a > '2022-05-15 00:01:00.000' and a < '2022-05-21 00:01:08.000' interval(5h) fill(linear) order by d desc\G;
explain verbose true select * from (select ts as a, c2 as b from meters order by c2 desc)\G;
select * from (select ts as a, c2 as b from meters order by c2 desc);
explain verbose true select * from (select ts as a, c2 as b from meters order by c2 desc) order by a desc\G;
select * from (select ts as a, c2 as b from meters order by c2 desc) order by a desc;
explain verbose true select a.ts, a.c2, b.c2 from meters as a join meters as b on a.ts = b.ts\G;
explain verbose true select a.ts, a.c2, b.c2 from meters as a join meters as b on a.ts = b.ts order by a.ts\G;
select a.ts, a.c2, b.c2 from meters as a join meters as b on a.ts = b.ts;
select a.ts, a.c2, b.c2 from meters as a join meters as b on a.ts = b.ts order by a.ts desc;
explain verbose true select a.ts, a.c2, b.c2 from meters as a join (select ts, c2 from meters order by ts desc) b on a.ts = b.ts order by a.ts desc\G;
explain verbose true select a.ts, a.c2, b.c2 from meters as a join (select ts, c2 from meters order by ts desc) b on a.ts = b.ts order by a.ts asc\G;
select a.ts, a.c2, b.c2 from meters as a join (select * from meters order by ts desc) b on a.ts = b.ts order by a.ts asc;
tests/script/tsim/sma/tsmaCreateInsertQuery.sim
浏览文件 @ 6d2ce5cd
......@@ -414,6 +414,21 @@ if $data05 != 30.000000000 then
return -1
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT
sql delete from stb;
print =============== query after delete in common vgroups
sql select _wstart, _wend, min(c1),max(c2),max(c1),max(c3) from stb interval(5m,10s) sliding(5m) order by _wstart;
if $rows != 0 then
print rows $rows != 0
return -1
endi
sleep 2000
print =============== query after delete in designated vgroups
sql select _wend, min(c1),max(c2),max(c1) from stb interval(5m,10s) sliding(5m) order by _wstart;
if $rows != 0 then
print rows $rows != 0
return -1
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT
\ No newline at end of file
utils/tsim/src/simExe.c
浏览文件 @ 6d2ce5cd
......@@ -434,7 +434,7 @@ bool simExecuteSystemCmd(SScript *script, char *option) {
simLogSql(buf, true);
int32_t code = system(buf);
int32_t repeatTimes = 0;
while (code < 0) {
while (code != 0) {
simError("script:%s, failed to execute %s , code %d, errno:%d %s, repeatTimes:%d", script->fileName, buf, code,
errno, strerror(errno), repeatTimes);
taosMsleep(1000);
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或