提交 40380ef9 编写于 作者: H Haojun Liao

refactor: do internal refactor.

上级 c3d702fe
...@@ -155,8 +155,8 @@ typedef struct SQueryTableDataCond { ...@@ -155,8 +155,8 @@ typedef struct SQueryTableDataCond {
int32_t numOfCols; int32_t numOfCols;
SColumnInfo* colList; SColumnInfo* colList;
int32_t type; // data block load type: int32_t type; // data block load type:
int32_t numOfTWindows; // int32_t numOfTWindows;
STimeWindow* twindows; STimeWindow twindows;
int64_t startVersion; int64_t startVersion;
int64_t endVersion; int64_t endVersion;
} SQueryTableDataCond; } SQueryTableDataCond;
......
...@@ -133,7 +133,7 @@ bool tsdbNextDataBlock(STsdbReader *pReader); ...@@ -133,7 +133,7 @@ bool tsdbNextDataBlock(STsdbReader *pReader);
void tsdbRetrieveDataBlockInfo(STsdbReader *pReader, SDataBlockInfo *pDataBlockInfo); void tsdbRetrieveDataBlockInfo(STsdbReader *pReader, SDataBlockInfo *pDataBlockInfo);
int32_t tsdbRetrieveDatablockSMA(STsdbReader *pReader, SColumnDataAgg ***pBlockStatis, bool *allHave); int32_t tsdbRetrieveDatablockSMA(STsdbReader *pReader, SColumnDataAgg ***pBlockStatis, bool *allHave);
SArray *tsdbRetrieveDataBlock(STsdbReader *pTsdbReadHandle, SArray *pColumnIdList); 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); int32_t tsdbGetFileBlocksDistInfo(STsdbReader *pReader, STableBlockDistInfo *pTableBlockInfo);
int64_t tsdbGetNumOfRowsInMemTable(STsdbReader *pHandle); int64_t tsdbGetNumOfRowsInMemTable(STsdbReader *pHandle);
void *tsdbGetIdx(SMeta *pMeta); void *tsdbGetIdx(SMeta *pMeta);
......
...@@ -356,14 +356,14 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, STsd ...@@ -356,14 +356,14 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, STsd
initReaderStatus(&pReader->status); initReaderStatus(&pReader->status);
pReader->pTsdb = 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->suid = pCond->suid;
pReader->order = pCond->order; pReader->order = pCond->order;
pReader->capacity = 4096; pReader->capacity = 4096;
pReader->idStr = (idstr != NULL) ? strdup(idstr) : NULL; pReader->idStr = (idstr != NULL) ? strdup(idstr) : NULL;
pReader->verRange = getQueryVerRange(pVnode, pCond, level); pReader->verRange = getQueryVerRange(pVnode, pCond, level);
pReader->type = pCond->type; pReader->type = pCond->type;
pReader->window = updateQueryTimeWindow(pVnode->pTsdb, pCond->twindows); pReader->window = updateQueryTimeWindow(pVnode->pTsdb, &pCond->twindows);
ASSERT(pCond->numOfCols > 0); ASSERT(pCond->numOfCols > 0);
...@@ -2954,7 +2954,7 @@ SArray* tsdbRetrieveDataBlock(STsdbReader* pReader, SArray* pIdList) { ...@@ -2954,7 +2954,7 @@ SArray* tsdbRetrieveDataBlock(STsdbReader* pReader, SArray* pIdList) {
return pReader->pResBlock->pDataBlock; 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)) { if (isEmptyQueryTimeWindow(&pReader->window)) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
...@@ -2964,7 +2964,7 @@ int32_t tsdbReaderReset(STsdbReader* pReader, SQueryTableDataCond* pCond, int32_ ...@@ -2964,7 +2964,7 @@ int32_t tsdbReaderReset(STsdbReader* pReader, SQueryTableDataCond* pCond, int32_
pReader->status.loadFromFile = true; pReader->status.loadFromFile = true;
pReader->status.pTableIter = NULL; 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 // allocate buffer in order to load data blocks from file
memset(&pReader->suppInfo.tsColAgg, 0, sizeof(SColumnDataAgg)); memset(&pReader->suppInfo.tsColAgg, 0, sizeof(SColumnDataAgg));
......
...@@ -278,9 +278,6 @@ typedef struct STableScanInfo { ...@@ -278,9 +278,6 @@ typedef struct STableScanInfo {
SScanInfo scanInfo; SScanInfo scanInfo;
int32_t scanTimes; int32_t scanTimes;
SNode* pFilterNode; // filter info, which is push down by optimizer 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; SSDataBlock* pResBlock;
SArray* pColMatchInfo; SArray* pColMatchInfo;
...@@ -289,14 +286,10 @@ typedef struct STableScanInfo { ...@@ -289,14 +286,10 @@ typedef struct STableScanInfo {
int32_t scanFlag; // table scan flag to denote if it is a repeat/reverse/main scan int32_t scanFlag; // table scan flag to denote if it is a repeat/reverse/main scan
int32_t dataBlockLoadFlag; 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. 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 SSampleExecInfo sample; // sample execution info
int32_t curTWinIdx;
int32_t currentGroupId; int32_t currentGroupId;
int32_t currentTable; int32_t currentTable;
uint64_t queryId; // todo remove it
uint64_t taskId; // todo remove it
struct { struct {
uint64_t uid; uint64_t uid;
......
...@@ -13,6 +13,7 @@ ...@@ -13,6 +13,7 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#include <common/ttime.h>
#include "function.h" #include "function.h"
#include "functionMgt.h" #include "functionMgt.h"
#include "index.h" #include "index.h"
...@@ -769,12 +770,9 @@ int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysi ...@@ -769,12 +770,9 @@ int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysi
// pCond->twindow = pTableScanNode->scanRange; // pCond->twindow = pTableScanNode->scanRange;
// TODO: get it from stable scan node // TODO: get it from stable scan node
pCond->numOfTWindows = 1; pCond->twindows = pTableScanNode->scanRange;
pCond->twindows = taosMemoryCalloc(pCond->numOfTWindows, sizeof(STimeWindow)); pCond->suid = pTableScanNode->scan.suid;
pCond->twindows[0] = pTableScanNode->scanRange; pCond->type = BLOCK_LOAD_OFFSET_ORDER;
pCond->suid = pTableScanNode->scan.suid;
pCond->type = BLOCK_LOAD_OFFSET_ORDER;
pCond->startVersion = -1; pCond->startVersion = -1;
pCond->endVersion = -1; pCond->endVersion = -1;
// pCond->type = pTableScanNode->scanFlag; // pCond->type = pTableScanNode->scanFlag;
...@@ -826,3 +824,87 @@ int32_t convertFillType(int32_t mode) { ...@@ -826,3 +824,87 @@ int32_t convertFillType(int32_t mode) {
return type; 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
...@@ -1038,6 +1038,7 @@ static bool overlapWithTimeWindow(STaskAttr* pQueryAttr, SDataBlockInfo* pBlockI ...@@ -1038,6 +1038,7 @@ static bool overlapWithTimeWindow(STaskAttr* pQueryAttr, SDataBlockInfo* pBlockI
#endif #endif
static uint32_t doFilterByBlockTimeWindow(STableScanInfo* pTableScanInfo, SSDataBlock* pBlock) { static uint32_t doFilterByBlockTimeWindow(STableScanInfo* pTableScanInfo, SSDataBlock* pBlock) {
#if 0
SqlFunctionCtx* pCtx = pTableScanInfo->pCtx; SqlFunctionCtx* pCtx = pTableScanInfo->pCtx;
uint32_t status = BLK_DATA_NOT_LOAD; uint32_t status = BLK_DATA_NOT_LOAD;
...@@ -1059,6 +1060,8 @@ static uint32_t doFilterByBlockTimeWindow(STableScanInfo* pTableScanInfo, SSData ...@@ -1059,6 +1060,8 @@ static uint32_t doFilterByBlockTimeWindow(STableScanInfo* pTableScanInfo, SSData
} }
return status; return status;
#endif
return 0;
} }
int32_t loadDataBlockOnDemand(SExecTaskInfo* pTaskInfo, STableScanInfo* pTableScanInfo, SSDataBlock* pBlock, int32_t loadDataBlockOnDemand(SExecTaskInfo* pTaskInfo, STableScanInfo* pTableScanInfo, SSDataBlock* pBlock,
...@@ -2887,12 +2890,11 @@ int32_t doPrepareScan(SOperatorInfo* pOperator, uint64_t uid, int64_t ts) { ...@@ -2887,12 +2890,11 @@ int32_t doPrepareScan(SOperatorInfo* pOperator, uint64_t uid, int64_t ts) {
ASSERT(found); ASSERT(found);
tsdbSetTableId(pInfo->dataReader, uid); tsdbSetTableId(pInfo->dataReader, uid);
int64_t oldSkey = pInfo->cond.twindows[0].skey; int64_t oldSkey = pInfo->cond.twindows.skey;
pInfo->cond.twindows[0].skey = ts + 1; pInfo->cond.twindows.skey = ts + 1;
tsdbReaderReset(pInfo->dataReader, &pInfo->cond, 0); tsdbReaderReset(pInfo->dataReader, &pInfo->cond);
pInfo->cond.twindows[0].skey = oldSkey; pInfo->cond.twindows.skey = oldSkey;
pInfo->scanTimes = 0; pInfo->scanTimes = 0;
pInfo->curTWinIdx = 0;
qDebug("tsdb reader offset seek to uid %ld ts %ld, table cur set to %d , all table num %d", uid, ts, qDebug("tsdb reader offset seek to uid %ld ts %ld, table cur set to %d , all table num %d", uid, ts,
pInfo->currentTable, tableSz); pInfo->currentTable, tableSz);
...@@ -4347,9 +4349,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo ...@@ -4347,9 +4349,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
cond.colList->type = TSDB_DATA_TYPE_TIMESTAMP; cond.colList->type = TSDB_DATA_TYPE_TIMESTAMP;
cond.colList->bytes = sizeof(TSKEY); cond.colList->bytes = sizeof(TSKEY);
cond.numOfTWindows = 1; cond.twindows = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
cond.twindows = taosMemoryCalloc(1, sizeof(STimeWindow));
cond.twindows[0] = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
cond.suid = pBlockNode->suid; cond.suid = pBlockNode->suid;
cond.type = BLOCK_LOAD_OFFSET_ORDER; cond.type = BLOCK_LOAD_OFFSET_ORDER;
} }
......
...@@ -293,13 +293,8 @@ static void prepareForDescendingScan(STableScanInfo* pTableScanInfo, SqlFunction ...@@ -293,13 +293,8 @@ static void prepareForDescendingScan(STableScanInfo* pTableScanInfo, SqlFunction
// setupQueryRangeForReverseScan(pTableScanInfo); // setupQueryRangeForReverseScan(pTableScanInfo);
pTableScanInfo->cond.order = TSDB_ORDER_DESC; pTableScanInfo->cond.order = TSDB_ORDER_DESC;
for (int32_t i = 0; i < pTableScanInfo->cond.numOfTWindows; ++i) { STimeWindow* pTWindow = &pTableScanInfo->cond.twindows;
STimeWindow* pTWindow = &pTableScanInfo->cond.twindows[i]; TSWAP(pTWindow->skey, pTWindow->ekey);
TSWAP(pTWindow->skey, pTWindow->ekey);
}
SQueryTableDataCond* pCond = &pTableScanInfo->cond;
taosqsort(pCond->twindows, pCond->numOfTWindows, sizeof(STimeWindow), pCond, compareTimeWindow);
} }
int32_t addTagPseudoColumnData(SReadHandle* pHandle, SExprInfo* pPseudoExpr, int32_t numOfPseudoExpr, int32_t addTagPseudoColumnData(SReadHandle* pHandle, SExprInfo* pPseudoExpr, int32_t numOfPseudoExpr,
...@@ -446,16 +441,10 @@ static SSDataBlock* doTableScanGroup(SOperatorInfo* pOperator) { ...@@ -446,16 +441,10 @@ static SSDataBlock* doTableScanGroup(SOperatorInfo* pOperator) {
// do the ascending order traverse in the first place. // do the ascending order traverse in the first place.
while (pTableScanInfo->scanTimes < pTableScanInfo->scanInfo.numOfAsc) { while (pTableScanInfo->scanTimes < pTableScanInfo->scanInfo.numOfAsc) {
while (pTableScanInfo->curTWinIdx < pTableScanInfo->cond.numOfTWindows) { SSDataBlock* p = doTableScanImpl(pOperator);
SSDataBlock* p = doTableScanImpl(pOperator); if (p != NULL) {
if (p != NULL) { ASSERT(p->info.uid != 0);
ASSERT(p->info.uid != 0); return p;
return p;
}
pTableScanInfo->curTWinIdx += 1;
if (pTableScanInfo->curTWinIdx < pTableScanInfo->cond.numOfTWindows) {
tsdbReaderReset(pTableScanInfo->dataReader, &pTableScanInfo->cond, pTableScanInfo->curTWinIdx);
}
} }
pTableScanInfo->scanTimes += 1; pTableScanInfo->scanTimes += 1;
...@@ -464,40 +453,25 @@ static SSDataBlock* doTableScanGroup(SOperatorInfo* pOperator) { ...@@ -464,40 +453,25 @@ static SSDataBlock* doTableScanGroup(SOperatorInfo* pOperator) {
setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED); setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);
pTableScanInfo->scanFlag = REPEAT_SCAN; pTableScanInfo->scanFlag = REPEAT_SCAN;
qDebug("%s start to repeat ascending order scan data blocks due to query func required", GET_TASKID(pTaskInfo)); 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 // do prepare for the next round table scan operation
tsdbReaderReset(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0); tsdbReaderReset(pTableScanInfo->dataReader, &pTableScanInfo->cond);
pTableScanInfo->curTWinIdx = 0;
} }
} }
int32_t total = pTableScanInfo->scanInfo.numOfAsc + pTableScanInfo->scanInfo.numOfDesc; int32_t total = pTableScanInfo->scanInfo.numOfAsc + pTableScanInfo->scanInfo.numOfDesc;
if (pTableScanInfo->scanTimes < total) { if (pTableScanInfo->scanTimes < total) {
if (pTableScanInfo->cond.order == TSDB_ORDER_ASC) { if (pTableScanInfo->cond.order == TSDB_ORDER_ASC) {
prepareForDescendingScan(pTableScanInfo, pTableScanInfo->pCtx, 0); prepareForDescendingScan(pTableScanInfo, pOperator->exprSupp.pCtx, 0);
tsdbReaderReset(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0); tsdbReaderReset(pTableScanInfo->dataReader, &pTableScanInfo->cond);
pTableScanInfo->curTWinIdx = 0;
} }
qDebug("%s start to descending order scan data blocks due to query func required", GET_TASKID(pTaskInfo)); 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->scanTimes < total) {
while (pTableScanInfo->curTWinIdx < pTableScanInfo->cond.numOfTWindows) { SSDataBlock* p = doTableScanImpl(pOperator);
SSDataBlock* p = doTableScanImpl(pOperator); if (p != NULL) {
if (p != NULL) { return p;
return p;
}
pTableScanInfo->curTWinIdx += 1;
if (pTableScanInfo->curTWinIdx < pTableScanInfo->cond.numOfTWindows) {
tsdbReaderReset(pTableScanInfo->dataReader, &pTableScanInfo->cond, pTableScanInfo->curTWinIdx);
}
} }
pTableScanInfo->scanTimes += 1; pTableScanInfo->scanTimes += 1;
...@@ -508,12 +482,7 @@ static SSDataBlock* doTableScanGroup(SOperatorInfo* pOperator) { ...@@ -508,12 +482,7 @@ static SSDataBlock* doTableScanGroup(SOperatorInfo* pOperator) {
qDebug("%s start to repeat descending order scan data blocks due to query func required", qDebug("%s start to repeat descending order scan data blocks due to query func required",
GET_TASKID(pTaskInfo)); GET_TASKID(pTaskInfo));
for (int32_t i = 0; i < pTableScanInfo->cond.numOfTWindows; ++i) { tsdbReaderReset(pTableScanInfo->dataReader, &pTableScanInfo->cond);
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;
} }
} }
} }
...@@ -540,9 +509,8 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) { ...@@ -540,9 +509,8 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
} }
STableKeyInfo* pTableInfo = taosArrayGet(pTaskInfo->tableqinfoList.pTableList, pInfo->currentTable); STableKeyInfo* pTableInfo = taosArrayGet(pTaskInfo->tableqinfoList.pTableList, pInfo->currentTable);
tsdbSetTableId(pInfo->dataReader, pTableInfo->uid); tsdbSetTableId(pInfo->dataReader, pTableInfo->uid);
tsdbReaderReset(pInfo->dataReader, &pInfo->cond, 0); tsdbReaderReset(pInfo->dataReader, &pInfo->cond);
pInfo->scanTimes = 0; pInfo->scanTimes = 0;
pInfo->curTWinIdx = 0;
} }
} }
...@@ -574,8 +542,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) { ...@@ -574,8 +542,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
SArray* tableList = taosArrayGetP(pTaskInfo->tableqinfoList.pGroupList, pInfo->currentGroupId); SArray* tableList = taosArrayGetP(pTaskInfo->tableqinfoList.pGroupList, pInfo->currentGroupId);
// tsdbSetTableList(pInfo->dataReader, tableList); // tsdbSetTableList(pInfo->dataReader, tableList);
tsdbReaderReset(pInfo->dataReader, &pInfo->cond, 0); tsdbReaderReset(pInfo->dataReader, &pInfo->cond);
pInfo->curTWinIdx = 0;
pInfo->scanTimes = 0; pInfo->scanTimes = 0;
result = doTableScanGroup(pOperator); result = doTableScanGroup(pOperator);
...@@ -646,7 +613,6 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, ...@@ -646,7 +613,6 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
pInfo->pFilterNode = pTableScanNode->scan.node.pConditions; pInfo->pFilterNode = pTableScanNode->scan.node.pConditions;
pInfo->scanFlag = MAIN_SCAN; pInfo->scanFlag = MAIN_SCAN;
pInfo->pColMatchInfo = pColList; pInfo->pColMatchInfo = pColList;
pInfo->curTWinIdx = 0;
pInfo->currentGroupId = -1; pInfo->currentGroupId = -1;
pOperator->name = "TableScanOperator"; // for debug purpose pOperator->name = "TableScanOperator"; // for debug purpose
...@@ -875,12 +841,7 @@ static void setGroupId(SStreamScanInfo* pInfo, SSDataBlock* pBlock, int32_t grou ...@@ -875,12 +841,7 @@ static void setGroupId(SStreamScanInfo* pInfo, SSDataBlock* pBlock, int32_t grou
} }
void resetTableScanInfo(STableScanInfo* pTableScanInfo, STimeWindow* pWin) { void resetTableScanInfo(STableScanInfo* pTableScanInfo, STimeWindow* pWin) {
pTableScanInfo->cond.twindows[0] = *pWin; pTableScanInfo->cond.twindows = *pWin;
pTableScanInfo->curTWinIdx = 0;
// tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0);
// if (!pTableScanInfo->dataReader) {
// return false;
// }
pTableScanInfo->scanTimes = 0; pTableScanInfo->scanTimes = 0;
pTableScanInfo->currentGroupId = -1; pTableScanInfo->currentGroupId = -1;
} }
......
...@@ -59,92 +59,6 @@ static void doCloseWindow(SResultRowInfo* pResultRowInfo, const SIntervalAggOper ...@@ -59,92 +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 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;
}
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;
}
// 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 order) {
STimeWindow w = {0};
if (pResultRowInfo->cur.pageId == -1) { // the first window, from the previous stored value
getInitialStartTimeWindow(pInterval, pInterval->precision, 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;
}
static int32_t setTimeWindowOutputBuf(SResultRowInfo* pResultRowInfo, STimeWindow* win, bool masterscan, static int32_t setTimeWindowOutputBuf(SResultRowInfo* pResultRowInfo, STimeWindow* win, bool masterscan,
SResultRow** pResult, int64_t tableGroupId, SqlFunctionCtx* pCtx, SResultRow** pResult, int64_t tableGroupId, SqlFunctionCtx* pCtx,
int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup, int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup,
...@@ -4679,7 +4593,7 @@ static int32_t outputPrevIntervalResult(SOperatorInfo* pOperatorInfo, uint64_t t ...@@ -4679,7 +4593,7 @@ static int32_t outputPrevIntervalResult(SOperatorInfo* pOperatorInfo, uint64_t t
continue; continue;
} }
STimeWindow* prevWin = &prevGrpWin->window; 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); finalizeWindowResult(pOperatorInfo, tableGroupId, prevWin, pResultBlock);
tdListPopNode(miaInfo->groupIntervals, listNode); tdListPopNode(miaInfo->groupIntervals, listNode);
} }
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册