xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge branch 'enh/tsdb_optimize' of github.com:taosdata/tdengine into enh/tsdb_optimize

This commit is contained in:
Haojun Liao 2023-06-19 17:57:00 +08:00
parent 524ef6edda 6d2ce5cd39
commit 77af819708
55 changed files with 4618 additions and 1588 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
docs/en/28-releases/01-tdengine.md
View File

@ -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" />

4
docs/en/28-releases/02-tools.md
View File

@ -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" />

4
docs/zh/28-releases/01-tdengine.md
View File

@ -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" />

4
docs/zh/28-releases/02-tools.md
View File

@ -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" />

1
include/common/tcommon.h
View File

@ -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

12
include/libs/nodes/plannodes.h
View File

@ -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;

1
include/libs/nodes/querynodes.h
View File

@ -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];

3
include/libs/qcom/query.h
View File

@ -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))

1
source/client/src/clientMain.c
View File

@ -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;

4
source/common/src/tdatablock.c
View File

@ -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);
}

1
source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
View File

@ -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);

46
source/dnode/vnode/src/sma/smaTimeRange.c
View File

@ -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;

120
source/dnode/vnode/src/tsdb/tsdbCommit2.c
View File

@ -21,8 +21,8 @@ typedef struct {
TFileSetArray *fsetArr;
TFileOpArray fopArray[1];
SSkmInfo skmTb[1];
SSkmInfo skmRow[1];
// SSkmInfo skmTb[1];
// SSkmInfo skmRow[1];
int32_t minutes;
int8_t precision;
@ -56,48 +56,29 @@ typedef struct {
SIterMerger *tombIterMerger;
// writer
SBlockData blockData[2];
int32_t blockDataIdx;
SDataFileWriter *dataWriter;
SSttFileWriter *sttWriter;
SFSetWriter *writer;
} SCommitter2;
static int32_t tsdbCommitOpenWriter(SCommitter2 *committer) {
int32_t code = 0;
int32_t lino = 0;
// stt writer
SSttFileWriterConfig config[1] = {{
SFSetWriterConfig config = {
.tsdb = committer->tsdb,
.toSttOnly = true,
.compactVersion = committer->compactVersion,
.minRow = committer->minRow,
.maxRow = committer->maxRow,
.szPage = committer->szPage,
.cmprAlg = committer->cmprAlg,
.compactVersion = committer->compactVersion,
.file =
{
.type = TSDB_FTYPE_STT,
.did = committer->ctx->did,
.fid = committer->ctx->fid,
.cid = committer->ctx->cid,
},
}};
.fid = committer->ctx->fid,
.cid = committer->ctx->cid,
.did = committer->ctx->did,
.level = 0,
};
code = tsdbSttFileWriterOpen(config, &committer->sttWriter);
TSDB_CHECK_CODE(code, lino, _exit);
// data writer
if (committer->sttTrigger == 1) {
// data writer
SDataFileWriterConfig config = {
.tsdb = committer->tsdb,
.cmprAlg = committer->cmprAlg,
.maxRow = committer->maxRow,
.szPage = committer->szPage,
.fid = committer->ctx->fid,
.cid = committer->ctx->cid,
.did = committer->ctx->did,
.compactVersion = committer->compactVersion,
};
config.toSttOnly = false;
if (committer->ctx->fset) {
for (int32_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ftype++) {
@ -107,11 +88,11 @@ static int32_t tsdbCommitOpenWriter(SCommitter2 *committer) {
}
}
}
code = tsdbDataFileWriterOpen(&config, &committer->dataWriter);
TSDB_CHECK_CODE(code, lino, _exit);
}
code = tsdbFSetWriterOpen(&config, &committer->writer);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(committer->tsdb->pVnode), lino, code);
@ -120,22 +101,10 @@ _exit:
}
static int32_t tsdbCommitCloseWriter(SCommitter2 *committer) {
int32_t code = 0;
int32_t lino = 0;
code = tsdbSttFileWriterClose(&committer->sttWriter, 0, committer->fopArray);
TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbDataFileWriterClose(&committer->dataWriter, 0, committer->fopArray);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(committer->tsdb->pVnode), lino, code);
}
return code;
return tsdbFSetWriterClose(&committer->writer, 0, committer->fopArray);
}
#if 0
static int32_t tsdbCommitTSDataToDataTableBegin(SCommitter2 *committer, const TABLEID *tbid) {
int32_t code = 0;
int32_t lino = 0;
@ -331,17 +300,30 @@ _exit:
}
return code;
}
#endif
static int32_t tsdbCommitTSData(SCommitter2 *committer) {
int32_t code = 0;
int32_t lino = 0;
int32_t code = 0;
int32_t lino = 0;
SMetaInfo info;
// loop iter
if (committer->sttTrigger == 1) {
code = tsdbCommitTSDataToData(committer);
for (SRowInfo *row; (row = tsdbIterMergerGetData(committer->dataIterMerger)) != NULL;) {
if (row->uid != committer->ctx->tbid->uid) {
// Ignore table of obsolescence
if (metaGetInfo(committer->tsdb->pVnode->pMeta, row->uid, &info, NULL) != 0) {
code = tsdbIterMergerSkipTableData(committer->dataIterMerger, (TABLEID *)row);
TSDB_CHECK_CODE(code, lino, _exit);
continue;
}
committer->ctx->tbid->suid = row->suid;
committer->ctx->tbid->uid = row->uid;
}
code = tsdbFSetWriteRow(committer->writer, row);
TSDB_CHECK_CODE(code, lino, _exit);
} else {
code = tsdbCommitTSDataToStt(committer);
code = tsdbIterMergerNext(committer->dataIterMerger);
TSDB_CHECK_CODE(code, lino, _exit);
}
@ -356,22 +338,12 @@ static int32_t tsdbCommitTombData(SCommitter2 *committer) {
int32_t code = 0;
int32_t lino = 0;
if (committer->dataWriter == NULL || tsdbSttFileWriterIsOpened(committer->sttWriter)) {
for (STombRecord *record; (record = tsdbIterMergerGetTombRecord(committer->tombIterMerger));) {
code = tsdbSttFileWriteTombRecord(committer->sttWriter, record);
TSDB_CHECK_CODE(code, lino, _exit);
for (STombRecord *record; (record = tsdbIterMergerGetTombRecord(committer->tombIterMerger));) {
code = tsdbFSetWriteTombRecord(committer->writer, record);
TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbIterMergerNext(committer->tombIterMerger);
TSDB_CHECK_CODE(code, lino, _exit);
}
} else {
for (STombRecord *record; (record = tsdbIterMergerGetTombRecord(committer->tombIterMerger));) {
code = tsdbDataFileWriteTombRecord(committer->dataWriter, record);
TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbIterMergerNext(committer->tombIterMerger);
TSDB_CHECK_CODE(code, lino, _exit);
}
code = tsdbIterMergerNext(committer->tombIterMerger);
TSDB_CHECK_CODE(code, lino, _exit);
}
_exit:
@ -532,8 +504,7 @@ static int32_t tsdbCommitFileSetBegin(SCommitter2 *committer) {
ASSERT(TARRAY2_SIZE(committer->dataIterArray) == 0);
ASSERT(committer->dataIterMerger == NULL);
ASSERT(committer->sttWriter == NULL);
ASSERT(committer->dataWriter == NULL);
ASSERT(committer->writer == NULL);
code = tsdbCommitOpenReader(committer);
TSDB_CHECK_CODE(code, lino, _exit);
@ -663,8 +634,7 @@ static int32_t tsdbCloseCommitter(SCommitter2 *committer, int32_t eno) {
ASSERT(0);
}
ASSERT(committer->dataWriter == NULL);
ASSERT(committer->sttWriter == NULL);
ASSERT(committer->writer == NULL);
ASSERT(committer->dataIterMerger == NULL);
ASSERT(committer->tombIterMerger == NULL);
TARRAY2_DESTROY(committer->dataIterArray, NULL);

1
source/dnode/vnode/src/tsdb/tsdbCommit2.h
View File

@ -15,6 +15,7 @@
#include "tsdbDataFileRW.h"
#include "tsdbFS2.h"
#include "tsdbFSetRW.h"
#include "tsdbIter.h"
#include "tsdbSttFileRW.h"

292
source/dnode/vnode/src/tsdb/tsdbFSetRW.c Normal file
View File

@ -0,0 +1,292 @@
/*
* 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 "tsdbFSetRW.h"
// SFSetWriter ==================================================
struct SFSetWriter {
SFSetWriterConfig config[1];
SSkmInfo skmTb[1];
SSkmInfo skmRow[1];
uint8_t *bufArr[10];
struct {
TABLEID tbid[1];
} ctx[1];
// writer
SBlockData blockData[2];
int32_t blockDataIdx;
SDataFileWriter *dataWriter;
SSttFileWriter *sttWriter;
};
static int32_t tsdbFSetWriteTableDataBegin(SFSetWriter *writer, const TABLEID *tbid) {
int32_t code = 0;
int32_t lino = 0;
writer->ctx->tbid->suid = tbid->suid;
writer->ctx->tbid->uid = tbid->uid;
code = tsdbUpdateSkmTb(writer->config->tsdb, writer->ctx->tbid, writer->skmTb);
TSDB_CHECK_CODE(code, lino, _exit);
writer->blockDataIdx = 0;
for (int32_t i = 0; i < ARRAY_SIZE(writer->blockData); i++) {
code = tBlockDataInit(&writer->blockData[i], writer->ctx->tbid, writer->skmTb->pTSchema, NULL, 0);
TSDB_CHECK_CODE(code, lino, _exit);
}
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(writer->config->tsdb->pVnode), lino, code);
}
return code;
}
static int32_t tsdbFSetWriteTableDataEnd(SFSetWriter *writer) {
if (writer->ctx->tbid->uid == 0) return 0;
int32_t code = 0;
int32_t lino = 0;
int32_t cidx = writer->blockDataIdx;
int32_t pidx = ((cidx + 1) & 1);
int32_t numRow = ((writer->blockData[pidx].nRow + writer->blockData[cidx].nRow) >> 1);
if (writer->blockData[pidx].nRow > 0 && numRow >= writer->config->minRow) {
ASSERT(writer->blockData[pidx].nRow == writer->config->maxRow);
SRowInfo row = {
.suid = writer->ctx->tbid->suid,
.uid = writer->ctx->tbid->uid,
.row = tsdbRowFromBlockData(writer->blockData + pidx, 0),
};
for (int32_t i = 0; i < numRow; i++) {
row.row.iRow = i;
code = tsdbDataFileWriteRow(writer->dataWriter, &row);
TSDB_CHECK_CODE(code, lino, _exit);
}
code = tsdbDataFileFlush(writer->dataWriter);
TSDB_CHECK_CODE(code, lino, _exit);
for (int32_t i = numRow; i < writer->blockData[pidx].nRow; i++) {
row.row.iRow = i;
code = tsdbDataFileWriteRow(writer->dataWriter, &row);
TSDB_CHECK_CODE(code, lino, _exit);
}
row.row = tsdbRowFromBlockData(writer->blockData + cidx, 0);
for (int32_t i = 0; i < writer->blockData[cidx].nRow; i++) {
row.row.iRow = i;
code = tsdbDataFileWriteRow(writer->dataWriter, &row);
TSDB_CHECK_CODE(code, lino, _exit);
}
} else {
// pidx
if (writer->blockData[pidx].nRow > 0) {
code = tsdbDataFileWriteBlockData(writer->dataWriter, &writer->blockData[pidx]);
TSDB_CHECK_CODE(code, lino, _exit);
}
// cidx
if (writer->blockData[cidx].nRow < writer->config->minRow) {
code = tsdbSttFileWriteBlockData(writer->sttWriter, &writer->blockData[cidx]);
TSDB_CHECK_CODE(code, lino, _exit);
} else {
code = tsdbDataFileWriteBlockData(writer->dataWriter, &writer->blockData[cidx]);
TSDB_CHECK_CODE(code, lino, _exit);
}
}
for (int32_t i = 0; i < ARRAY_SIZE(writer->blockData); i++) {
tBlockDataReset(&writer->blockData[i]);
}
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(writer->config->tsdb->pVnode), lino, code);
}
return code;
}
int32_t tsdbFSetWriterOpen(SFSetWriterConfig *config, SFSetWriter **writer) {
int32_t code = 0;
int32_t lino = 0;
writer[0] = taosMemoryCalloc(1, sizeof(*writer[0]));
if (writer[0] == NULL) return TSDB_CODE_OUT_OF_MEMORY;
writer[0]->config[0] = config[0];
// data writer
if (!config->toSttOnly) {
SDataFileWriterConfig dataWriterConfig = {
.tsdb = config->tsdb,
.cmprAlg = config->cmprAlg,
.maxRow = config->maxRow,
.szPage = config->szPage,
.fid = config->fid,
.cid = config->cid,
.did = config->did,
.compactVersion = config->compactVersion,
.skmTb = writer[0]->skmTb,
.skmRow = writer[0]->skmRow,
.bufArr = writer[0]->bufArr,
};
for (int32_t ftype = 0; ftype < TSDB_FTYPE_MAX; ++ftype) {
dataWriterConfig.files[ftype].exist = config->files[ftype].exist;
dataWriterConfig.files[ftype].file = config->files[ftype].file;
}
code = tsdbDataFileWriterOpen(&dataWriterConfig, &writer[0]->dataWriter);
TSDB_CHECK_CODE(code, lino, _exit);
}
// stt writer
SSttFileWriterConfig sttWriterConfig = {
.tsdb = config->tsdb,
.maxRow = config->maxRow,
.szPage = config->szPage,
.cmprAlg = config->cmprAlg,
.compactVersion = config->compactVersion,
.did = config->did,
.fid = config->fid,
.cid = config->cid,
.level = config->level,
.skmTb = writer[0]->skmTb,
.skmRow = writer[0]->skmRow,
.bufArr = writer[0]->bufArr,
};
code = tsdbSttFileWriterOpen(&sttWriterConfig, &writer[0]->sttWriter);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(config->tsdb->pVnode), lino, code);
}
return code;
}
int32_t tsdbFSetWriterClose(SFSetWriter **writer, bool abort, TFileOpArray *fopArr) {
if (writer[0] == NULL) return 0;
int32_t code = 0;
int32_t lino = 0;
STsdb *tsdb = writer[0]->config->tsdb;
// end
if (!writer[0]->config->toSttOnly) {
code = tsdbDataFileWriterClose(&writer[0]->dataWriter, abort, fopArr);
TSDB_CHECK_CODE(code, lino, _exit);
}
code = tsdbSttFileWriterClose(&writer[0]->sttWriter, abort, fopArr);
TSDB_CHECK_CODE(code, lino, _exit);
// free
for (int32_t i = 0; i < ARRAY_SIZE(writer[0]->blockData); i++) {
tBlockDataDestroy(&writer[0]->blockData[i]);
}
for (int32_t i = 0; i < ARRAY_SIZE(writer[0]->bufArr); i++) {
tFree(writer[0]->bufArr[i]);
}
tDestroyTSchema(writer[0]->skmRow->pTSchema);
tDestroyTSchema(writer[0]->skmTb->pTSchema);
taosMemoryFree(writer[0]);
writer[0] = NULL;
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(tsdb->pVnode), lino, code);
}
return code;
}
int32_t tsdbFSetWriteRow(SFSetWriter *writer, SRowInfo *row) {
int32_t code = 0;
int32_t lino = 0;
if (writer->config->toSttOnly) {
code = tsdbSttFileWriteRow(writer->sttWriter, row);
TSDB_CHECK_CODE(code, lino, _exit);
} else {
if (writer->ctx->tbid->uid != row->uid) {
code = tsdbFSetWriteTableDataEnd(writer);
TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbFSetWriteTableDataBegin(writer, (TABLEID *)row);
TSDB_CHECK_CODE(code, lino, _exit);
}
if (row->row.type == TSDBROW_ROW_FMT) {
code = tsdbUpdateSkmRow(writer->config->tsdb, writer->ctx->tbid, TSDBROW_SVERSION(&row->row), writer->skmRow);
TSDB_CHECK_CODE(code, lino, _exit);
}
TSDBKEY key = TSDBROW_KEY(&row->row);
if (key.version <= writer->config->compactVersion //
&& writer->blockData[writer->blockDataIdx].nRow > 0 //
&& key.ts == writer->blockData[writer->blockDataIdx].aTSKEY[writer->blockData[writer->blockDataIdx].nRow - 1]) {
code = tBlockDataUpdateRow(&writer->blockData[writer->blockDataIdx], &row->row, writer->skmRow->pTSchema);
TSDB_CHECK_CODE(code, lino, _exit);
} else {
if (writer->blockData[writer->blockDataIdx].nRow >= writer->config->maxRow) {
int32_t idx = ((writer->blockDataIdx + 1) & 1);
if (writer->blockData[idx].nRow >= writer->config->maxRow) {
code = tsdbDataFileWriteBlockData(writer->dataWriter, &writer->blockData[idx]);
TSDB_CHECK_CODE(code, lino, _exit);
tBlockDataClear(&writer->blockData[idx]);
}
writer->blockDataIdx = idx;
}
code =
tBlockDataAppendRow(&writer->blockData[writer->blockDataIdx], &row->row, writer->skmRow->pTSchema, row->uid);
TSDB_CHECK_CODE(code, lino, _exit);
}
}
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(writer->config->tsdb->pVnode), lino, code);
}
return code;
}
int32_t tsdbFSetWriteTombRecord(SFSetWriter *writer, const STombRecord *tombRecord) {
int32_t code = 0;
int32_t lino = 0;
if (writer->config->toSttOnly || tsdbSttFileWriterIsOpened(writer->sttWriter)) {
code = tsdbSttFileWriteTombRecord(writer->sttWriter, tombRecord);
TSDB_CHECK_CODE(code, lino, _exit);
} else {
code = tsdbDataFileWriteTombRecord(writer->dataWriter, tombRecord);
TSDB_CHECK_CODE(code, lino, _exit);
}
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(writer->config->tsdb->pVnode), lino, code);
}
return code;
}

55
source/dnode/vnode/src/tsdb/tsdbFSetRW.h Normal file
View File

@ -0,0 +1,55 @@
/*
* 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 "tsdbDataFileRW.h"
#include "tsdbSttFileRW.h"
#ifndef _TSDB_FSET_RW_H
#define _TSDB_FSET_RW_H
#ifdef __cplusplus
extern "C" {
#endif
//
typedef struct SFSetWriter SFSetWriter;
typedef struct {
STsdb *tsdb;
bool toSttOnly;
int64_t compactVersion;
int32_t minRow;
int32_t maxRow;
int32_t szPage;
int8_t cmprAlg;
int32_t fid;
int64_t cid;
SDiskID did;
int32_t level;
struct {
bool exist;
STFile file;
} files[TSDB_FTYPE_MAX];
} SFSetWriterConfig;
int32_t tsdbFSetWriterOpen(SFSetWriterConfig *config, SFSetWriter **writer);
int32_t tsdbFSetWriterClose(SFSetWriter **writer, bool abort, TFileOpArray *fopArr);
int32_t tsdbFSetWriteRow(SFSetWriter *writer, SRowInfo *row);
int32_t tsdbFSetWriteTombRecord(SFSetWriter *writer, const STombRecord *tombRecord);
#ifdef __cplusplus
}
#endif
#endif /*_TSDB_FSET_RW_H*/

168
source/dnode/vnode/src/tsdb/tsdbMerge.c
View File

@ -19,15 +19,13 @@ typedef struct {
STsdb *tsdb;
TFileSetArray *fsetArr;
int32_t sttTrigger;
int32_t maxRow;
int32_t minRow;
int32_t szPage;
int8_t cmprAlg;
int64_t compactVersion;
int64_t cid;
SSkmInfo skmTb[1];
SSkmInfo skmRow[1];
int32_t sttTrigger;
int32_t maxRow;
int32_t minRow;
int32_t szPage;
int8_t cmprAlg;
int64_t compactVersion;
int64_t cid;
// context
struct {
@ -37,10 +35,7 @@ typedef struct {
bool toData;
int32_t level;
SSttLvl *lvl;
// STFileObj *fobj;
TABLEID tbid[1];
int32_t blockDataIdx;
SBlockData blockData[2];
} ctx[1];
TFileOpArray fopArr[1];
@ -53,8 +48,7 @@ typedef struct {
TTsdbIterArray tombIterArr[1];
SIterMerger *tombIterMerger;
// writer
SSttFileWriter *sttWriter;
SDataFileWriter *dataWriter;
SFSetWriter *writer;
} SMerger;
static int32_t tsdbMergerOpen(SMerger *merger) {
@ -86,8 +80,7 @@ static int32_t tsdbMergerClose(SMerger *merger) {
}
taosThreadRwlockUnlock(&merger->tsdb->rwLock);
ASSERT(merger->dataWriter == NULL);
ASSERT(merger->sttWriter == NULL);
ASSERT(merger->writer == NULL);
ASSERT(merger->dataIterMerger == NULL);
ASSERT(TARRAY2_SIZE(merger->dataIterArr) == 0);
ASSERT(TARRAY2_SIZE(merger->sttReaderArr) == 0);
@ -96,11 +89,6 @@ static int32_t tsdbMergerClose(SMerger *merger) {
TARRAY2_DESTROY(merger->dataIterArr, NULL);
TARRAY2_DESTROY(merger->sttReaderArr, NULL);
TARRAY2_DESTROY(merger->fopArr, NULL);
for (int32_t i = 0; i < ARRAY_SIZE(merger->ctx->blockData); i++) {
tBlockDataDestroy(merger->ctx->blockData + i);
}
tDestroyTSchema(merger->skmTb->pTSchema);
tDestroyTSchema(merger->skmRow->pTSchema);
_exit:
if (code) {
@ -109,6 +97,7 @@ _exit:
return code;
}
#if 0
static int32_t tsdbMergeToDataTableEnd(SMerger *merger) {
if (merger->ctx->blockData[0].nRow + merger->ctx->blockData[1].nRow == 0) return 0;
@ -297,6 +286,7 @@ _exit:
}
return code;
}
#endif
static int32_t tsdbMergeFileSetBeginOpenReader(SMerger *merger) {
int32_t code = 0;
@ -413,56 +403,36 @@ static int32_t tsdbMergeFileSetBeginOpenWriter(SMerger *merger) {
code = TSDB_CODE_FS_NO_VALID_DISK;
TSDB_CHECK_CODE(code, lino, _exit);
}
{
// to new level
SSttFileWriterConfig config[1] = {{
.tsdb = merger->tsdb,
.maxRow = merger->maxRow,
.szPage = merger->szPage,
.cmprAlg = merger->cmprAlg,
.compactVersion = merger->compactVersion,
.file =
{
.type = TSDB_FTYPE_STT,
.did = did,
.fid = merger->ctx->fset->fid,
.cid = merger->cid,
.size = 0,
.stt = {{
.level = merger->ctx->level,
}},
},
}};
code = tsdbSttFileWriterOpen(config, &merger->sttWriter);
TSDB_CHECK_CODE(code, lino, _exit);
}
SFSetWriterConfig config = {
.tsdb = merger->tsdb,
.toSttOnly = true,
.compactVersion = merger->compactVersion,
.minRow = merger->minRow,
.maxRow = merger->maxRow,
.szPage = merger->szPage,
.cmprAlg = merger->cmprAlg,
.fid = merger->ctx->fset->fid,
.cid = merger->cid,
.did = did,
.level = merger->ctx->level,
};
if (merger->ctx->toData) {
SDataFileWriterConfig config[1] = {{
.tsdb = merger->tsdb,
.cmprAlg = merger->cmprAlg,
.maxRow = merger->maxRow,
.szPage = merger->szPage,
.fid = merger->ctx->fset->fid,
.cid = merger->cid,
.did = did,
.compactVersion = merger->compactVersion,
}};
config.toSttOnly = false;
for (int32_t i = 0; i < TSDB_FTYPE_MAX; i++) {
if (merger->ctx->fset->farr[i]) {
config->files[i].exist = true;
config->files[i].file = merger->ctx->fset->farr[i]->f[0];
for (int32_t ftype = 0; ftype < TSDB_FTYPE_MAX; ++ftype) {
if (merger->ctx->fset->farr[ftype]) {
config.files[ftype].exist = true;
config.files[ftype].file = merger->ctx->fset->farr[ftype]->f[0];
} else {
config->files[i].exist = false;
config.files[ftype].exist = false;
}
}
code = tsdbDataFileWriterOpen(config, &merger->dataWriter);
TSDB_CHECK_CODE(code, lino, _exit);
}
code = tsdbFSetWriterOpen(&config, &merger->writer);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
TSDB_ERROR_LOG(vid, lino, code);
@ -477,15 +447,10 @@ static int32_t tsdbMergeFileSetBegin(SMerger *merger) {
ASSERT(TARRAY2_SIZE(merger->sttReaderArr) == 0);
ASSERT(TARRAY2_SIZE(merger->dataIterArr) == 0);
ASSERT(merger->dataIterMerger == NULL);
ASSERT(merger->sttWriter == NULL);
ASSERT(merger->dataWriter == NULL);
ASSERT(merger->writer == NULL);
merger->ctx->tbid->suid = 0;
merger->ctx->tbid->uid = 0;
merger->ctx->blockDataIdx = 0;
for (int32_t i = 0; i < ARRAY_SIZE(merger->ctx->blockData); ++i) {
tBlockDataReset(merger->ctx->blockData + i);
}
// open reader
code = tsdbMergeFileSetBeginOpenReader(merger);
@ -507,23 +472,7 @@ _exit:
}
static int32_t tsdbMergeFileSetEndCloseWriter(SMerger *merger) {
int32_t code = 0;
int32_t lino = 0;
int32_t vid = TD_VID(merger->tsdb->pVnode);
code = tsdbSttFileWriterClose(&merger->sttWriter, 0, merger->fopArr);
TSDB_CHECK_CODE(code, lino, _exit);
if (merger->ctx->toData) {
code = tsdbDataFileWriterClose(&merger->dataWriter, 0, merger->fopArr);
TSDB_CHECK_CODE(code, lino, _exit);
}
_exit:
if (code) {
TSDB_ERROR_LOG(vid, lino, code);
}
return code;
return tsdbFSetWriterClose(&merger->writer, 0, merger->fopArr);
}
static int32_t tsdbMergeFileSetEndCloseIter(SMerger *merger) {
@ -567,12 +516,49 @@ static int32_t tsdbMergeFileSet(SMerger *merger, STFileSet *fset) {
code = tsdbMergeFileSetBegin(merger);
TSDB_CHECK_CODE(code, lino, _exit);
// do merge
if (merger->ctx->toData) {
code = tsdbMergeToDataLevel(merger);
// data
SMetaInfo info;
SRowInfo *row;
merger->ctx->tbid->suid = 0;
merger->ctx->tbid->uid = 0;
while ((row = tsdbIterMergerGetData(merger->dataIterMerger)) != NULL) {
if (row->uid != merger->ctx->tbid->uid) {
if (metaGetInfo(merger->tsdb->pVnode->pMeta, row->uid, &info, NULL) != 0) {
code = tsdbIterMergerSkipTableData(merger->dataIterMerger, (TABLEID *)row);
TSDB_CHECK_CODE(code, lino, _exit);
continue;
}
merger->ctx->tbid->uid = row->uid;
merger->ctx->tbid->suid = row->suid;
}
code = tsdbFSetWriteRow(merger->writer, row);
TSDB_CHECK_CODE(code, lino, _exit);
} else {
code = tsdbMergeToUpperLevel(merger);
code = tsdbIterMergerNext(merger->dataIterMerger);
TSDB_CHECK_CODE(code, lino, _exit);
}
// tomb
STombRecord *record;
merger->ctx->tbid->suid = 0;
merger->ctx->tbid->uid = 0;
while ((record = tsdbIterMergerGetTombRecord(merger->tombIterMerger)) != NULL) {
if (record->uid != merger->ctx->tbid->uid) {
merger->ctx->tbid->uid = record->uid;
merger->ctx->tbid->suid = record->suid;
if (metaGetInfo(merger->tsdb->pVnode->pMeta, record->uid, &info, NULL) != 0) {
code = tsdbIterMergerSkipTableData(merger->tombIterMerger, merger->ctx->tbid);
TSDB_CHECK_CODE(code, lino, _exit);
continue;
}
}
code = tsdbFSetWriteTombRecord(merger->writer, record);
TSDB_CHECK_CODE(code, lino, _exit);
code = tsdbIterMergerNext(merger->tombIterMerger);
TSDB_CHECK_CODE(code, lino, _exit);
}

1
source/dnode/vnode/src/tsdb/tsdbMerge.h
View File

@ -15,6 +15,7 @@
#include "tsdbDataFileRW.h"
#include "tsdbFS2.h"
#include "tsdbFSetRW.h"
#include "tsdbIter.h"
#include "tsdbSttFileRW.h"
#include "tsdbUtil2.h"

237
source/dnode/vnode/src/tsdb/tsdbRetention.c
View File

@ -14,6 +14,7 @@
*/
#include "tsdb.h"
#include "tsdbFS2.h"
static bool tsdbShouldDoRetentionImpl(STsdb *pTsdb, int64_t now) {
for (int32_t iSet = 0; iSet < taosArrayGetSize(pTsdb->fs.aDFileSet); iSet++) {
@ -111,4 +112,238 @@ int32_t tsdbCommitRetention(STsdb *pTsdb) {
taosThreadRwlockUnlock(&pTsdb->rwLock);
tsdbInfo("vgId:%d %s done", TD_VID(pTsdb->pVnode), __func__);
return 0;
}
}
// new ==============
typedef struct {
STsdb *tsdb;
int32_t szPage;
int64_t now;
int64_t cid;
TFileSetArray *fsetArr;
TFileOpArray *fopArr;
struct {
int32_t fsetArrIdx;
STFileSet *fset;
} ctx[1];
} SRTXer;
static int32_t tsdbDoRemoveFileObject(SRTXer *rtxer, const STFileObj *fobj) {
STFileOp op = {
.optype = TSDB_FOP_REMOVE,
.fid = fobj->f->fid,
.of = fobj->f[0],
};
return TARRAY2_APPEND(rtxer->fopArr, op);
}
static int32_t tsdbDoCopyFile(SRTXer *rtxer, const STFileObj *from, const STFile *to) {
int32_t code = 0;
int32_t lino = 0;
char fname[TSDB_FILENAME_LEN];
TdFilePtr fdFrom = NULL;
TdFilePtr fdTo = NULL;
tsdbTFileName(rtxer->tsdb, to, fname);
fdFrom = taosOpenFile(from->fname, TD_FILE_READ);
if (fdFrom == NULL) code = terrno;
TSDB_CHECK_CODE(code, lino, _exit);
fdTo = taosOpenFile(fname, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
if (fdTo == NULL) code = terrno;
TSDB_CHECK_CODE(code, lino, _exit);
int64_t n = taosFSendFile(fdTo, fdFrom, 0, tsdbLogicToFileSize(from->f->size, rtxer->szPage));
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
TSDB_CHECK_CODE(code, lino, _exit);
}
taosCloseFile(&fdFrom);
taosCloseFile(&fdTo);
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(rtxer->tsdb->pVnode), lino, code);
taosCloseFile(&fdFrom);
taosCloseFile(&fdTo);
}
return code;
}
static int32_t tsdbDoMigrateFileObj(SRTXer *rtxer, const STFileObj *fobj, const SDiskID *did) {
int32_t code = 0;
int32_t lino = 0;
STFileOp op = {0};
// remove old
op = (STFileOp){
.optype = TSDB_FOP_REMOVE,
.fid = fobj->f->fid,
.of = fobj->f[0],
};
code = TARRAY2_APPEND(rtxer->fopArr, op);
TSDB_CHECK_CODE(code, lino, _exit);
// create new
op = (STFileOp){
.optype = TSDB_FOP_CREATE,
.fid = fobj->f->fid,
.nf =
{
.type = fobj->f->type,
.did = did[0],
.fid = fobj->f->fid,
.cid = rtxer->cid,
.size = fobj->f->size,
.stt[0] =
{
.level = fobj->f->stt[0].level,
},
},
};
code = TARRAY2_APPEND(rtxer->fopArr, op);
TSDB_CHECK_CODE(code, lino, _exit);
// do copy the file
code = tsdbDoCopyFile(rtxer, fobj, &op.nf);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(rtxer->tsdb->pVnode), lino, code);
}
return code;
}
static int32_t tsdbDoRetentionBegin(STsdb *tsdb, SRTXer *rtxer) {
int32_t code = 0;
int32_t lino = 0;
// TODO: wait for merge and compact task done
rtxer->tsdb = tsdb;
rtxer->szPage = tsdb->pVnode->config.tsdbPageSize;
rtxer->now = taosGetTimestampMs();
rtxer->cid = tsdbFSAllocEid(tsdb->pFS);
code = tsdbFSCreateCopySnapshot(tsdb->pFS, &rtxer->fsetArr);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(rtxer->tsdb->pVnode), lino, code);
} else {
tsdbInfo("vid:%d, cid:%" PRId64 ", %s done", TD_VID(rtxer->tsdb->pVnode), rtxer->cid, __func__);
}
return code;
}
static int32_t tsdbDoRetentionEnd(SRTXer *rtxer) {
int32_t code = 0;
int32_t lino = 0;
if (TARRAY2_SIZE(rtxer->fopArr) == 0) goto _exit;
code = tsdbFSEditBegin(rtxer->tsdb->pFS, rtxer->fopArr, TSDB_FEDIT_MERGE);
TSDB_CHECK_CODE(code, lino, _exit);
taosThreadRwlockWrlock(&rtxer->tsdb->rwLock);
code = tsdbFSEditCommit(rtxer->tsdb->pFS);
if (code) {
taosThreadRwlockUnlock(&rtxer->tsdb->rwLock);
TSDB_CHECK_CODE(code, lino, _exit);
}
taosThreadRwlockUnlock(&rtxer->tsdb->rwLock);
TARRAY2_DESTROY(rtxer->fopArr, NULL);
tsdbFSDestroyCopySnapshot(&rtxer->fsetArr);
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(rtxer->tsdb->pVnode), lino, code);
} else {
tsdbInfo("vid:%d, cid:%" PRId64 ", %s done", TD_VID(rtxer->tsdb->pVnode), rtxer->cid, __func__);
}
return code;
}
static int32_t tsdbDoRetention2(STsdb *tsdb) {
int32_t code = 0;
int32_t lino = 0;
SRTXer rtxer[1] = {0};
code = tsdbDoRetentionBegin(tsdb, rtxer);
TSDB_CHECK_CODE(code, lino, _exit);
while (rtxer->ctx->fsetArrIdx < TARRAY2_SIZE(rtxer->fsetArr)) {
rtxer->ctx->fset = TARRAY2_GET(rtxer->fsetArr, rtxer->ctx->fsetArrIdx);
STFileObj *fobj;
int32_t expLevel = tsdbFidLevel(rtxer->ctx->fset->fid, &rtxer->tsdb->keepCfg, rtxer->now);
if (expLevel < 0) { // remove the file set
for (int32_t ftype = 0; (ftype < TSDB_FTYPE_MAX) && (fobj = rtxer->ctx->fset->farr[ftype], 1); ++ftype) {
if (fobj == NULL) continue;
code = tsdbDoRemoveFileObject(rtxer, fobj);
TSDB_CHECK_CODE(code, lino, _exit);
}
SSttLvl *lvl;
TARRAY2_FOREACH(rtxer->ctx->fset->lvlArr, lvl) {
TARRAY2_FOREACH(lvl->fobjArr, fobj) {
code = tsdbDoRemoveFileObject(rtxer, fobj);
TSDB_CHECK_CODE(code, lino, _exit);
}
}
} else if (expLevel == 0) {
continue;
} else {
SDiskID did;
if (tfsAllocDisk(rtxer->tsdb->pVnode->pTfs, expLevel, &did) < 0) {
code = terrno;
TSDB_CHECK_CODE(code, lino, _exit);
}
// data
for (int32_t ftype = 0; ftype < TSDB_FTYPE_MAX && (fobj = rtxer->ctx->fset->farr[ftype], 1); ++ftype) {
if (fobj == NULL) continue;
if (fobj->f->did.level == did.level) continue;
code = tsdbDoMigrateFileObj(rtxer, fobj, &did);
TSDB_CHECK_CODE(code, lino, _exit);
}
// stt
SSttLvl *lvl;
TARRAY2_FOREACH(rtxer->ctx->fset->lvlArr, lvl) {
TARRAY2_FOREACH(lvl->fobjArr, fobj) {
if (fobj->f->did.level == did.level) continue;
code = tsdbDoMigrateFileObj(rtxer, fobj, &did);
TSDB_CHECK_CODE(code, lino, _exit);
}
}
}
}
code = tsdbDoRetentionEnd(rtxer);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
TSDB_ERROR_LOG(TD_VID(rtxer->tsdb->pVnode), lino, code);
}
return code;
}

1966
source/dnode/vnode/src/tsdb/tsdbSnapshot.c
View File

File diff suppressed because it is too large Load Diff

19
source/dnode/vnode/src/tsdb/tsdbSttFileRW.c
View File

@ -654,7 +654,17 @@ static int32_t tsdbSttFWriterDoOpen(SSttFileWriter *writer) {
if (!writer->config->skmRow) writer->config->skmRow = writer->skmRow;
if (!writer->config->bufArr) writer->config->bufArr = writer->bufArr;
writer->file[0] = writer->config->file;
writer->file[0] = (STFile){
.type = TSDB_FTYPE_STT,
.did = writer->config->did,
.fid = writer->config->fid,
.cid = writer->config->cid,
.size = 0,
.stt[0] =
{
.level = writer->config->level,
},
};
// open file
int32_t flag = TD_FILE_READ | TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC;
@ -735,7 +745,7 @@ static int32_t tsdbSttFWriterCloseCommit(SSttFileWriter *writer, TFileOpArray *o
ASSERT(writer->file->size > 0);
STFileOp op = (STFileOp){
.optype = TSDB_FOP_CREATE,
.fid = writer->config->file.fid,
.fid = writer->config->fid,
.nf = writer->file[0],
};
@ -751,16 +761,13 @@ _exit:
static int32_t tsdbSttFWriterCloseAbort(SSttFileWriter *writer) {
char fname[TSDB_FILENAME_LEN];
tsdbTFileName(writer->config->tsdb, &writer->config->file, fname);
tsdbTFileName(writer->config->tsdb, writer->file, fname);
tsdbCloseFile(&writer->fd);
taosRemoveFile(fname);
return 0;
}
int32_t tsdbSttFileWriterOpen(const SSttFileWriterConfig *config, SSttFileWriter **writer) {
ASSERT(config->file.type == TSDB_FTYPE_STT);
ASSERT(config->file.size == 0);
writer[0] = taosMemoryCalloc(1, sizeof(*writer[0]));
if (writer[0] == NULL) return TSDB_CODE_OUT_OF_MEMORY;

5
source/dnode/vnode/src/tsdb/tsdbSttFileRW.h
View File

@ -71,7 +71,10 @@ struct SSttFileWriterConfig {
int32_t szPage;
int8_t cmprAlg;
int64_t compactVersion;
STFile file;
SDiskID did;
int32_t fid;
int64_t cid;
int32_t level;
SSkmInfo *skmTb;
SSkmInfo *skmRow;
uint8_t **bufArr;

2
source/dnode/vnode/src/vnd/vnodeSvr.c
View File

@ -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);
}

2
source/libs/command/inc/commandInt.h
View File

@ -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)))

30
source/libs/command/src/explain.c
View File

@ -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));

4
source/libs/executor/inc/executorInt.h
View File

@ -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.

2
source/libs/executor/inc/tsort.h
View File

@ -146,7 +146,7 @@ void* tsortGetValue(STupleHandle* pVHandle, int32_t colId);
* @return
*/
uint64_t tsortGetGroupId(STupleHandle* pVHandle);
void* tsortGetBlockInfo(STupleHandle* pVHandle);
/**
*
* @param pSortHandle

17
source/libs/executor/src/aggregateoperator.c
View File

@ -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);

1
source/libs/executor/src/cachescanoperator.c
View File

@ -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,

5
source/libs/executor/src/eventwindowoperator.c
View File

@ -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);

1
source/libs/executor/src/exchangeoperator.c
View File

@ -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);
}

12
source/libs/executor/src/filloperator.c
View File

@ -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;

14
source/libs/executor/src/groupoperator.c
View File

@ -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,

5
source/libs/executor/src/joinoperator.c
View File

@ -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;
}

27
source/libs/executor/src/projectoperator.c
View File

@ -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

1
source/libs/executor/src/scanoperator.c
View File

@ -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;

10
source/libs/executor/src/sortoperator.c
View File

@ -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,

143
source/libs/executor/src/timesliceoperator.c
View File

@ -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);
doHandleTimeslice(pOperator, pBlock);
if (checkWindowBoundReached(pSliceInfo) || checkThresholdReached(pSliceInfo, pOperator->resultInfo.threshold)) {
doFilter(pResBlock, pOperator->exprSupp.pFilterInfo, NULL);
if (pResBlock->info.rows != 0) {
goto _finished;
}
}
int32_t code = initKeeperInfo(pSliceInfo, pBlock, &pOperator->exprSupp);
if (code != TSDB_CODE_SUCCESS) {
T_LONG_JMP(pTaskInfo->env, code);
}
// 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;

66
source/libs/executor/src/timewindowoperator.c
View File

@ -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;

1
source/libs/executor/src/tsort.c
View File

@ -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};

9
source/libs/nodes/src/nodesCloneFuncs.c
View File

@ -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;
}

33
source/libs/nodes/src/nodesCodeFuncs.c
View File

@ -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);
}

40
source/libs/nodes/src/nodesMsgFuncs.c
View File

@ -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;
}

21
source/libs/planner/src/planLogicCreater.c
View File

@ -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) {

89
source/libs/planner/src/planOptimizer.c
View File

@ -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},

18
source/libs/planner/src/planPhysiCreater.c
View File

@ -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);

8
source/libs/planner/src/planSpliter.c
View File

@ -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);

2
source/libs/scheduler/src/schTask.c
View File

@ -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));

4
source/util/src/terror.c
View File

@ -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")

1
tests/parallel_test/cases.task
View File

@ -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

37
tests/script/tsim/query/explain_tsorder.sim Normal file
View File

@ -0,0 +1,37 @@
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-----------------------

2560
tests/script/tsim/query/r/explain_tsorder.result Normal file
View File

File diff suppressed because it is too large Load Diff

73
tests/script/tsim/query/t/explain_tsorder.sql Normal file
View File

@ -0,0 +1,73 @@
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;

17
tests/script/tsim/sma/tsmaCreateInsertQuery.sim
View File

@ -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

2
utils/tsim/src/simExe.c
View File

@ -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);