diff --git a/include/common/tcommon.h b/include/common/tcommon.h
index 28d771bbbd50aec70bad5d7c2e5cd1529c8e40a5..614e7c9974b6ed6bde82c9d9e354245c871c231c 100644
--- a/include/common/tcommon.h
+++ b/include/common/tcommon.h
@@ -155,8 +155,8 @@ typedef struct SQueryTableDataCond {
int32_t numOfCols;
SColumnInfo* colList;
int32_t type; // data block load type:
- int32_t numOfTWindows;
- STimeWindow* twindows;
+// int32_t numOfTWindows;
+ STimeWindow twindows;
int64_t startVersion;
int64_t endVersion;
} SQueryTableDataCond;
diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c
index 5b3993dd4027db3f7a86dad6f28bc5feaed84bf9..ab5ce3742aed34b0408d1bd2e2e511416b317763 100644
--- a/source/common/src/tglobal.c
+++ b/source/common/src/tglobal.c
@@ -418,7 +418,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
if (cfgAddInt32(pCfg, "numOfMnodeReadThreads", tsNumOfMnodeReadThreads, 1, 1024, 0) != 0) return -1;
tsNumOfVnodeQueryThreads = tsNumOfCores / 2;
- tsNumOfVnodeQueryThreads = TMAX(tsNumOfVnodeQueryThreads, 1);
+ tsNumOfVnodeQueryThreads = TMAX(tsNumOfVnodeQueryThreads, 2);
if (cfgAddInt32(pCfg, "numOfVnodeQueryThreads", tsNumOfVnodeQueryThreads, 1, 1024, 0) != 0) return -1;
tsNumOfVnodeFetchThreads = TRANGE(tsNumOfVnodeFetchThreads, 1, 1);
diff --git a/source/dnode/vnode/inc/vnode.h b/source/dnode/vnode/inc/vnode.h
index 62a39711ec7af9bf33b85b5565ecea3de1b27eb7..b42b0f2b44bd96303a14942a00eda376e56fe3e0 100644
--- a/source/dnode/vnode/inc/vnode.h
+++ b/source/dnode/vnode/inc/vnode.h
@@ -130,7 +130,7 @@ bool tsdbNextDataBlock(STsdbReader *pReader);
void tsdbRetrieveDataBlockInfo(STsdbReader *pReader, SDataBlockInfo *pDataBlockInfo);
int32_t tsdbRetrieveDatablockSMA(STsdbReader *pReader, SColumnDataAgg ***pBlockStatis, bool *allHave);
SArray *tsdbRetrieveDataBlock(STsdbReader *pTsdbReadHandle, SArray *pColumnIdList);
-int32_t tsdbReaderReset(STsdbReader *pReader, SQueryTableDataCond *pCond, int32_t tWinIdx);
+int32_t tsdbReaderReset(STsdbReader *pReader, SQueryTableDataCond *pCond);
int32_t tsdbGetFileBlocksDistInfo(STsdbReader *pReader, STableBlockDistInfo *pTableBlockInfo);
int64_t tsdbGetNumOfRowsInMemTable(STsdbReader *pHandle);
void *tsdbGetIdx(SMeta *pMeta);
diff --git a/source/dnode/vnode/src/tsdb/tsdbRead.c b/source/dnode/vnode/src/tsdb/tsdbRead.c
index 9e01468fc790d1e5608784e8d08fa1526a71177c..f650480ff2b58debb9a92d0f53587ecb3a13af8a 100644
--- a/source/dnode/vnode/src/tsdb/tsdbRead.c
+++ b/source/dnode/vnode/src/tsdb/tsdbRead.c
@@ -356,14 +356,14 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, STsd
initReaderStatus(&pReader->status);
pReader->pTsdb =
- getTsdbByRetentions(pVnode, pCond->twindows[0].skey, pVnode->config.tsdbCfg.retentions, idstr, &level);
+ getTsdbByRetentions(pVnode, pCond->twindows.skey, pVnode->config.tsdbCfg.retentions, idstr, &level);
pReader->suid = pCond->suid;
pReader->order = pCond->order;
pReader->capacity = 4096;
pReader->idStr = (idstr != NULL) ? strdup(idstr) : NULL;
pReader->verRange = getQueryVerRange(pVnode, pCond, level);
pReader->type = pCond->type;
- pReader->window = updateQueryTimeWindow(pVnode->pTsdb, pCond->twindows);
+ pReader->window = updateQueryTimeWindow(pVnode->pTsdb, &pCond->twindows);
ASSERT(pCond->numOfCols > 0);
@@ -2954,7 +2954,7 @@ SArray* tsdbRetrieveDataBlock(STsdbReader* pReader, SArray* pIdList) {
return pReader->pResBlock->pDataBlock;
}
-int32_t tsdbReaderReset(STsdbReader* pReader, SQueryTableDataCond* pCond, int32_t tWinIdx) {
+int32_t tsdbReaderReset(STsdbReader* pReader, SQueryTableDataCond* pCond) {
if (isEmptyQueryTimeWindow(&pReader->window)) {
return TSDB_CODE_SUCCESS;
}
@@ -2964,7 +2964,7 @@ int32_t tsdbReaderReset(STsdbReader* pReader, SQueryTableDataCond* pCond, int32_
pReader->status.loadFromFile = true;
pReader->status.pTableIter = NULL;
- pReader->window = updateQueryTimeWindow(pReader->pTsdb, &pCond->twindows[tWinIdx]);
+ pReader->window = updateQueryTimeWindow(pReader->pTsdb, &pCond->twindows);
// allocate buffer in order to load data blocks from file
memset(&pReader->suppInfo.tsColAgg, 0, sizeof(SColumnDataAgg));
diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h
index ae0d0bebbd49b05e07c6bedc70d5bafaefd640f6..7cba2c9eaf2f148f008665b31ffbbf77ec8c1238 100644
--- a/source/libs/executor/inc/executorimpl.h
+++ b/source/libs/executor/inc/executorimpl.h
@@ -279,9 +279,6 @@ typedef struct STableScanInfo {
SScanInfo scanInfo;
int32_t scanTimes;
SNode* pFilterNode; // filter info, which is push down by optimizer
- SqlFunctionCtx* pCtx; // which belongs to the direct upstream operator operator query context,todo: remove this by using SExprSup
- int32_t* rowEntryInfoOffset; // todo: remove this by using SExprSup
- SExprInfo* pExpr;// todo: remove this by using SExprSup
SSDataBlock* pResBlock;
SArray* pColMatchInfo;
@@ -290,14 +287,10 @@ typedef struct STableScanInfo {
int32_t scanFlag; // table scan flag to denote if it is a repeat/reverse/main scan
int32_t dataBlockLoadFlag;
SInterval interval; // if the upstream is an interval operator, the interval info is also kept here to get the time window to check if current data block needs to be loaded.
-
SSampleExecInfo sample; // sample execution info
- int32_t curTWinIdx;
int32_t currentGroupId;
int32_t currentTable;
- uint64_t queryId; // todo remove it
- uint64_t taskId; // todo remove it
#if 0
struct {
@@ -905,7 +898,7 @@ int32_t aggDecodeResultRow(SOperatorInfo* pOperator, char* result);
int32_t aggEncodeResultRow(SOperatorInfo* pOperator, char** result, int32_t* length);
STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts, SInterval* pInterval,
- int32_t precision, int32_t order);
+ int32_t order);
int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimaryColumn, int32_t startPos, TSKEY ekey,
__block_search_fn_t searchFn, STableQueryInfo* item, int32_t order);
int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order);
@@ -921,7 +914,6 @@ int32_t updateSessionWindowInfo(SResultWindowInfo* pWinInfo, TSKEY* pStartTs,
TSKEY* pEndTs, int32_t rows, int32_t start, int64_t gap, SHashObj* pStDeleted);
bool functionNeedToExecute(SqlFunctionCtx* pCtx);
-int32_t compareTimeWindow(const void* p1, const void* p2, const void* param);
int32_t finalizeResultRowIntoResultDataBlock(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPosition,
SqlFunctionCtx* pCtx, SExprInfo* pExprInfo, int32_t numOfExprs, const int32_t* rowCellOffset,
SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);
diff --git a/source/libs/executor/src/executil.c b/source/libs/executor/src/executil.c
index 1f11f4be986d65ea623cc8b7bf0711e55100454e..e3579f9dcbc8899338bdcec37c5adffa0c1520f9 100644
--- a/source/libs/executor/src/executil.c
+++ b/source/libs/executor/src/executil.c
@@ -13,6 +13,7 @@
* along with this program. If not, see .
*/
+#include
#include "function.h"
#include "functionMgt.h"
#include "index.h"
@@ -769,12 +770,9 @@ int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysi
// pCond->twindow = pTableScanNode->scanRange;
// TODO: get it from stable scan node
- pCond->numOfTWindows = 1;
- pCond->twindows = taosMemoryCalloc(pCond->numOfTWindows, sizeof(STimeWindow));
- pCond->twindows[0] = pTableScanNode->scanRange;
- pCond->suid = pTableScanNode->scan.suid;
-
- pCond->type = BLOCK_LOAD_OFFSET_ORDER;
+ pCond->twindows = pTableScanNode->scanRange;
+ pCond->suid = pTableScanNode->scan.suid;
+ pCond->type = BLOCK_LOAD_OFFSET_ORDER;
pCond->startVersion = -1;
pCond->endVersion = -1;
// pCond->type = pTableScanNode->scanFlag;
@@ -826,3 +824,87 @@ int32_t convertFillType(int32_t mode) {
return type;
}
+
+static void getInitialStartTimeWindow(SInterval* pInterval, TSKEY ts, STimeWindow* w, bool ascQuery) {
+ if (ascQuery) {
+ getAlignQueryTimeWindow(pInterval, pInterval->precision, ts, w);
+ } else {
+ // the start position of the first time window in the endpoint that spreads beyond the queried last timestamp
+ getAlignQueryTimeWindow(pInterval, pInterval->precision, ts, w);
+
+ int64_t key = w->skey;
+ while (key < ts) { // moving towards end
+ key = taosTimeAdd(key, pInterval->sliding, pInterval->slidingUnit, pInterval->precision);
+ if (key >= ts) {
+ break;
+ }
+
+ w->skey = key;
+ }
+ }
+}
+
+static STimeWindow doCalculateTimeWindow(int64_t ts, SInterval* pInterval) {
+ STimeWindow w = {0};
+
+ if (pInterval->intervalUnit == 'n' || pInterval->intervalUnit == 'y') {
+ w.skey = taosTimeTruncate(ts, pInterval, pInterval->precision);
+ w.ekey = taosTimeAdd(w.skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision) - 1;
+ } else {
+ int64_t st = w.skey;
+
+ if (st > ts) {
+ st -= ((st - ts + pInterval->sliding - 1) / pInterval->sliding) * pInterval->sliding;
+ }
+
+ int64_t et = st + pInterval->interval - 1;
+ if (et < ts) {
+ st += ((ts - et + pInterval->sliding - 1) / pInterval->sliding) * pInterval->sliding;
+ }
+
+ w.skey = st;
+ w.ekey = taosTimeAdd(w.skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision) - 1;
+ }
+
+ return w;
+}
+
+static STimeWindow getFirstQualifiedTimeWindow(int64_t ts, STimeWindow* pWindow, SInterval* pInterval, int32_t order) {
+ int32_t factor = (order == TSDB_ORDER_ASC)? -1:1;
+
+ STimeWindow win = *pWindow;
+ STimeWindow save = win;
+ while(win.skey <= ts && win.ekey >= ts) {
+ save = win;
+ win.skey = taosTimeAdd(win.skey, factor * pInterval->sliding, pInterval->slidingUnit, pInterval->precision);
+ win.ekey = taosTimeAdd(win.ekey, factor * pInterval->sliding, pInterval->slidingUnit, pInterval->precision);
+ }
+
+ return save;
+}
+
+// get the correct time window according to the handled timestamp
+STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts, SInterval* pInterval,
+ int32_t order) {
+ STimeWindow w = {0};
+ if (pResultRowInfo->cur.pageId == -1) { // the first window, from the previous stored value
+ getInitialStartTimeWindow(pInterval, ts, &w, (order == TSDB_ORDER_ASC));
+ w.ekey = taosTimeAdd(w.skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision) - 1;
+ return w;
+ }
+
+ w = getResultRowByPos(pBuf, &pResultRowInfo->cur)->win;
+
+ // in case of typical time window, we can calculate time window directly.
+ if (w.skey > ts || w.ekey < ts) {
+ w = doCalculateTimeWindow(ts, pInterval);
+ }
+
+ if (pInterval->interval != pInterval->sliding) {
+ // it is an sliding window query, in which sliding value is not equalled to
+ // interval value, and we need to find the first qualified time window.
+ w = getFirstQualifiedTimeWindow(ts, &w, pInterval, order);
+ }
+
+ return w;
+}
\ No newline at end of file
diff --git a/source/libs/executor/src/executorMain.c b/source/libs/executor/src/executorMain.c
index 53ebc322a8e556abda119a3066b063b50be81edf..12fcd103e889115b5f909d2918efe08ce03042be 100644
--- a/source/libs/executor/src/executorMain.c
+++ b/source/libs/executor/src/executorMain.c
@@ -327,12 +327,11 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, const STqOffsetVal* pOffset) {
ASSERT(found);
tsdbSetTableId(pTableScanInfo->dataReader, uid);
- int64_t oldSkey = pTableScanInfo->cond.twindows[0].skey;
- pTableScanInfo->cond.twindows[0].skey = ts + 1;
- tsdbReaderReset(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0);
- pTableScanInfo->cond.twindows[0].skey = oldSkey;
+ int64_t oldSkey = pTableScanInfo->cond.twindows.skey;
+ pTableScanInfo->cond.twindows.skey = ts + 1;
+ tsdbReaderReset(pTableScanInfo->dataReader, &pTableScanInfo->cond);
+ pTableScanInfo->cond.twindows.skey = oldSkey;
pTableScanInfo->scanTimes = 0;
- pTableScanInfo->curTWinIdx = 0;
qDebug("tsdb reader offset seek to uid %ld ts %ld, table cur set to %d , all table num %d", uid, ts,
pTableScanInfo->currentTable, tableSz);
diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c
index f2f696b21f6a84f9ceb9db2223c3f26d5cd3b499..6a2b9ea2a21bcd52c10bb709da03ef16c7581fa7 100644
--- a/source/libs/executor/src/executorimpl.c
+++ b/source/libs/executor/src/executorimpl.c
@@ -1038,6 +1038,7 @@ static bool overlapWithTimeWindow(STaskAttr* pQueryAttr, SDataBlockInfo* pBlockI
#endif
static uint32_t doFilterByBlockTimeWindow(STableScanInfo* pTableScanInfo, SSDataBlock* pBlock) {
+#if 0
SqlFunctionCtx* pCtx = pTableScanInfo->pCtx;
uint32_t status = BLK_DATA_NOT_LOAD;
@@ -1059,6 +1060,8 @@ static uint32_t doFilterByBlockTimeWindow(STableScanInfo* pTableScanInfo, SSData
}
return status;
+#endif
+ return 0;
}
int32_t loadDataBlockOnDemand(SExecTaskInfo* pTaskInfo, STableScanInfo* pTableScanInfo, SSDataBlock* pBlock,
@@ -2887,12 +2890,11 @@ int32_t doPrepareScan(SOperatorInfo* pOperator, uint64_t uid, int64_t ts) {
ASSERT(found);
tsdbSetTableId(pInfo->dataReader, uid);
- int64_t oldSkey = pInfo->cond.twindows[0].skey;
- pInfo->cond.twindows[0].skey = ts + 1;
- tsdbReaderReset(pInfo->dataReader, &pInfo->cond, 0);
- pInfo->cond.twindows[0].skey = oldSkey;
+ int64_t oldSkey = pInfo->cond.twindows.skey;
+ pInfo->cond.twindows.skey = ts + 1;
+ tsdbReaderReset(pInfo->dataReader, &pInfo->cond);
+ pInfo->cond.twindows.skey = oldSkey;
pInfo->scanTimes = 0;
- pInfo->curTWinIdx = 0;
qDebug("tsdb reader offset seek to uid %" PRId64 " ts %" PRId64 ", table cur set to %d , all table num %d", uid, ts,
pInfo->currentTable, tableSz);
@@ -4348,9 +4350,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
cond.colList->type = TSDB_DATA_TYPE_TIMESTAMP;
cond.colList->bytes = sizeof(TSKEY);
- cond.numOfTWindows = 1;
- cond.twindows = taosMemoryCalloc(1, sizeof(STimeWindow));
- cond.twindows[0] = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
+ cond.twindows = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
cond.suid = pBlockNode->suid;
cond.type = BLOCK_LOAD_OFFSET_ORDER;
}
@@ -4549,18 +4549,6 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
return pOptr;
}
-int32_t compareTimeWindow(const void* p1, const void* p2, const void* param) {
- const SQueryTableDataCond* pCond = param;
- const STimeWindow* pWin1 = p1;
- const STimeWindow* pWin2 = p2;
- if (pCond->order == TSDB_ORDER_ASC) {
- return pWin1->skey - pWin2->skey;
- } else if (pCond->order == TSDB_ORDER_DESC) {
- return pWin2->skey - pWin1->skey;
- }
- return 0;
-}
-
SArray* extractColumnInfo(SNodeList* pNodeList) {
size_t numOfCols = LIST_LENGTH(pNodeList);
SArray* pList = taosArrayInit(numOfCols, sizeof(SColumn));
diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c
index f7155b2431aec345d6ef8c78d9e7e0e75798be6a..7410205d43b9b73f54271e8a79f11701f5af14b5 100644
--- a/source/libs/executor/src/scanoperator.c
+++ b/source/libs/executor/src/scanoperator.c
@@ -294,13 +294,8 @@ static void prepareForDescendingScan(STableScanInfo* pTableScanInfo, SqlFunction
// setupQueryRangeForReverseScan(pTableScanInfo);
pTableScanInfo->cond.order = TSDB_ORDER_DESC;
- for (int32_t i = 0; i < pTableScanInfo->cond.numOfTWindows; ++i) {
- STimeWindow* pTWindow = &pTableScanInfo->cond.twindows[i];
- TSWAP(pTWindow->skey, pTWindow->ekey);
- }
-
- SQueryTableDataCond* pCond = &pTableScanInfo->cond;
- taosqsort(pCond->twindows, pCond->numOfTWindows, sizeof(STimeWindow), pCond, compareTimeWindow);
+ STimeWindow* pTWindow = &pTableScanInfo->cond.twindows;
+ TSWAP(pTWindow->skey, pTWindow->ekey);
}
int32_t addTagPseudoColumnData(SReadHandle* pHandle, SExprInfo* pPseudoExpr, int32_t numOfPseudoExpr,
@@ -450,16 +445,10 @@ static SSDataBlock* doTableScanGroup(SOperatorInfo* pOperator) {
// do the ascending order traverse in the first place.
while (pTableScanInfo->scanTimes < pTableScanInfo->scanInfo.numOfAsc) {
- while (pTableScanInfo->curTWinIdx < pTableScanInfo->cond.numOfTWindows) {
- SSDataBlock* p = doTableScanImpl(pOperator);
- if (p != NULL) {
- ASSERT(p->info.uid != 0);
- return p;
- }
- pTableScanInfo->curTWinIdx += 1;
- if (pTableScanInfo->curTWinIdx < pTableScanInfo->cond.numOfTWindows) {
- tsdbReaderReset(pTableScanInfo->dataReader, &pTableScanInfo->cond, pTableScanInfo->curTWinIdx);
- }
+ SSDataBlock* p = doTableScanImpl(pOperator);
+ if (p != NULL) {
+ ASSERT(p->info.uid != 0);
+ return p;
}
pTableScanInfo->scanTimes += 1;
@@ -468,40 +457,25 @@ static SSDataBlock* doTableScanGroup(SOperatorInfo* pOperator) {
setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);
pTableScanInfo->scanFlag = REPEAT_SCAN;
qDebug("%s start to repeat ascending order scan data blocks due to query func required", GET_TASKID(pTaskInfo));
- for (int32_t i = 0; i < pTableScanInfo->cond.numOfTWindows; ++i) {
- STimeWindow* pWin = &pTableScanInfo->cond.twindows[i];
- qDebug("%s qrange:%" PRId64 "-%" PRId64, GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey);
- }
+
// do prepare for the next round table scan operation
- tsdbReaderReset(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0);
- pTableScanInfo->curTWinIdx = 0;
+ tsdbReaderReset(pTableScanInfo->dataReader, &pTableScanInfo->cond);
}
}
int32_t total = pTableScanInfo->scanInfo.numOfAsc + pTableScanInfo->scanInfo.numOfDesc;
if (pTableScanInfo->scanTimes < total) {
if (pTableScanInfo->cond.order == TSDB_ORDER_ASC) {
- prepareForDescendingScan(pTableScanInfo, pTableScanInfo->pCtx, 0);
- tsdbReaderReset(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0);
- pTableScanInfo->curTWinIdx = 0;
+ prepareForDescendingScan(pTableScanInfo, pOperator->exprSupp.pCtx, 0);
+ tsdbReaderReset(pTableScanInfo->dataReader, &pTableScanInfo->cond);
}
qDebug("%s start to descending order scan data blocks due to query func required", GET_TASKID(pTaskInfo));
- for (int32_t i = 0; i < pTableScanInfo->cond.numOfTWindows; ++i) {
- STimeWindow* pWin = &pTableScanInfo->cond.twindows[i];
- qDebug("%s qrange:%" PRId64 "-%" PRId64, GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey);
- }
while (pTableScanInfo->scanTimes < total) {
- while (pTableScanInfo->curTWinIdx < pTableScanInfo->cond.numOfTWindows) {
- SSDataBlock* p = doTableScanImpl(pOperator);
- if (p != NULL) {
- return p;
- }
- pTableScanInfo->curTWinIdx += 1;
- if (pTableScanInfo->curTWinIdx < pTableScanInfo->cond.numOfTWindows) {
- tsdbReaderReset(pTableScanInfo->dataReader, &pTableScanInfo->cond, pTableScanInfo->curTWinIdx);
- }
+ SSDataBlock* p = doTableScanImpl(pOperator);
+ if (p != NULL) {
+ return p;
}
pTableScanInfo->scanTimes += 1;
@@ -512,12 +486,7 @@ static SSDataBlock* doTableScanGroup(SOperatorInfo* pOperator) {
qDebug("%s start to repeat descending order scan data blocks due to query func required",
GET_TASKID(pTaskInfo));
- for (int32_t i = 0; i < pTableScanInfo->cond.numOfTWindows; ++i) {
- STimeWindow* pWin = &pTableScanInfo->cond.twindows[i];
- qDebug("%s qrange:%" PRId64 "-%" PRId64, GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey);
- }
- tsdbReaderReset(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0);
- pTableScanInfo->curTWinIdx = 0;
+ tsdbReaderReset(pTableScanInfo->dataReader, &pTableScanInfo->cond);
}
}
}
@@ -544,9 +513,8 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
}
STableKeyInfo* pTableInfo = taosArrayGet(pTaskInfo->tableqinfoList.pTableList, pInfo->currentTable);
tsdbSetTableId(pInfo->dataReader, pTableInfo->uid);
- tsdbReaderReset(pInfo->dataReader, &pInfo->cond, 0);
+ tsdbReaderReset(pInfo->dataReader, &pInfo->cond);
pInfo->scanTimes = 0;
- pInfo->curTWinIdx = 0;
}
}
@@ -578,8 +546,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
SArray* tableList = taosArrayGetP(pTaskInfo->tableqinfoList.pGroupList, pInfo->currentGroupId);
// tsdbSetTableList(pInfo->dataReader, tableList);
- tsdbReaderReset(pInfo->dataReader, &pInfo->cond, 0);
- pInfo->curTWinIdx = 0;
+ tsdbReaderReset(pInfo->dataReader, &pInfo->cond);
pInfo->scanTimes = 0;
result = doTableScanGroup(pOperator);
@@ -650,7 +617,6 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
pInfo->pFilterNode = pTableScanNode->scan.node.pConditions;
pInfo->scanFlag = MAIN_SCAN;
pInfo->pColMatchInfo = pColList;
- pInfo->curTWinIdx = 0;
pInfo->currentGroupId = -1;
pOperator->name = "TableScanOperator"; // for debug purpose
@@ -880,12 +846,7 @@ static void setGroupId(SStreamScanInfo* pInfo, SSDataBlock* pBlock, int32_t grou
}
void resetTableScanInfo(STableScanInfo* pTableScanInfo, STimeWindow* pWin) {
- pTableScanInfo->cond.twindows[0] = *pWin;
- pTableScanInfo->curTWinIdx = 0;
- // tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0);
- // if (!pTableScanInfo->dataReader) {
- // return false;
- // }
+ pTableScanInfo->cond.twindows = *pWin;
pTableScanInfo->scanTimes = 0;
pTableScanInfo->currentGroupId = -1;
}
@@ -943,8 +904,7 @@ static bool prepareDataScan(SStreamScanInfo* pInfo, SSDataBlock* pSDB, int32_t t
setGroupId(pInfo, pSDB, GROUPID_COLUMN_INDEX, *pRowIndex);
(*pRowIndex) += updateSessionWindowInfo(pCurWin, tsCols, NULL, pSDB->info.rows, *pRowIndex, gap, NULL);
} else {
- win = getActiveTimeWindow(NULL, &dumyInfo, tsCols[*pRowIndex], &pInfo->interval, pInfo->interval.precision,
- TSDB_ORDER_ASC);
+ win = getActiveTimeWindow(NULL, &dumyInfo, tsCols[*pRowIndex], &pInfo->interval, TSDB_ORDER_ASC);
setGroupId(pInfo, pSDB, GROUPID_COLUMN_INDEX, *pRowIndex);
(*pRowIndex) +=
getNumOfRowsInTimeWindow(&pSDB->info, tsCols, *pRowIndex, win.ekey, binarySearchForKey, NULL, TSDB_ORDER_ASC);
diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c
index f9810913e362b8fea75699976685eb6911f825df..fb1169b3d32bf3482c0910d6d7dac57b3cd24cad 100644
--- a/source/libs/executor/src/timewindowoperator.c
+++ b/source/libs/executor/src/timewindowoperator.c
@@ -59,136 +59,6 @@ static void doCloseWindow(SResultRowInfo* pResultRowInfo, const SIntervalAggOper
static TSKEY getStartTsKey(STimeWindow* win, const TSKEY* tsCols) { return tsCols == NULL ? win->skey : tsCols[0]; }
-static void getInitialStartTimeWindow(SInterval* pInterval, int32_t precision, TSKEY ts, STimeWindow* w,
- bool ascQuery) {
- if (ascQuery) {
- getAlignQueryTimeWindow(pInterval, precision, ts, w);
- } else {
- // the start position of the first time window in the endpoint that spreads beyond the queried last timestamp
- getAlignQueryTimeWindow(pInterval, precision, ts, w);
-
- int64_t key = w->skey;
- while (key < ts) { // moving towards end
- key = taosTimeAdd(key, pInterval->sliding, pInterval->slidingUnit, precision);
- if (key >= ts) {
- break;
- }
-
- w->skey = key;
- }
- }
-}
-
-static STimeWindow getFirstQualifiedTimeWindow(int64_t ts, STimeWindow* pWindow, SInterval* pInterval, int32_t order) {
- int32_t factor = (order == TSDB_ORDER_ASC)? -1:1;
-
- STimeWindow win = *pWindow;
- STimeWindow save = win;
- while(win.skey <= ts && win.ekey >= ts) {
- save = win;
- win.skey = taosTimeAdd(win.skey, factor * pInterval->sliding, pInterval->slidingUnit, pInterval->precision);
- win.ekey = taosTimeAdd(win.ekey, factor * pInterval->sliding, pInterval->slidingUnit, pInterval->precision);
- }
-
- return save;
-}
-
-// todo do refactor
-// get the correct time window according to the handled timestamp
-STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts, SInterval* pInterval,
- int32_t precision, int32_t order) {
- STimeWindow w = {0};
-
- if (pResultRowInfo->cur.pageId == -1) { // the first window, from the previous stored value
- getInitialStartTimeWindow(pInterval, precision, ts, &w, true);
- w.ekey = taosTimeAdd(w.skey, pInterval->interval, pInterval->intervalUnit, precision) - 1;
- return w;
- }
-
- w = getResultRowByPos(pBuf, &pResultRowInfo->cur)->win;
-
- if (pInterval->interval == pInterval->sliding) {
- if (w.skey > ts || w.ekey < ts) {
- if (pInterval->intervalUnit == 'n' || pInterval->intervalUnit == 'y') {
- w.skey = taosTimeTruncate(ts, pInterval, precision);
- w.ekey = taosTimeAdd(w.skey, pInterval->interval, pInterval->intervalUnit, precision) - 1;
- } else {
- int64_t st = w.skey;
-
- if (st > ts) {
- st -= ((st - ts + pInterval->sliding - 1) / pInterval->sliding) * pInterval->sliding;
- }
-
- int64_t et = st + pInterval->interval - 1;
- if (et < ts) {
- st += ((ts - et + pInterval->sliding - 1) / pInterval->sliding) * pInterval->sliding;
- }
-
- w.skey = st;
- w.ekey = taosTimeAdd(w.skey, pInterval->interval, pInterval->intervalUnit, precision) - 1;
- }
- }
- } else { // it is an sliding window query, in which sliding value is not equalled to
- // interval value, and we need to find the first qualified time window for asc/desc traverse respectively.
- if (order == TSDB_ORDER_ASC) {
- if (w.skey <= ts && w.ekey >= ts) {
- // ts is resident in current time window, but we need to find the first
- //qualified time window that cover this timestamp
- w = getFirstQualifiedTimeWindow(ts, &w, pInterval, order);
- } else {
- // todo refactor:
- if (pInterval->intervalUnit == 'n' || pInterval->intervalUnit == 'y') {
- w.skey = taosTimeTruncate(ts, pInterval, precision);
- w.ekey = taosTimeAdd(w.skey, pInterval->interval, pInterval->intervalUnit, precision) - 1;
- } else {
- int64_t st = w.skey;
-
- if (st > ts) {
- st -= ((st - ts + pInterval->sliding - 1) / pInterval->sliding) * pInterval->sliding;
- }
-
- int64_t et = st + pInterval->interval - 1;
- if (et < ts) {
- st += ((ts - et + pInterval->sliding - 1) / pInterval->sliding) * pInterval->sliding;
- }
-
- w.skey = st;
- w.ekey = taosTimeAdd(w.skey, pInterval->interval, pInterval->intervalUnit, precision) - 1;
-
- w = getFirstQualifiedTimeWindow(ts, &w, pInterval, order);
- }
- }
- } else {
- if (w.skey <= ts && w.ekey >= ts) {
- w = getFirstQualifiedTimeWindow(ts, &w, pInterval, order);
- } else {
- // todo refactor:
- if (pInterval->intervalUnit == 'n' || pInterval->intervalUnit == 'y') {
- w.skey = taosTimeTruncate(ts, pInterval, precision);
- w.ekey = taosTimeAdd(w.skey, pInterval->interval, pInterval->intervalUnit, precision) - 1;
- } else {
- int64_t st = w.skey;
-
- if (st > ts) {
- st -= ((st - ts + pInterval->sliding - 1) / pInterval->sliding) * pInterval->sliding;
- }
-
- int64_t et = st + pInterval->interval - 1;
- if (et < ts) {
- st += ((ts - et + pInterval->sliding - 1) / pInterval->sliding) * pInterval->sliding;
- }
-
- w.skey = st;
- w.ekey = taosTimeAdd(w.skey, pInterval->interval, pInterval->intervalUnit, precision) - 1;
-
- w = getFirstQualifiedTimeWindow(ts, &w, pInterval, order);
- }
- }
- }
- }
- return w;
-}
-
static int32_t setTimeWindowOutputBuf(SResultRowInfo* pResultRowInfo, STimeWindow* win, bool masterscan,
SResultRow** pResult, int64_t tableGroupId, SqlFunctionCtx* pCtx,
int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup,
@@ -930,8 +800,7 @@ static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResul
TSKEY ts = getStartTsKey(&pBlock->info.window, tsCols);
SResultRow* pResult = NULL;
- STimeWindow win = getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts, &pInfo->interval,
- pInfo->interval.precision, pInfo->order);
+ STimeWindow win = getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts, &pInfo->interval, pInfo->order);
int32_t ret = TSDB_CODE_SUCCESS;
if (!pInfo->ignoreExpiredData || !isCloseWindow(&win, &pInfo->twAggSup)) {
ret = setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, pSup->pCtx,
@@ -1287,11 +1156,13 @@ static SSDataBlock* doBuildIntervalResult(SOperatorInfo* pOperator) {
while (1) {
doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
doFilter(pInfo->pCondition, pBlock);
+
bool hasRemain = hasDataInGroupInfo(&pInfo->groupResInfo);
if (!hasRemain) {
doSetOperatorCompleted(pOperator);
break;
}
+
if (pBlock->info.rows > 0) {
break;
}
@@ -1390,7 +1261,7 @@ void doDeleteSpecifyIntervalWindow(SAggSupporter* pAggSup, SSDataBlock* pBlock,
for (int32_t i = 0; i < pBlock->info.rows; i++) {
SResultRowInfo dumyInfo;
dumyInfo.cur.pageId = -1;
- STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, tsStarts[i], pInterval, pInterval->precision, TSDB_ORDER_ASC);
+ STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, tsStarts[i], pInterval, TSDB_ORDER_ASC);
doDeleteIntervalWindow(pAggSup, win.skey, groupIds[i]);
if (pUpWins) {
SWinRes winRes = {.ts = win.skey, .groupId = groupIds[i]};
@@ -1412,7 +1283,7 @@ static void doClearWindows(SAggSupporter* pAggSup, SExprSupp* pSup1, SInterval*
for (int32_t i = 0; i < pBlock->info.rows; i += step) {
SResultRowInfo dumyInfo;
dumyInfo.cur.pageId = -1;
- STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, tsCols[i], pInterval, pInterval->precision, TSDB_ORDER_ASC);
+ STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, tsCols[i], pInterval, TSDB_ORDER_ASC);
step = getNumOfRowsInTimeWindow(&pBlock->info, tsCols, i, win.ekey, binarySearchForKey, NULL, TSDB_ORDER_ASC);
uint64_t winGpId = pGpDatas ? pGpDatas[i] : pBlock->info.groupId;
bool res = doClearWindow(pAggSup, pSup1, (char*)&win.skey, sizeof(TKEY), winGpId, numOfOutput);
@@ -1452,7 +1323,7 @@ static int32_t closeIntervalWindow(SHashObj* pHashMap, STimeWindowAggSupp* pSup,
TSKEY ts = *(int64_t*)((char*)key + sizeof(uint64_t));
SResultRowInfo dumyInfo;
dumyInfo.cur.pageId = -1;
- STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, ts, pInterval, pInterval->precision, TSDB_ORDER_ASC);
+ STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, ts, pInterval, TSDB_ORDER_ASC);
SWinRes winRe = {
.ts = win.skey,
.groupId = groupId,
@@ -2547,8 +2418,7 @@ static void doHashInterval(SOperatorInfo* pOperatorInfo, SSDataBlock* pSDataBloc
int32_t startPos = ascScan ? 0 : (pSDataBlock->info.rows - 1);
TSKEY ts = getStartTsKey(&pSDataBlock->info.window, tsCols);
- STimeWindow nextWin = getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts, &pInfo->interval,
- pInfo->interval.precision, pInfo->order);
+ STimeWindow nextWin = getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts, &pInfo->interval, pInfo->order);
while (1) {
bool isClosed = isCloseWindow(&nextWin, &pInfo->twAggSup);
if (pInfo->ignoreExpiredData && isClosed) {
@@ -4725,7 +4595,7 @@ static int32_t outputPrevIntervalResult(SOperatorInfo* pOperatorInfo, uint64_t t
continue;
}
STimeWindow* prevWin = &prevGrpWin->window;
- if ((ascScan && newWin->skey > prevWin->ekey || (!ascScan) && newWin->skey < prevWin->ekey)) {
+ if ((ascScan && newWin->skey > prevWin->ekey) || ((!ascScan) && newWin->skey < prevWin->ekey)) {
finalizeWindowResult(pOperatorInfo, tableGroupId, prevWin, pResultBlock);
tdListPopNode(miaInfo->groupIntervals, listNode);
}
@@ -4751,7 +4621,7 @@ static void doMergeIntervalAggImpl(SOperatorInfo* pOperatorInfo, SResultRowInfo*
SResultRow* pResult = NULL;
STimeWindow win = getActiveTimeWindow(iaInfo->aggSup.pResultBuf, pResultRowInfo, blockStartTs, &iaInfo->interval,
- iaInfo->interval.precision, iaInfo->order);
+ iaInfo->order);
int32_t ret =
setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, pExprSup->pCtx,
diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c
index 1706c129cef9d859d879de9128df975e21d1c5ee..932bfb8793ae933c52e38e71de4d5b38ee89a2d2 100644
--- a/source/libs/function/src/builtinsimpl.c
+++ b/source/libs/function/src/builtinsimpl.c
@@ -1863,8 +1863,6 @@ static void stddevTransferInfo(SStddevRes* pInput, SStddevRes* pOutput) {
}
pOutput->count += pInput->count;
-
- return;
}
int32_t stddevFunctionMerge(SqlFunctionCtx* pCtx) {
@@ -1874,14 +1872,13 @@ int32_t stddevFunctionMerge(SqlFunctionCtx* pCtx) {
SStddevRes* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
- int32_t start = pInput->startRowIndex;
- char* data = colDataGetData(pCol, start);
- SStddevRes* pInputInfo = (SStddevRes*)varDataVal(data);
-
- stddevTransferInfo(pInputInfo, pInfo);
+ for(int32_t i = pInput->startRowIndex; i < pInput->startRowIndex + pInput->numOfRows; ++i) {
+ char* data = colDataGetData(pCol, i);
+ SStddevRes* pInputInfo = (SStddevRes*)varDataVal(data);
+ stddevTransferInfo(pInputInfo, pInfo);
+ }
SET_VAL(GET_RES_INFO(pCtx), 1, 1);
-
return TSDB_CODE_SUCCESS;
}
diff --git a/source/util/src/tworker.c b/source/util/src/tworker.c
index 68f96c0385b6c25a4736343917e875f84d4e2c9e..88bd36f0cb0f048e301c93e28c2eba21de6ec36f 100644
--- a/source/util/src/tworker.c
+++ b/source/util/src/tworker.c
@@ -39,7 +39,7 @@ int32_t tQWorkerInit(SQWorkerPool *pool) {
worker->pool = pool;
}
- uDebug("worker:%s is initialized, min:%d max:%d", pool->name, pool->min, pool->max);
+ uInfo("worker:%s is initialized, min:%d max:%d", pool->name, pool->min, pool->max);
return 0;
}