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

fix bugs found in regression test.

This commit is contained in:
hjxilinx 2020-02-18 21:47:17 +08:00
parent 05e8b2ab7e
commit acc083b2b2
1 changed files with 121 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

130
src/system/detail/src/vnodeQueryImpl.c
View File

@ -74,7 +74,7 @@ static void validateTimestampForSupplementResult(SQueryRuntimeEnv *pRuntimeEn
static void getBasicCacheInfoSnapshot(SQuery *pQuery, SCacheInfo *pCacheInfo, int32_t vid); static void getBasicCacheInfoSnapshot(SQuery *pQuery, SCacheInfo *pCacheInfo, int32_t vid);
static TSKEY getQueryPositionForCacheInvalid(SQueryRuntimeEnv *pRuntimeEnv, __block_search_fn_t searchFn); static TSKEY getQueryPositionForCacheInvalid(SQueryRuntimeEnv *pRuntimeEnv, __block_search_fn_t searchFn);
static bool functionNeedToExecute(SQueryRuntimeEnv *pRuntimeEnv, SQLFunctionCtx *pCtx, int32_t functionId); static bool functionNeedToExecute(SQueryRuntimeEnv *pRuntimeEnv, SQLFunctionCtx *pCtx, int32_t functionId);
static void getNextTimeWindow(SQueryRuntimeEnv *pRuntimeEnv, STimeWindow *pTimeWindow); static void getNextTimeWindow(SQuery *pQuery, STimeWindow *pTimeWindow);
static int32_t getGroupResultId(int32_t groupIndex) { static int32_t getGroupResultId(int32_t groupIndex) {
int32_t base = 200000; int32_t base = 200000;
@ -1024,20 +1024,26 @@ SBlockInfo getBlockBasicInfo(SQueryRuntimeEnv *pRuntimeEnv, void *pBlock, int32_
* @return TRUE means query not completed, FALSE means query is completed * @return TRUE means query not completed, FALSE means query is completed
*/ */
static bool queryPausedInCurrentBlock(SQuery *pQuery, SBlockInfo *pBlockInfo, int32_t forwardStep) { static bool queryPausedInCurrentBlock(SQuery *pQuery, SBlockInfo *pBlockInfo, int32_t forwardStep) {
if (Q_STATUS_EQUAL(pQuery->over, QUERY_RESBUF_FULL)) {
// assert(pQuery->checkBufferInLoop == 1 && pQuery->over == QUERY_RESBUF_FULL && pQuery->pointsOffset == 0);
assert((QUERY_IS_ASC_QUERY(pQuery) && forwardStep + pQuery->pos <= pBlockInfo->size) ||
(!QUERY_IS_ASC_QUERY(pQuery) && pQuery->pos - forwardStep + 1 >= 0));
// current query completed // current query completed
if ((pQuery->lastKey > pQuery->ekey && QUERY_IS_ASC_QUERY(pQuery)) || if ((pQuery->lastKey > pQuery->ekey && QUERY_IS_ASC_QUERY(pQuery)) ||
(pQuery->lastKey < pQuery->ekey && !QUERY_IS_ASC_QUERY(pQuery))) { (pQuery->lastKey < pQuery->ekey && !QUERY_IS_ASC_QUERY(pQuery))) {
setQueryStatus(pQuery, QUERY_COMPLETED); setQueryStatus(pQuery, QUERY_COMPLETED);
return true;
} }
// output buffer is full, pause current query
if (Q_STATUS_EQUAL(pQuery->over, QUERY_RESBUF_FULL)) {
assert((QUERY_IS_ASC_QUERY(pQuery) && forwardStep + pQuery->pos <= pBlockInfo->size) ||
(!QUERY_IS_ASC_QUERY(pQuery) && pQuery->pos - forwardStep + 1 >= 0));
return true; return true;
} else { // query completed }
if (Q_STATUS_EQUAL(pQuery->over, QUERY_COMPLETED)) {
return true;
}
// query completed
if ((pQuery->ekey <= pBlockInfo->keyLast && QUERY_IS_ASC_QUERY(pQuery)) || if ((pQuery->ekey <= pBlockInfo->keyLast && QUERY_IS_ASC_QUERY(pQuery)) ||
(pQuery->ekey >= pBlockInfo->keyFirst && !QUERY_IS_ASC_QUERY(pQuery))) { (pQuery->ekey >= pBlockInfo->keyFirst && !QUERY_IS_ASC_QUERY(pQuery))) {
setQueryStatus(pQuery, QUERY_COMPLETED); setQueryStatus(pQuery, QUERY_COMPLETED);
@ -1045,7 +1051,6 @@ static bool queryPausedInCurrentBlock(SQuery *pQuery, SBlockInfo *pBlockInfo, in
} }
return false; return false;
}
} }
/** /**
@ -1242,9 +1247,9 @@ static void *getGenericDataBlock(SMeterObj *pMeterObj, SQueryRuntimeEnv *pRuntim
} }
} }
static SBlockInfo getBlockInfo(SQueryRuntimeEnv* pRuntimeEnv) { static SBlockInfo getBlockInfo(SQueryRuntimeEnv *pRuntimeEnv) {
SQuery *pQuery = pRuntimeEnv->pQuery; SQuery * pQuery = pRuntimeEnv->pQuery;
SMeterObj* pMeterObj = pRuntimeEnv->pMeterObj; SMeterObj *pMeterObj = pRuntimeEnv->pMeterObj;
void *pBlock = getGenericDataBlock(pMeterObj, pRuntimeEnv, pQuery->slot); void *pBlock = getGenericDataBlock(pMeterObj, pRuntimeEnv, pQuery->slot);
assert(pBlock != NULL); assert(pBlock != NULL);
@ -1253,7 +1258,6 @@ static SBlockInfo getBlockInfo(SQueryRuntimeEnv* pRuntimeEnv) {
return getBlockBasicInfo(pRuntimeEnv, pBlock, blockType); return getBlockBasicInfo(pRuntimeEnv, pBlock, blockType);
} }
static int32_t getFileIdFromKey(int32_t vid, TSKEY key) { static int32_t getFileIdFromKey(int32_t vid, TSKEY key) {
SVnodeObj *pVnode = &vnodeList[vid]; SVnodeObj *pVnode = &vnodeList[vid];
int64_t delta = (int64_t)pVnode->cfg.daysPerFile * tsMsPerDay[(uint8_t)pVnode->cfg.precision]; int64_t delta = (int64_t)pVnode->cfg.daysPerFile * tsMsPerDay[(uint8_t)pVnode->cfg.precision];
@ -1513,6 +1517,15 @@ static STimeWindow getActiveTimeWindow(SWindowResInfo *pWindowResInfo, int64_t t
} }
assert(ts >= w.skey && ts <= w.ekey && w.skey != 0); assert(ts >= w.skey && ts <= w.ekey && w.skey != 0);
// query border check
if (w.ekey > pQuery->ekey && QUERY_IS_ASC_QUERY(pQuery)) {
w.ekey = pQuery->ekey;
}
if (w.skey < pQuery->ekey && !QUERY_IS_ASC_QUERY(pQuery)) {
w.skey = pQuery->ekey;
}
return w; return w;
} }
@ -1557,7 +1570,7 @@ static int32_t addNewWindowResultBuf(SWindowResult *pWindowRes, SQueryDiskbasedR
static int32_t setWindowOutputBufByKey(SQueryRuntimeEnv *pRuntimeEnv, SWindowResInfo *pWindowResInfo, int32_t sid, static int32_t setWindowOutputBufByKey(SQueryRuntimeEnv *pRuntimeEnv, SWindowResInfo *pWindowResInfo, int32_t sid,
STimeWindow *win) { STimeWindow *win) {
assert(win->skey < win->ekey); assert(win->skey <= win->ekey);
SQueryDiskbasedResultBuf *pResultBuf = pRuntimeEnv->pResultBuf; SQueryDiskbasedResultBuf *pResultBuf = pRuntimeEnv->pResultBuf;
SWindowResult *pWindowRes = doSetTimeWindowFromKey(pRuntimeEnv, pWindowResInfo, (char *)&win->skey, TSDB_KEYSIZE); SWindowResult *pWindowRes = doSetTimeWindowFromKey(pRuntimeEnv, pWindowResInfo, (char *)&win->skey, TSDB_KEYSIZE);
@ -1661,47 +1674,47 @@ static int32_t getNumOfRowsInTimeWindow(SQuery *pQuery, SBlockInfo *pBlockInfo,
TSKEY ekey, __block_search_fn_t searchFn, bool updateLastKey) { TSKEY ekey, __block_search_fn_t searchFn, bool updateLastKey) {
assert(startPos >= 0 && startPos < pBlockInfo->size); assert(startPos >= 0 && startPos < pBlockInfo->size);
int32_t forwardStep = -1; int32_t num = -1;
int32_t order = pQuery->order.order; int32_t order = pQuery->order.order;
int32_t step = GET_FORWARD_DIRECTION_FACTOR(order); int32_t step = GET_FORWARD_DIRECTION_FACTOR(order);
if (QUERY_IS_ASC_QUERY(pQuery)) { if (QUERY_IS_ASC_QUERY(pQuery)) {
if (ekey < pBlockInfo->keyLast) { if (ekey < pBlockInfo->keyLast) {
forwardStep = getForwardStepsInBlock(pBlockInfo->size, searchFn, ekey, startPos, order, pPrimaryColumn); num = getForwardStepsInBlock(pBlockInfo->size, searchFn, ekey, startPos, order, pPrimaryColumn);
if (forwardStep == 0) { // no qualified data in current block, do not update the lastKey value if (num == 0) { // no qualified data in current block, do not update the lastKey value
assert(ekey < pPrimaryColumn[startPos]); assert(ekey < pPrimaryColumn[startPos]);
} else { } else {
if (updateLastKey) { if (updateLastKey) {
pQuery->lastKey = MAX(ekey, pPrimaryColumn[startPos + (forwardStep - 1)]) + step; pQuery->lastKey = MAX(ekey, pPrimaryColumn[startPos + (num - 1)]) + step;
} }
} }
} else { } else {
forwardStep = pBlockInfo->size - startPos; num = pBlockInfo->size - startPos;
if (updateLastKey) { if (updateLastKey) {
pQuery->lastKey = pBlockInfo->keyLast + step; pQuery->lastKey = pBlockInfo->keyLast + step;
} }
} }
} else { // desc } else { // desc
if (ekey > pBlockInfo->keyFirst) { if (ekey > pBlockInfo->keyFirst) {
forwardStep = getForwardStepsInBlock(pBlockInfo->size, searchFn, ekey, startPos, order, pPrimaryColumn); num = getForwardStepsInBlock(pBlockInfo->size, searchFn, ekey, startPos, order, pPrimaryColumn);
if (forwardStep == 0) { // no qualified data in current block, do not update the lastKey value if (num == 0) { // no qualified data in current block, do not update the lastKey value
assert(ekey > pPrimaryColumn[startPos]); assert(ekey > pPrimaryColumn[startPos]);
} else { } else {
if (updateLastKey) { if (updateLastKey) {
pQuery->lastKey = MIN(ekey, pPrimaryColumn[startPos - (forwardStep - 1)]) + step; pQuery->lastKey = MIN(ekey, pPrimaryColumn[startPos - (num - 1)]) + step;
} }
} }
} else { } else {
forwardStep = startPos + 1; num = startPos + 1;
if (updateLastKey) { if (updateLastKey) {
pQuery->lastKey = pBlockInfo->keyFirst + step; pQuery->lastKey = pBlockInfo->keyFirst + step;
} }
} }
} }
assert(forwardStep >= 0); assert(num >= 0);
return forwardStep; return num;
} }
static void doBlockwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SWindowStatus *pStatus, STimeWindow *pWin, static void doBlockwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SWindowStatus *pStatus, STimeWindow *pWin,
@ -1723,7 +1736,7 @@ static void doBlockwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SWindowStat
} }
} }
static void doRowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SWindowStatus *pStatus, STimeWindow *pWin) { static void doRowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SWindowStatus *pStatus, STimeWindow *pWin, int32_t offset) {
SQuery * pQuery = pRuntimeEnv->pQuery; SQuery * pQuery = pRuntimeEnv->pQuery;
SQLFunctionCtx *pCtx = pRuntimeEnv->pCtx; SQLFunctionCtx *pCtx = pRuntimeEnv->pCtx;
@ -1733,7 +1746,7 @@ static void doRowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SWindowStatus
int32_t functionId = pQuery->pSelectExpr[k].pBase.functionId; int32_t functionId = pQuery->pSelectExpr[k].pBase.functionId;
if (functionNeedToExecute(pRuntimeEnv, &pCtx[k], functionId)) { if (functionNeedToExecute(pRuntimeEnv, &pCtx[k], functionId)) {
aAggs[functionId].xFunction(&pCtx[k]); aAggs[functionId].xFunctionF(&pCtx[k], offset);
} }
} }
} }
@ -1745,14 +1758,14 @@ static int32_t getNextQualifiedWindow(SQueryRuntimeEnv *pRuntimeEnv, STimeWindow
SQuery *pQuery = pRuntimeEnv->pQuery; SQuery *pQuery = pRuntimeEnv->pQuery;
while (1) { while (1) {
getNextTimeWindow(pRuntimeEnv, pNextWin); getNextTimeWindow(pQuery, pNextWin);
if (pWindowResInfo->startTime > pNextWin->skey || (pNextWin->skey > pQuery->ekey && QUERY_IS_ASC_QUERY(pQuery)) || if (pWindowResInfo->startTime > pNextWin->skey || (pNextWin->skey > pQuery->ekey && QUERY_IS_ASC_QUERY(pQuery)) ||
(pNextWin->ekey < pQuery->ekey && !QUERY_IS_ASC_QUERY(pQuery))) { (pNextWin->ekey < pQuery->ekey && !QUERY_IS_ASC_QUERY(pQuery))) {
return -1; return -1;
} }
// next time window not in current block // next time window is not in current block
if ((pNextWin->skey > pBlockInfo->keyLast && QUERY_IS_ASC_QUERY(pQuery)) || if ((pNextWin->skey > pBlockInfo->keyLast && QUERY_IS_ASC_QUERY(pQuery)) ||
(pNextWin->ekey < pBlockInfo->keyFirst && !QUERY_IS_ASC_QUERY(pQuery))) { (pNextWin->ekey < pBlockInfo->keyFirst && !QUERY_IS_ASC_QUERY(pQuery))) {
return -1; return -1;
@ -1770,6 +1783,13 @@ static int32_t getNextQualifiedWindow(SQueryRuntimeEnv *pRuntimeEnv, STimeWindow
continue; continue;
} }
if (pNextWin->ekey > pQuery->ekey && QUERY_IS_ASC_QUERY(pQuery)) {
pNextWin->ekey = pQuery->ekey;
}
if (pNextWin->skey < pQuery->ekey && !QUERY_IS_ASC_QUERY(pQuery)) {
pNextWin->skey = pQuery->ekey;
}
return startPos; return startPos;
} }
} }
@ -2297,35 +2317,34 @@ static int32_t rowwiseApplyAllFunctions(SQueryRuntimeEnv *pRuntimeEnv, int32_t *
offset -= pCtx[0].startOffset; offset -= pCtx[0].startOffset;
SWindowStatus *pStatus = getTimeWindowResStatus(pWindowResInfo, curTimeWindow(pWindowResInfo)); SWindowStatus *pStatus = getTimeWindowResStatus(pWindowResInfo, curTimeWindow(pWindowResInfo));
doRowwiseApplyFunctions(pRuntimeEnv, pStatus, &win); doRowwiseApplyFunctions(pRuntimeEnv, pStatus, &win, offset);
lastKey = ts; lastKey = ts;
int32_t prev = pWindowResInfo->curIndex;
STimeWindow nextWin = win; STimeWindow nextWin = win;
int32_t index = pWindowResInfo->curIndex;
int32_t sid = pRuntimeEnv->pMeterObj->sid;
while (1) { while (1) {
getNextTimeWindow(pRuntimeEnv, &nextWin); getNextTimeWindow(pQuery, &nextWin);
if (pWindowResInfo->startTime > nextWin.skey || (nextWin.skey > pQuery->ekey && QUERY_IS_ASC_QUERY(pQuery)) || if (pWindowResInfo->startTime > nextWin.skey || (nextWin.skey > pQuery->ekey && QUERY_IS_ASC_QUERY(pQuery)) ||
(nextWin.skey > pQuery->skey && !QUERY_IS_ASC_QUERY(pQuery))) { (nextWin.skey > pQuery->skey && !QUERY_IS_ASC_QUERY(pQuery))) {
pWindowResInfo->curIndex = prev;
break; break;
} }
if (ts >= nextWin.skey && ts <= nextWin.ekey) { if (ts < nextWin.skey || ts > nextWin.ekey) {
break;
}
// null data, failed to allocate more memory buffer // null data, failed to allocate more memory buffer
if (setWindowOutputBufByKey(pRuntimeEnv, pWindowResInfo, pRuntimeEnv->pMeterObj->sid, &nextWin) != if (setWindowOutputBufByKey(pRuntimeEnv, pWindowResInfo, sid, &nextWin) != TSDB_CODE_SUCCESS) {
TSDB_CODE_SUCCESS) {
pWindowResInfo->curIndex = prev;
break; break;
} }
pStatus = getTimeWindowResStatus(pWindowResInfo, curTimeWindow(pWindowResInfo)); pStatus = getTimeWindowResStatus(pWindowResInfo, curTimeWindow(pWindowResInfo));
doRowwiseApplyFunctions(pRuntimeEnv, pStatus, &nextWin); doRowwiseApplyFunctions(pRuntimeEnv, pStatus, &nextWin, offset);
} else {
pWindowResInfo->curIndex = prev;
break;
}
} }
pWindowResInfo->curIndex = index;
} else { // other queries } else { // other queries
// decide which group this rows belongs to according to current state value // decide which group this rows belongs to according to current state value
if (groupbyStateValue) { if (groupbyStateValue) {
@ -2455,6 +2474,11 @@ static int32_t applyFunctionsOnBlock(SQueryRuntimeEnv *pRuntimeEnv, SBlockInfo *
TSKEY lastKey = (QUERY_IS_ASC_QUERY(pQuery)) ? pBlockInfo->keyLast : pBlockInfo->keyFirst; TSKEY lastKey = (QUERY_IS_ASC_QUERY(pQuery)) ? pBlockInfo->keyLast : pBlockInfo->keyFirst;
doCheckQueryCompleted(pRuntimeEnv, lastKey, pWindowResInfo); doCheckQueryCompleted(pRuntimeEnv, lastKey, pWindowResInfo);
// interval query with limit applied
if (pQuery->intervalTime > 0 && pQuery->limit.limit > 0 && pQuery->limit.limit <= numOfClosedTimeWindow(pWindowResInfo)) {
setQueryStatus(pQuery, QUERY_COMPLETED);
}
assert(*numOfRes >= 0); assert(*numOfRes >= 0);
// check if buffer is large enough for accommodating all qualified points // check if buffer is large enough for accommodating all qualified points
@ -4093,9 +4117,9 @@ static bool forwardQueryStartPosIfNeeded(SQInfo *pQInfo, STableQuerySupportObj *
if (pQuery->intervalTime > 0) { if (pQuery->intervalTime > 0) {
int16_t step = GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order); int16_t step = GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order);
__block_search_fn_t searchFn = vnodeSearchKeyFunc[pRuntimeEnv->pMeterObj->searchAlgorithm]; __block_search_fn_t searchFn = vnodeSearchKeyFunc[pRuntimeEnv->pMeterObj->searchAlgorithm];
SWindowResInfo* pWindowResInfo = &pRuntimeEnv->windowResInfo; SWindowResInfo * pWindowResInfo = &pRuntimeEnv->windowResInfo;
TSKEY* primaryKey = (TSKEY*) pRuntimeEnv->primaryColBuffer->data; TSKEY * primaryKey = (TSKEY *)pRuntimeEnv->primaryColBuffer->data;
STimeWindow win = getActiveTimeWindow(pWindowResInfo, pWindowResInfo->prevSKey, pQuery); STimeWindow win = getActiveTimeWindow(pWindowResInfo, pWindowResInfo->prevSKey, pQuery);
while (pQuery->limit.offset > 0) { while (pQuery->limit.offset > 0) {
@ -5188,12 +5212,12 @@ static void doHandleDataBlockImpl(SQueryRuntimeEnv *pRuntimeEnv, SBlockInfo *pbl
} }
} }
static void getNextTimeWindow(SQueryRuntimeEnv *pRuntimeEnv, STimeWindow *pTimeWindow) { // previous time window may not be of the same size of pQuery->intervalTime
SQuery *pQuery = pRuntimeEnv->pQuery; static void getNextTimeWindow(SQuery *pQuery, STimeWindow *pTimeWindow) {
int32_t factor = GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order); int32_t factor = GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order);
pTimeWindow->skey += (pQuery->slidingTime * factor); pTimeWindow->skey += (pQuery->slidingTime * factor);
pTimeWindow->ekey += (pQuery->slidingTime * factor); pTimeWindow->ekey = pTimeWindow->skey + (pQuery->intervalTime - 1);
} }
static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) { static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
@ -5245,16 +5269,18 @@ static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
int32_t nextPos = accessPos + step; int32_t nextPos = accessPos + step;
/* /*
* set the next access position, nextPos only required by * set the next access position, nextPos only required when the interval query and projection query
* 1. interval query. * that cause output buffer overflow. When the query is completed, no need to load the next block any more.
* 2. multi-output query that may cause buffer overflow.
*/ */
if (!Q_STATUS_EQUAL(pQuery->over, QUERY_COMPLETED) && Q_STATUS_EQUAL(pQuery->over, QUERY_RESBUF_FULL)) {
if (nextPos >= blockInfo.size || nextPos < 0) { if (nextPos >= blockInfo.size || nextPos < 0) {
moveToNextBlock(pRuntimeEnv, step, searchFn, !LOAD_DATA); moveToNextBlock(pRuntimeEnv, step, searchFn, !LOAD_DATA);
// slot/pos/fileId is updated in moveToNextBlock function // slot/pos/fileId is updated in moveToNextBlock function
savePointPosition(&pRuntimeEnv->nextPos, pQuery->fileId, pQuery->slot, pQuery->pos); savePointPosition(&pRuntimeEnv->nextPos, pQuery->fileId, pQuery->slot, pQuery->pos);
} else { } else {
savePointPosition(&pRuntimeEnv->nextPos, pQuery->fileId, pQuery->slot, accessPos + step); savePointPosition(&pRuntimeEnv->nextPos, pQuery->fileId, pQuery->slot, nextPos);
}
} }
break; break;
@ -7393,7 +7419,7 @@ static int32_t doCopyToSData(STableQuerySupportObj *pSupporter, SWindowResult *r
if (orderType == TSQL_SO_ASC) { if (orderType == TSQL_SO_ASC) {
startIdx = pSupporter->subgroupIdx; startIdx = pSupporter->subgroupIdx;
step = 1; step = 1;
} else {// desc order copy all data } else { // desc order copy all data
startIdx = totalSubset - pSupporter->subgroupIdx - 1; startIdx = totalSubset - pSupporter->subgroupIdx - 1;
step = -1; step = -1;
} }