mirror of
https://github.com/taosdata/TDengine
synced 2026-05-24 10:09:01 +00:00
fac36002c1
* ci(stream): add ci and fix issue * ci: adj test * ci: add test and fix issue * ci: adj test * ci(stream): add test for fill operator and fix issue * fix: fix issue * fix: fix issue * fix: adjust defalut rec value
1093 lines
48 KiB
C
1093 lines
48 KiB
C
/*
|
|
* 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 "executorInt.h"
|
|
#include "filter.h"
|
|
#include "function.h"
|
|
#include "functionMgt.h"
|
|
#include "operator.h"
|
|
#include "querytask.h"
|
|
#include "streamexecutorInt.h"
|
|
#include "streaminterval.h"
|
|
#include "tchecksum.h"
|
|
#include "tcommon.h"
|
|
#include "tcompare.h"
|
|
#include "tdatablock.h"
|
|
#include "tfill.h"
|
|
#include "tglobal.h"
|
|
#include "tlog.h"
|
|
#include "ttime.h"
|
|
|
|
#define STREAM_INTERVAL_NONBLOCK_OP_STATE_NAME "StreamIntervalNonblockHistoryState"
|
|
|
|
// static int32_t buildWinResultKey(SRowBuffPos* pPos, SSHashObj* pUpdatedMap, SSHashObj* pResultCache, bool savePos) {
|
|
// int32_t code = TSDB_CODE_SUCCESS;
|
|
// int32_t lino = 0;
|
|
// SWinKey* pKey = pPos->pKey;
|
|
// SRowBuffPos* pPrevResPos = tSimpleHashGet(pResultCache, &pKey->groupId, sizeof(uint64_t));
|
|
// if (pPrevResPos != NULL) {
|
|
// SWinKey* pPrevResKey = (SWinKey*)pPrevResPos->pKey;
|
|
// if (savePos) {
|
|
// code = tSimpleHashPut(pUpdatedMap, pPrevResKey, sizeof(SWinKey), &pPrevResPos, POINTER_BYTES);
|
|
// } else {
|
|
// code = tSimpleHashPut(pUpdatedMap, pPrevResKey, sizeof(SWinKey), NULL, 0);
|
|
// }
|
|
// QUERY_CHECK_CODE(code, lino, _end);
|
|
// }
|
|
// code = tSimpleHashPut(pResultCache, &pKey->groupId, sizeof(uint64_t), &pPos, POINTER_BYTES);
|
|
// QUERY_CHECK_CODE(code, lino, _end);
|
|
|
|
// _end:
|
|
// if (code != TSDB_CODE_SUCCESS) {
|
|
// qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
|
|
// }
|
|
// return code;
|
|
// }
|
|
|
|
void releaseFlusedPos(void* pRes) {
|
|
SRowBuffPos* pPos = *(SRowBuffPos**)pRes;
|
|
if (pPos != NULL && pPos->needFree) {
|
|
pPos->beUsed = false;
|
|
}
|
|
}
|
|
|
|
void getStateKeepInfo(SNonBlockAggSupporter* pNbSup, bool isRecOp, int32_t* pNumRes, TSKEY* pTsRes) {
|
|
if (isRecOp) {
|
|
(*pNumRes) = 0;
|
|
(*pTsRes) = INT64_MIN;
|
|
} else {
|
|
(*pNumRes) = pNbSup->numOfKeep;
|
|
(*pTsRes) = pNbSup->tsOfKeep;
|
|
}
|
|
}
|
|
|
|
void streamIntervalNonblockReleaseState(SOperatorInfo* pOperator) {
|
|
SStreamIntervalSliceOperatorInfo* pInfo = pOperator->info;
|
|
SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
|
|
pAggSup->stateStore.streamStateClearExpiredState(pAggSup->pState, pInfo->nbSup.numOfKeep, pInfo->nbSup.tsOfKeep);
|
|
pAggSup->stateStore.streamStateCommit(pAggSup->pState);
|
|
int32_t resSize = sizeof(TSKEY);
|
|
pAggSup->stateStore.streamStateSaveInfo(pAggSup->pState, STREAM_INTERVAL_NONBLOCK_OP_STATE_NAME,
|
|
strlen(STREAM_INTERVAL_NONBLOCK_OP_STATE_NAME), &pInfo->twAggSup.maxTs,
|
|
resSize);
|
|
|
|
SOperatorInfo* downstream = pOperator->pDownstream[0];
|
|
if (downstream->fpSet.releaseStreamStateFn) {
|
|
downstream->fpSet.releaseStreamStateFn(downstream);
|
|
}
|
|
qDebug("%s===stream===streamIntervalNonblockReleaseState:%" PRId64, GET_TASKID(pOperator->pTaskInfo),
|
|
pInfo->twAggSup.maxTs);
|
|
}
|
|
|
|
void streamIntervalNonblockReloadState(SOperatorInfo* pOperator) {
|
|
int32_t code = TSDB_CODE_SUCCESS;
|
|
int32_t lino = 0;
|
|
SStreamIntervalSliceOperatorInfo* pInfo = pOperator->info;
|
|
SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
|
|
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
|
|
int32_t size = 0;
|
|
void* pBuf = NULL;
|
|
code = pAggSup->stateStore.streamStateGetInfo(pAggSup->pState, STREAM_INTERVAL_NONBLOCK_OP_STATE_NAME,
|
|
strlen(STREAM_INTERVAL_NONBLOCK_OP_STATE_NAME), &pBuf, &size);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
|
|
TSKEY ts = *(TSKEY*)pBuf;
|
|
taosMemoryFreeClear(pBuf);
|
|
pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, ts);
|
|
pAggSup->stateStore.streamStateReloadInfo(pAggSup->pState, ts);
|
|
|
|
SOperatorInfo* downstream = pOperator->pDownstream[0];
|
|
if (downstream->fpSet.reloadStreamStateFn) {
|
|
downstream->fpSet.reloadStreamStateFn(downstream);
|
|
}
|
|
qDebug("%s===stream===streamIntervalNonblockReloadState:%" PRId64, GET_TASKID(pOperator->pTaskInfo), ts);
|
|
|
|
_end:
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
|
|
}
|
|
}
|
|
|
|
int32_t saveRecWindowToDisc(SSessionKey* pWinKey, uint64_t uid, EStreamType mode, STableTsDataState* pTsDataState,
|
|
SStreamAggSupporter* pAggSup) {
|
|
int32_t len = copyRecDataToBuff(pWinKey->win.skey, pWinKey->win.ekey, uid, -1, mode, NULL, 0,
|
|
pTsDataState->pRecValueBuff, pTsDataState->recValueLen);
|
|
return pAggSup->stateStore.streamStateSessionSaveToDisk(pTsDataState, pWinKey, pTsDataState->pRecValueBuff, len);
|
|
}
|
|
|
|
static int32_t checkAndSaveWinStateToDisc(int32_t startIndex, SArray* pUpdated, uint64_t uid,
|
|
STableTsDataState* pTsDataState, SStreamAggSupporter* pAggSup,
|
|
SInterval* pInterval) {
|
|
int32_t code = TSDB_CODE_SUCCESS;
|
|
int32_t lino = 0;
|
|
int32_t mode = 0;
|
|
int32_t size = taosArrayGetSize(pUpdated);
|
|
for (int32_t i = startIndex; i < size; i++) {
|
|
SRowBuffPos* pWinPos = taosArrayGetP(pUpdated, i);
|
|
SWinKey* pKey = pWinPos->pKey;
|
|
int32_t winRes = pAggSup->stateStore.streamStateGetRecFlag(pAggSup->pState, pKey, sizeof(SWinKey), &mode);
|
|
if (winRes == TSDB_CODE_SUCCESS) {
|
|
SSessionKey winKey = {
|
|
.win.skey = pKey->ts, .win.ekey = taosTimeGetIntervalEnd(pKey->ts, pInterval), .groupId = pKey->groupId};
|
|
code = saveRecWindowToDisc(&winKey, uid, mode, pTsDataState, pAggSup);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
}
|
|
}
|
|
|
|
_end:
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
qError("%s failed at line %d since %s.", __func__, lino, tstrerror(code));
|
|
}
|
|
return code;
|
|
}
|
|
|
|
int32_t doStreamIntervalNonblockAggImpl(SOperatorInfo* pOperator, SSDataBlock* pBlock) {
|
|
int32_t code = TSDB_CODE_SUCCESS;
|
|
int32_t lino = 0;
|
|
SStreamIntervalSliceOperatorInfo* pInfo = (SStreamIntervalSliceOperatorInfo*)pOperator->info;
|
|
SResultRowInfo* pResultRowInfo = &(pInfo->binfo.resultRowInfo);
|
|
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
|
|
SExprSupp* pSup = &pOperator->exprSupp;
|
|
int32_t numOfOutput = pSup->numOfExprs;
|
|
TSKEY* tsCols = NULL;
|
|
int64_t groupId = pBlock->info.id.groupId;
|
|
SResultRow* pResult = NULL;
|
|
int32_t forwardRows = 0;
|
|
|
|
SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, pInfo->primaryTsIndex);
|
|
tsCols = (int64_t*)pColDataInfo->pData;
|
|
|
|
int32_t startPos = 0;
|
|
TSKEY curTs = getStartTsKey(&pBlock->info.window, tsCols);
|
|
SInervalSlicePoint curPoint = {0};
|
|
SInervalSlicePoint prevPoint = {0};
|
|
STimeWindow curWin = getActiveTimeWindow(NULL, pResultRowInfo, curTs, &pInfo->interval, TSDB_ORDER_ASC);
|
|
while (1) {
|
|
int32_t winCode = TSDB_CODE_SUCCESS;
|
|
code = getIntervalSliceCurStateBuf(&pInfo->streamAggSup, &pInfo->interval, pInfo->hasInterpoFunc, &curWin, groupId,
|
|
&curPoint, &prevPoint, &winCode);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
|
|
if (winCode != TSDB_CODE_SUCCESS && pInfo->hasInterpoFunc == false && pInfo->nbSup.numOfKeep == 1) {
|
|
SWinKey curKey = {.ts = curPoint.winKey.win.skey, .groupId = groupId};
|
|
code = getIntervalSlicePrevStateBuf(&pInfo->streamAggSup, &pInfo->interval, &curKey, &prevPoint);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
}
|
|
|
|
if (IS_VALID_WIN_KEY(prevPoint.winKey.win.skey) && winCode != TSDB_CODE_SUCCESS) {
|
|
if (pInfo->hasInterpoFunc && isInterpoWindowFinished(&prevPoint) == false) {
|
|
code = setIntervalSliceOutputBuf(&pInfo->streamAggSup, &prevPoint, pSup->pCtx, numOfOutput,
|
|
pSup->rowEntryInfoOffset);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
|
|
resetIntervalSliceFunctionKey(pSup->pCtx, numOfOutput);
|
|
doSetElapsedEndKey(prevPoint.winKey.win.ekey, &pOperator->exprSupp);
|
|
doStreamSliceInterpolation(prevPoint.pLastRow, prevPoint.winKey.win.ekey, curTs, pBlock, startPos,
|
|
&pOperator->exprSupp, INTERVAL_SLICE_END, pInfo->pOffsetInfo);
|
|
updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &prevPoint.winKey.win, 1);
|
|
code = applyAggFunctionOnPartialTuples(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, startPos, 0,
|
|
pBlock->info.rows, numOfOutput);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
setInterpoWindowFinished(&prevPoint);
|
|
}
|
|
|
|
if (pInfo->nbSup.numOfKeep == 1) {
|
|
void* pResPtr = taosArrayPush(pInfo->pUpdated, &prevPoint.pResPos);
|
|
QUERY_CHECK_NULL(pResPtr, code, lino, _end, terrno);
|
|
int32_t mode = 0;
|
|
SWinKey* pKey = prevPoint.pResPos->pKey;
|
|
int32_t winRes = pInfo->streamAggSup.stateStore.streamStateGetRecFlag(pInfo->streamAggSup.pState, pKey,
|
|
sizeof(SWinKey), &mode);
|
|
if (winRes == TSDB_CODE_SUCCESS) {
|
|
code = saveRecWindowToDisc(&prevPoint.winKey, pBlock->info.id.uid, mode, pInfo->basic.pTsDataState,
|
|
&pInfo->streamAggSup);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
}
|
|
} else {
|
|
SWinKey curKey = {.groupId = groupId};
|
|
curKey.ts = taosTimeAdd(curTs, -pInfo->interval.interval, pInfo->interval.intervalUnit,
|
|
pInfo->interval.precision, NULL) +
|
|
1;
|
|
int32_t startIndex = taosArrayGetSize(pInfo->pUpdated);
|
|
code = pInfo->streamAggSup.stateStore.streamStateGetAllPrev(pInfo->streamAggSup.pState, &curKey,
|
|
pInfo->pUpdated, pInfo->nbSup.numOfKeep);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
if (!isRecalculateOperator(&pInfo->basic)) {
|
|
code = checkAndSaveWinStateToDisc(startIndex, pInfo->pUpdated, 0, pInfo->basic.pTsDataState, &pInfo->streamAggSup, &pInfo->interval);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
}
|
|
}
|
|
}
|
|
|
|
code =
|
|
setIntervalSliceOutputBuf(&pInfo->streamAggSup, &curPoint, pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
|
|
resetIntervalSliceFunctionKey(pSup->pCtx, numOfOutput);
|
|
if (pInfo->hasInterpoFunc && IS_VALID_WIN_KEY(prevPoint.winKey.win.skey) && curPoint.winKey.win.skey != curTs) {
|
|
doStreamSliceInterpolation(prevPoint.pLastRow, curPoint.winKey.win.skey, curTs, pBlock, startPos,
|
|
&pOperator->exprSupp, INTERVAL_SLICE_START, pInfo->pOffsetInfo);
|
|
}
|
|
forwardRows = getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, curWin.ekey, binarySearchForKey, NULL,
|
|
TSDB_ORDER_ASC);
|
|
int32_t prevEndPos = (forwardRows - 1) + startPos;
|
|
if (pInfo->hasInterpoFunc) {
|
|
int32_t endRowId = getQualifiedRowNumDesc(pSup, pBlock, tsCols, prevEndPos, false);
|
|
TSKEY endRowTs = tsCols[endRowId];
|
|
transBlockToSliceResultRow(pBlock, endRowId, endRowTs, curPoint.pLastRow, 0, NULL, NULL, pInfo->pOffsetInfo);
|
|
}
|
|
|
|
updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &curPoint.winKey.win, 1);
|
|
code = applyAggFunctionOnPartialTuples(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, startPos,
|
|
forwardRows, pBlock->info.rows, numOfOutput);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
curPoint.pResPos->beUpdated = true;
|
|
|
|
if (curPoint.pLastRow->key == curPoint.winKey.win.ekey) {
|
|
setInterpoWindowFinished(&curPoint);
|
|
}
|
|
releaseOutputBuf(pInfo->streamAggSup.pState, curPoint.pResPos, &pInfo->streamAggSup.stateStore);
|
|
|
|
startPos = getNextQualifiedWindow(&pInfo->interval, &curWin, &pBlock->info, tsCols, prevEndPos, TSDB_ORDER_ASC);
|
|
if (startPos < 0) {
|
|
break;
|
|
}
|
|
curTs = tsCols[startPos];
|
|
}
|
|
|
|
_end:
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
|
|
}
|
|
return code;
|
|
}
|
|
|
|
int32_t getHistoryRemainResultInfo(SStreamAggSupporter* pAggSup, int32_t numOfState, SArray* pUpdated,
|
|
int32_t capacity) {
|
|
int32_t code = TSDB_CODE_SUCCESS;
|
|
int32_t lino = 0;
|
|
|
|
if (pAggSup->pCur == NULL) {
|
|
goto _end;
|
|
}
|
|
|
|
int32_t num = capacity - taosArrayGetSize(pUpdated);
|
|
for (int32_t i = 0; i < num; i++) {
|
|
int32_t winCode = pAggSup->stateStore.streamStateNLastStateGetKVByCur(pAggSup->pCur, numOfState, pUpdated);
|
|
if (winCode == TSDB_CODE_FAILED) {
|
|
pAggSup->stateStore.streamStateFreeCur(pAggSup->pCur);
|
|
pAggSup->pCur = NULL;
|
|
break;
|
|
}
|
|
|
|
pAggSup->stateStore.streamStateLastStateCurNext(pAggSup->pCur);
|
|
num = capacity - taosArrayGetSize(pUpdated);
|
|
}
|
|
|
|
_end:
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
qError("%s failed at line %d since %s.", __func__, lino, tstrerror(code));
|
|
}
|
|
return code;
|
|
}
|
|
|
|
int32_t buildIntervalHistoryResult(SOperatorInfo* pOperator) {
|
|
int32_t code = TSDB_CODE_SUCCESS;
|
|
int32_t lino = 0;
|
|
SStreamIntervalSliceOperatorInfo* pInfo = pOperator->info;
|
|
SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
|
|
SStreamNotifyEventSupp* pNotifySup = &pInfo->basic.notifyEventSup;
|
|
bool addNotifyEvent = false;
|
|
|
|
code = getHistoryRemainResultInfo(pAggSup, pInfo->nbSup.numOfKeep, pInfo->pUpdated, pOperator->resultInfo.capacity);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
if (taosArrayGetSize(pInfo->pUpdated) > 0) {
|
|
taosArraySort(pInfo->pUpdated, winPosCmprImpl);
|
|
if (pInfo->nbSup.numOfKeep > 1) {
|
|
taosArrayRemoveDuplicate(pInfo->pUpdated, winPosCmprImpl, releaseFlusedPos);
|
|
}
|
|
initMultiResInfoFromArrayList(&pInfo->groupResInfo, pInfo->pUpdated);
|
|
pInfo->pUpdated = taosArrayInit(1024, POINTER_BYTES);
|
|
QUERY_CHECK_NULL(pInfo->pUpdated, code, lino, _end, terrno);
|
|
|
|
doBuildStreamIntervalResult(pOperator, pInfo->streamAggSup.pState, pInfo->binfo.pRes, &pInfo->groupResInfo, addNotifyEvent ? pNotifySup->pSessionKeys : NULL);
|
|
}
|
|
|
|
_end:
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
qError("%s failed at line %d since %s.", __func__, lino, tstrerror(code));
|
|
}
|
|
return code;
|
|
}
|
|
|
|
static void removeDataDeleteResults(SArray* pUpdatedWins, SArray* pDelWins) {
|
|
int32_t size = taosArrayGetSize(pUpdatedWins);
|
|
int32_t delSize = taosArrayGetSize(pDelWins);
|
|
if (delSize == 0 || size == 0) {
|
|
return;
|
|
}
|
|
taosArraySort(pDelWins, winKeyCmprImpl);
|
|
taosArrayRemoveDuplicate(pDelWins, winKeyCmprImpl, NULL);
|
|
delSize = taosArrayGetSize(pDelWins);
|
|
|
|
for (int32_t i = 0; i < size; i++) {
|
|
SRowBuffPos* pPos = (SRowBuffPos*) taosArrayGetP(pUpdatedWins, i);
|
|
SWinKey* pResKey = (SWinKey*)pPos->pKey;
|
|
int32_t index = binarySearchCom(pDelWins, delSize, pResKey, TSDB_ORDER_DESC, compareWinKey);
|
|
if (index >= 0 && 0 == compareWinKey(pResKey, pDelWins, index)) {
|
|
taosArrayRemove(pDelWins, index);
|
|
delSize = taosArrayGetSize(pDelWins);
|
|
}
|
|
}
|
|
}
|
|
|
|
static int32_t doTransformRecalculateWindows(SExecTaskInfo* pTaskInfo, SInterval* pInterval, SSDataBlock* pBlock,
|
|
SArray* pUpWins) {
|
|
int32_t code = TSDB_CODE_SUCCESS;
|
|
int32_t lino = 0;
|
|
SColumnInfoData* pStartTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
|
|
TSKEY* startTsCols = (TSKEY*)pStartTsCol->pData;
|
|
SColumnInfoData* pEndTsCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX);
|
|
TSKEY* endTsCols = (TSKEY*)pEndTsCol->pData;
|
|
SColumnInfoData* pCalStTsCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);
|
|
TSKEY* calStTsCols = (TSKEY*)pCalStTsCol->pData;
|
|
SColumnInfoData* pCalEnTsCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);
|
|
TSKEY* calEnTsCols = (TSKEY*)pCalEnTsCol->pData;
|
|
SColumnInfoData* pGpCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
|
|
uint64_t* pGpDatas = (uint64_t*)pGpCol->pData;
|
|
for (int32_t i = 0; i < pBlock->info.rows; i++) {
|
|
STimeWindow win = {.skey = startTsCols[i], .ekey = endTsCols[i]};
|
|
do {
|
|
if (!inCalSlidingWindow(pInterval, &win, calStTsCols[i], calEnTsCols[i], pBlock->info.type)) {
|
|
getNextTimeWindow(pInterval, &win, TSDB_ORDER_ASC);
|
|
continue;
|
|
}
|
|
|
|
uint64_t winGpId = pGpDatas[i];
|
|
SWinKey winRes = {.ts = win.skey, .groupId = winGpId};
|
|
void* pTmp = taosArrayPush(pUpWins, &winRes);
|
|
QUERY_CHECK_NULL(pTmp, code, lino, _end, terrno);
|
|
getNextTimeWindow(pInterval, &win, TSDB_ORDER_ASC);
|
|
} while (win.ekey <= endTsCols[i]);
|
|
}
|
|
_end:
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
|
|
}
|
|
return code;
|
|
}
|
|
|
|
static int32_t buildOtherResult(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
|
|
int32_t code = TSDB_CODE_SUCCESS;
|
|
int32_t lino = 0;
|
|
SStreamIntervalSliceOperatorInfo* pInfo = pOperator->info;
|
|
SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
|
|
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
|
|
if (isHistoryOperator(&pInfo->basic) && isSingleOperator(&pInfo->basic)) {
|
|
code = buildIntervalHistoryResult(pOperator);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
if (pInfo->binfo.pRes->info.rows != 0) {
|
|
printDataBlock(pInfo->binfo.pRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
|
|
(*ppRes) = pInfo->binfo.pRes;
|
|
return code;
|
|
}
|
|
}
|
|
|
|
if (pInfo->recvCkBlock) {
|
|
pInfo->recvCkBlock = false;
|
|
printDataBlock(pInfo->pCheckpointRes, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
|
|
(*ppRes) = pInfo->pCheckpointRes;
|
|
return code;
|
|
}
|
|
|
|
if (pInfo->twAggSup.minTs != INT64_MAX) {
|
|
pInfo->nbSup.tsOfKeep = pInfo->twAggSup.minTs;
|
|
}
|
|
|
|
if ( !(isFinalOperator(&pInfo->basic) && (isRecalculateOperator(&pInfo->basic) || isHistoryOperator(&pInfo->basic))) ) {
|
|
int32_t numOfKeep = 0;
|
|
TSKEY tsOfKeep = INT64_MAX;
|
|
getStateKeepInfo(&pInfo->nbSup, isRecalculateOperator(&pInfo->basic), &numOfKeep, &tsOfKeep);
|
|
pAggSup->stateStore.streamStateClearExpiredState(pAggSup->pState, numOfKeep, tsOfKeep);
|
|
}
|
|
|
|
pInfo->twAggSup.minTs = INT64_MAX;
|
|
pInfo->basic.numOfRecv = 0;
|
|
setStreamOperatorCompleted(pOperator);
|
|
if (isFinalOperator(&pInfo->basic) && isRecalculateOperator(&pInfo->basic) && tSimpleHashGetSize(pInfo->nbSup.pPullDataMap) == 0) {
|
|
qInfo("===stream===%s recalculate is finished.", GET_TASKID(pTaskInfo));
|
|
pAggSup->stateStore.streamStateClearExpiredState(pAggSup->pState, 0, INT64_MAX);
|
|
pTaskInfo->streamInfo.recoverScanFinished = true;
|
|
}
|
|
(*ppRes) = NULL;
|
|
|
|
_end:
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
|
|
}
|
|
return code;
|
|
}
|
|
|
|
int32_t copyNewResult(SSHashObj** ppWinUpdated, SArray* pUpdated, __compar_fn_t compar) {
|
|
int32_t code = TSDB_CODE_SUCCESS;
|
|
int32_t lino = 0;
|
|
void* pIte = NULL;
|
|
int32_t iter = 0;
|
|
while ((pIte = tSimpleHashIterate(*ppWinUpdated, pIte, &iter)) != NULL) {
|
|
void* tmp = taosArrayPush(pUpdated, pIte);
|
|
QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
|
|
}
|
|
taosArraySort(pUpdated, compar);
|
|
if (tSimpleHashGetSize(*ppWinUpdated) < 4096) {
|
|
tSimpleHashClear(*ppWinUpdated);
|
|
} else {
|
|
tSimpleHashClear(*ppWinUpdated);
|
|
tSimpleHashCleanup(*ppWinUpdated);
|
|
_hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
|
|
(*ppWinUpdated) = tSimpleHashInit(1024, hashFn);
|
|
}
|
|
|
|
_end:
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
|
|
}
|
|
return code;
|
|
}
|
|
|
|
static int32_t closeNonblockIntervalWindow(SSHashObj* pHashMap, STimeWindowAggSupp* pTwSup, SInterval* pInterval,
|
|
SArray* pUpdated, SExecTaskInfo* pTaskInfo) {
|
|
int32_t code = TSDB_CODE_SUCCESS;
|
|
int32_t lino = 0;
|
|
void* pIte = NULL;
|
|
int32_t iter = 0;
|
|
while ((pIte = tSimpleHashIterate(pHashMap, pIte, &iter)) != NULL) {
|
|
void* key = tSimpleHashGetKey(pIte, NULL);
|
|
SWinKey* pWinKey = (SWinKey*)key;
|
|
|
|
STimeWindow win = {
|
|
.skey = pWinKey->ts,
|
|
.ekey = taosTimeAdd(win.skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision, NULL) - 1,
|
|
};
|
|
|
|
if (isCloseWindow(&win, pTwSup)) {
|
|
void* pTemp = taosArrayPush(pUpdated, pIte);
|
|
QUERY_CHECK_NULL(pTemp, code, lino, _end, terrno);
|
|
|
|
int32_t tmpRes = tSimpleHashIterateRemove(pHashMap, pWinKey, sizeof(SWinKey), &pIte, &iter);
|
|
qTrace("%s at line %d res:%d", __func__, __LINE__, tmpRes);
|
|
}
|
|
}
|
|
|
|
_end:
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
|
|
}
|
|
return code;
|
|
}
|
|
|
|
static int32_t doProcessRecalculateReq(SOperatorInfo* pOperator, SSDataBlock* pBlock) {
|
|
int32_t code = TSDB_CODE_SUCCESS;
|
|
int32_t lino = 0;
|
|
SStreamIntervalSliceOperatorInfo* pInfo = pOperator->info;
|
|
SInterval* pInterval = &pInfo->interval;
|
|
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
|
|
SColumnInfoData* pStartTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
|
|
TSKEY* startTsCols = (TSKEY*)pStartTsCol->pData;
|
|
SColumnInfoData* pEndTsCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX);
|
|
TSKEY* endTsCols = (TSKEY*)pEndTsCol->pData;
|
|
SColumnInfoData* pCalStTsCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);
|
|
TSKEY* calStTsCols = (TSKEY*)pCalStTsCol->pData;
|
|
SColumnInfoData* pCalEnTsCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);
|
|
TSKEY* calEnTsCols = (TSKEY*)pCalEnTsCol->pData;
|
|
SColumnInfoData* pGpCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
|
|
uint64_t* pGpDatas = (uint64_t*)pGpCol->pData;
|
|
SColumnInfoData* pUidCol = taosArrayGet(pBlock->pDataBlock, UID_COLUMN_INDEX);
|
|
uint64_t* pUidDatas = (uint64_t*)pUidCol->pData;
|
|
|
|
for (int32_t i = 0; i < pBlock->info.rows; i++) {
|
|
SResultRowInfo dumyInfo = {0};
|
|
dumyInfo.cur.pageId = -1;
|
|
|
|
STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, startTsCols[i], pInterval, TSDB_ORDER_ASC);
|
|
|
|
do {
|
|
if (!inCalSlidingWindow(pInterval, &win, calStTsCols[i], calEnTsCols[i], pBlock->info.type)) {
|
|
getNextTimeWindow(pInterval, &win, TSDB_ORDER_ASC);
|
|
continue;
|
|
}
|
|
|
|
SWinKey key = {.ts = win.skey, .groupId = pGpDatas[i]};
|
|
bool isLastWin = false;
|
|
if (pInfo->streamAggSup.stateStore.streamStateCheck(pInfo->streamAggSup.pState, &key,
|
|
isFinalOperator(&pInfo->basic), &isLastWin)) {
|
|
qDebug("===stream===%s set recalculate flag ts:%" PRId64 ",group id:%" PRIu64, GET_TASKID(pTaskInfo), key.ts,
|
|
key.groupId);
|
|
pInfo->streamAggSup.stateStore.streamStateSetRecFlag(pInfo->streamAggSup.pState, &key, sizeof(SWinKey),
|
|
pBlock->info.type);
|
|
if ((isFinalOperator(&pInfo->basic) && isCloseWindow(&win, &pInfo->twAggSup)) || (isSingleOperator(&pInfo->basic) && isLastWin == false) ) {
|
|
SSessionKey winKey = {.win = win, .groupId = key.groupId};
|
|
code = saveRecWindowToDisc(&winKey, pUidDatas[i], pBlock->info.type, pInfo->basic.pTsDataState,
|
|
&pInfo->streamAggSup);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
}
|
|
} else {
|
|
SSessionKey winKey = {.win = win, .groupId = key.groupId};
|
|
code = saveRecWindowToDisc(&winKey, pUidDatas[i], pBlock->info.type, pInfo->basic.pTsDataState,
|
|
&pInfo->streamAggSup);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
}
|
|
getNextTimeWindow(pInterval, &win, TSDB_ORDER_ASC);
|
|
} while (win.ekey <= endTsCols[i]);
|
|
}
|
|
_end:
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
|
|
}
|
|
return code;
|
|
}
|
|
|
|
static int32_t addDataPullWindowInfo(SSHashObj* pPullMap, SPullWindowInfo* pPullKey, int32_t numOfChild, SExecTaskInfo* pTaskInfo) {
|
|
int32_t code = TSDB_CODE_SUCCESS;
|
|
int32_t lino = 0;
|
|
SArray* childIds = taosArrayInit(numOfChild, sizeof(int32_t));
|
|
QUERY_CHECK_NULL(childIds, code, lino, _end, terrno);
|
|
for (int32_t i = 0; i < numOfChild; i++) {
|
|
void* pTemp = taosArrayPush(childIds, &i);
|
|
QUERY_CHECK_NULL(pTemp, code, lino, _end, terrno);
|
|
}
|
|
code = tSimpleHashPut(pPullMap, pPullKey, sizeof(SPullWindowInfo), &childIds, POINTER_BYTES);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
_end:
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
qError("%s %s failed at line %d since %s", GET_TASKID(pTaskInfo), __func__, lino, tstrerror(code));
|
|
}
|
|
return code;
|
|
}
|
|
|
|
int32_t buildRetriveRequest(SExecTaskInfo* pTaskInfo, SStreamAggSupporter* pAggSup, STableTsDataState* pTsDataState,
|
|
SNonBlockAggSupporter* pNbSup) {
|
|
int32_t code = TSDB_CODE_SUCCESS;
|
|
int32_t lino = 0;
|
|
code = pAggSup->stateStore.streamStateMergeAllScanRange(pTsDataState);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
|
|
while (1) {
|
|
SScanRange range = {0};
|
|
code = pAggSup->stateStore.streamStatePopScanRange(pTsDataState, &range);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
if (IS_INVALID_RANGE(range)) {
|
|
break;
|
|
}
|
|
void* pIte = NULL;
|
|
int32_t iter = 0;
|
|
while ((pIte = tSimpleHashIterate(range.pGroupIds, pIte, &iter)) != NULL) {
|
|
uint64_t groupId = *(uint64_t*)tSimpleHashGetKey(pIte, NULL);
|
|
SPullWindowInfo pullReq = {.window = range.win, .groupId = groupId, .calWin = range.calWin};
|
|
code = addDataPullWindowInfo(pNbSup->pPullDataMap, &pullReq, pNbSup->numOfChild, pTaskInfo);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
void* pTemp = taosArrayPush(pNbSup->pPullWins, &pullReq);
|
|
QUERY_CHECK_NULL(pTemp, code, lino, _end, terrno);
|
|
}
|
|
}
|
|
|
|
_end:
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
|
|
pTaskInfo->code = code;
|
|
}
|
|
return code;
|
|
}
|
|
|
|
int32_t processDataPullOver(SSDataBlock* pBlock, SSHashObj* pPullMap, SExecTaskInfo* pTaskInfo) {
|
|
int32_t code = TSDB_CODE_SUCCESS;
|
|
int32_t lino = 0;
|
|
SColumnInfoData* pStartCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
|
|
TSKEY* pTsStartData = (TSKEY*)pStartCol->pData;
|
|
SColumnInfoData* pEndCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX);
|
|
TSKEY* pTsEndData = (TSKEY*)pEndCol->pData;
|
|
SColumnInfoData* pCalStartCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);
|
|
TSKEY* pCalTsStartData = (TSKEY*)pCalStartCol->pData;
|
|
SColumnInfoData* pCalEndCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);
|
|
TSKEY* pCalTsEndData = (TSKEY*)pCalEndCol->pData;
|
|
SColumnInfoData* pGroupCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
|
|
uint64_t* pGroupIdData = (uint64_t*)pGroupCol->pData;
|
|
int32_t chId = getChildIndex(pBlock);
|
|
for (int32_t i = 0; i < pBlock->info.rows; i++) {
|
|
SPullWindowInfo pull = {.window.skey = pTsStartData[i],
|
|
.window.ekey = pTsEndData[i],
|
|
.groupId = pGroupIdData[i],
|
|
.calWin.skey = pCalTsStartData[i],
|
|
.calWin.ekey = pCalTsEndData[i]};
|
|
void* pChIds = tSimpleHashGet(pPullMap, &pull, sizeof(SPullWindowInfo));
|
|
if (pChIds == NULL) {
|
|
qInfo("===stream===%s did not find retrive window. ts:%" PRId64 ",groupId:%" PRIu64 ",child id %d",
|
|
GET_TASKID(pTaskInfo), pull.window.skey, pull.groupId, chId);
|
|
continue;
|
|
}
|
|
SArray* chArray = *(SArray**)pChIds;
|
|
int32_t index = taosArraySearchIdx(chArray, &chId, compareInt32Val, TD_EQ);
|
|
if (index == -1) {
|
|
qInfo("===stream===%s did not find child id. retrive window ts:%" PRId64 ",groupId:%" PRIu64 ",child id %d",
|
|
GET_TASKID(pTaskInfo), pull.window.skey, pull.groupId, chId);
|
|
continue;
|
|
}
|
|
qDebug("===stream===%s retrive window %" PRId64 " delete child id %d", GET_TASKID(pTaskInfo), pull.window.skey,
|
|
chId);
|
|
taosArrayRemove(chArray, index);
|
|
if (taosArrayGetSize(chArray) == 0) {
|
|
// pull data is over
|
|
taosArrayDestroy(chArray);
|
|
int32_t tmpRes = tSimpleHashRemove(pPullMap, &pull, sizeof(SPullWindowInfo));
|
|
qDebug("===stream===%s retrive pull data over. ts:%" PRId64 ",groupId:%" PRIu64 , GET_TASKID(pTaskInfo), pull.window.skey,
|
|
pull.groupId);
|
|
}
|
|
}
|
|
|
|
_end:
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
|
|
}
|
|
return code;
|
|
}
|
|
|
|
int32_t doStreamIntervalNonblockAggNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
|
|
int32_t code = TSDB_CODE_SUCCESS;
|
|
int32_t lino = 0;
|
|
SStreamIntervalSliceOperatorInfo* pInfo = pOperator->info;
|
|
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
|
|
SStorageAPI* pAPI = &pOperator->pTaskInfo->storageAPI;
|
|
SExprSupp* pSup = &pOperator->exprSupp;
|
|
SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
|
|
|
|
qDebug("stask:%s %s status: %d", GET_TASKID(pTaskInfo), getStreamOpName(pOperator->operatorType), pOperator->status);
|
|
|
|
if (pOperator->status == OP_EXEC_DONE) {
|
|
(*ppRes) = NULL;
|
|
return code;
|
|
}
|
|
|
|
code = buildIntervalSliceResult(pOperator, ppRes);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
if ((*ppRes) != NULL) {
|
|
return code;
|
|
}
|
|
|
|
if (isHistoryOperator(&pInfo->basic) && !isFinalOperator(&pInfo->basic)) {
|
|
pAggSup->stateStore.streamStateClearExpiredState(pAggSup->pState, pInfo->nbSup.numOfKeep, pInfo->nbSup.tsOfKeep);
|
|
}
|
|
|
|
if (pOperator->status == OP_RES_TO_RETURN) {
|
|
return buildOtherResult(pOperator, ppRes);
|
|
}
|
|
|
|
SOperatorInfo* downstream = pOperator->pDownstream[0];
|
|
|
|
while (1) {
|
|
if (isTaskKilled(pTaskInfo)) {
|
|
qInfo("===stream===%s task is killed, code %s", GET_TASKID(pTaskInfo), tstrerror(pTaskInfo->code));
|
|
(*ppRes) = NULL;
|
|
return code;
|
|
}
|
|
SSDataBlock* pBlock = NULL;
|
|
code = downstream->fpSet.getNextFn(downstream, &pBlock);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
|
|
if (pBlock == NULL) {
|
|
qDebug("===stream===%s return data:%s. rev rows:%d", GET_TASKID(pTaskInfo),
|
|
getStreamOpName(pOperator->operatorType), pInfo->basic.numOfRecv);
|
|
if (isFinalOperator(&pInfo->basic)) {
|
|
if (isRecalculateOperator(&pInfo->basic)) {
|
|
code = buildRetriveRequest(pTaskInfo, pAggSup, pInfo->basic.pTsDataState, &pInfo->nbSup);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
} else {
|
|
code = pAggSup->stateStore.streamStateFlushReaminInfoToDisk(pInfo->basic.pTsDataState);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
}
|
|
}
|
|
pOperator->status = OP_RES_TO_RETURN;
|
|
break;
|
|
}
|
|
pInfo->basic.numOfRecv += pBlock->info.rows;
|
|
|
|
printSpecDataBlock(pBlock, getStreamOpName(pOperator->operatorType), "recv", GET_TASKID(pTaskInfo));
|
|
setStreamOperatorState(&pInfo->basic, pBlock->info.type);
|
|
|
|
switch (pBlock->info.type) {
|
|
case STREAM_NORMAL:
|
|
case STREAM_INVALID:
|
|
case STREAM_PULL_DATA: {
|
|
SExprSupp* pExprSup = &pInfo->scalarSup;
|
|
if (pExprSup->pExprInfo != NULL) {
|
|
code = projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
}
|
|
} break;
|
|
case STREAM_CHECKPOINT: {
|
|
pInfo->recvCkBlock = true;
|
|
pAggSup->stateStore.streamStateCommit(pAggSup->pState);
|
|
code = copyDataBlock(pInfo->pCheckpointRes, pBlock);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
continue;
|
|
} break;
|
|
case STREAM_CREATE_CHILD_TABLE:
|
|
case STREAM_DROP_CHILD_TABLE: {
|
|
printDataBlock(pBlock, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
|
|
(*ppRes) = pBlock;
|
|
return code;
|
|
} break;
|
|
case STREAM_RECALCULATE_DATA:
|
|
case STREAM_RECALCULATE_DELETE: {
|
|
if (isRecalculateOperator(&pInfo->basic)) {
|
|
if (!isSemiOperator(&pInfo->basic)) {
|
|
code = doTransformRecalculateWindows(pTaskInfo, &pInfo->interval, pBlock, pInfo->pDelWins);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
if (isFinalOperator(&pInfo->basic)) {
|
|
saveRecalculateData(&pAggSup->stateStore, pInfo->basic.pTsDataState, pBlock, pBlock->info.type);
|
|
}
|
|
continue;
|
|
}
|
|
}
|
|
|
|
if (isSemiOperator(&pInfo->basic)) {
|
|
(*ppRes) = pBlock;
|
|
return code;
|
|
} else {
|
|
code = doProcessRecalculateReq(pOperator, pBlock);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
}
|
|
continue;
|
|
} break;
|
|
case STREAM_PULL_OVER: {
|
|
code = processDataPullOver(pBlock, pInfo->nbSup.pPullDataMap, pTaskInfo);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
continue;
|
|
}
|
|
default: {
|
|
qDebug("===stream===%s ignore recv block. type:%d", GET_TASKID(pTaskInfo), pBlock->info.type);
|
|
continue;
|
|
} break;
|
|
}
|
|
|
|
if (pBlock->info.type == STREAM_NORMAL && pBlock->info.version != 0) {
|
|
// set input version
|
|
pTaskInfo->version = pBlock->info.version;
|
|
}
|
|
|
|
code = setInputDataBlock(pSup, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
|
|
pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.window.ekey);
|
|
code = pInfo->nbSup.pWindowAggFn(pOperator, pBlock);
|
|
if (code == TSDB_CODE_STREAM_INTERNAL_ERROR) {
|
|
pOperator->status = OP_RES_TO_RETURN;
|
|
code = TSDB_CODE_SUCCESS;
|
|
}
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
|
|
if (pAggSup->pScanBlock->info.rows > 0) {
|
|
(*ppRes) = pAggSup->pScanBlock;
|
|
printDataBlock(pAggSup->pScanBlock, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
|
|
return code;
|
|
}
|
|
|
|
if (taosArrayGetSize(pInfo->pUpdated) > 0) {
|
|
break;
|
|
}
|
|
}
|
|
|
|
if (pOperator->status == OP_RES_TO_RETURN &&
|
|
(isHistoryOperator(&pInfo->basic) || isRecalculateOperator(&pInfo->basic) || isSemiOperator(&pInfo->basic))) {
|
|
code = copyNewResult(&pAggSup->pResultRows, pInfo->pUpdated, winPosCmprImpl);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
|
|
if (isSingleOperator(&pInfo->basic)) {
|
|
if (pAggSup->pCur == NULL) {
|
|
pAggSup->pCur = pAggSup->stateStore.streamStateGetLastStateCur(pAggSup->pState);
|
|
}
|
|
code =
|
|
getHistoryRemainResultInfo(pAggSup, pInfo->nbSup.numOfKeep, pInfo->pUpdated, pOperator->resultInfo.capacity);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
}
|
|
}
|
|
|
|
if (pOperator->status == OP_RES_TO_RETURN && pInfo->destHasPrimaryKey && isFinalOperator(&pInfo->basic)) {
|
|
code = closeNonblockIntervalWindow(pAggSup->pResultRows, &pInfo->twAggSup, &pInfo->interval, pInfo->pUpdated,
|
|
pTaskInfo);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
if (!isHistoryOperator(&pInfo->basic) && !isRecalculateOperator(&pInfo->basic)) {
|
|
code = checkAndSaveWinStateToDisc(0, pInfo->pUpdated, 0, pInfo->basic.pTsDataState, &pInfo->streamAggSup, &pInfo->interval);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
}
|
|
}
|
|
|
|
taosArraySort(pInfo->pUpdated, winPosCmprImpl);
|
|
if (pInfo->nbSup.numOfKeep > 1) {
|
|
taosArrayRemoveDuplicate(pInfo->pUpdated, winPosCmprImpl, releaseFlusedPos);
|
|
}
|
|
if (!isSemiOperator(&pInfo->basic) && !pInfo->destHasPrimaryKey) {
|
|
removeDataDeleteResults(pInfo->pUpdated, pInfo->pDelWins);
|
|
}
|
|
|
|
initMultiResInfoFromArrayList(&pInfo->groupResInfo, pInfo->pUpdated);
|
|
pInfo->pUpdated = taosArrayInit(1024, POINTER_BYTES);
|
|
QUERY_CHECK_NULL(pInfo->pUpdated, code, lino, _end, terrno);
|
|
|
|
code = blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
|
|
code = buildIntervalSliceResult(pOperator, ppRes);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
if ((*ppRes) != NULL) {
|
|
return code;
|
|
}
|
|
|
|
return buildOtherResult(pOperator, ppRes);
|
|
|
|
_end:
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
|
|
pTaskInfo->code = code;
|
|
}
|
|
return code;
|
|
}
|
|
|
|
int32_t doStreamSemiIntervalNonblockAggImpl(SOperatorInfo* pOperator, SSDataBlock* pBlock) {
|
|
int32_t code = TSDB_CODE_SUCCESS;
|
|
int32_t lino = 0;
|
|
SStreamIntervalSliceOperatorInfo* pInfo = (SStreamIntervalSliceOperatorInfo*)pOperator->info;
|
|
SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
|
|
SResultRowInfo* pResultRowInfo = &(pInfo->binfo.resultRowInfo);
|
|
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
|
|
SExprSupp* pSup = &pOperator->exprSupp;
|
|
int32_t numOfOutput = pSup->numOfExprs;
|
|
TSKEY* tsCols = NULL;
|
|
int64_t groupId = pBlock->info.id.groupId;
|
|
SResultRow* pResult = NULL;
|
|
int32_t forwardRows = 0;
|
|
|
|
SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, pInfo->primaryTsIndex);
|
|
tsCols = (int64_t*)pColDataInfo->pData;
|
|
|
|
int32_t startPos = 0;
|
|
TSKEY curTs = getStartTsKey(&pBlock->info.window, tsCols);
|
|
SInervalSlicePoint curPoint = {0};
|
|
SInervalSlicePoint prevPoint = {0};
|
|
STimeWindow curWin = getActiveTimeWindow(NULL, pResultRowInfo, curTs, &pInfo->interval, TSDB_ORDER_ASC);
|
|
while (1) {
|
|
int32_t winCode = TSDB_CODE_SUCCESS;
|
|
code = getIntervalSliceCurStateBuf(&pInfo->streamAggSup, &pInfo->interval, pInfo->hasInterpoFunc, &curWin, groupId,
|
|
&curPoint, &prevPoint, &winCode);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
|
|
if (winCode != TSDB_CODE_SUCCESS) {
|
|
SWinKey key = {.ts = curPoint.winKey.win.skey, .groupId = groupId};
|
|
code = tSimpleHashPut(pAggSup->pResultRows, &key, sizeof(SWinKey), &curPoint.pResPos, POINTER_BYTES);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
}
|
|
|
|
code =
|
|
setIntervalSliceOutputBuf(&pInfo->streamAggSup, &curPoint, pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
|
|
forwardRows = getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, curWin.ekey, binarySearchForKey, NULL,
|
|
TSDB_ORDER_ASC);
|
|
int32_t prevEndPos = (forwardRows - 1) + startPos;
|
|
|
|
updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &curPoint.winKey.win, 1);
|
|
code = applyAggFunctionOnPartialTuples(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, startPos,
|
|
forwardRows, pBlock->info.rows, numOfOutput);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
curPoint.pResPos->beUpdated = true;
|
|
|
|
if (curPoint.pLastRow->key == curPoint.winKey.win.ekey) {
|
|
setInterpoWindowFinished(&curPoint);
|
|
}
|
|
|
|
startPos = getNextQualifiedWindow(&pInfo->interval, &curWin, &pBlock->info, tsCols, prevEndPos, TSDB_ORDER_ASC);
|
|
if (startPos < 0) {
|
|
break;
|
|
}
|
|
curTs = tsCols[startPos];
|
|
}
|
|
|
|
if (isHistoryOperator(&pInfo->basic) &&
|
|
tSimpleHashGetSize(pAggSup->pResultRows) > pOperator->resultInfo.capacity * 10) {
|
|
code = copyNewResult(&pAggSup->pResultRows, pInfo->pUpdated, winPosCmprImpl);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
}
|
|
|
|
_end:
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
|
|
}
|
|
return code;
|
|
}
|
|
|
|
void adjustDownstreamBasicInfo(SOperatorInfo* downstream, struct SSteamOpBasicInfo* pBasic) {
|
|
SExecTaskInfo* pTaskInfo = downstream->pTaskInfo;
|
|
if (downstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION) {
|
|
SStreamPartitionOperatorInfo* pPartionInfo = downstream->info;
|
|
pPartionInfo->basic.operatorFlag = pBasic->operatorFlag;
|
|
}
|
|
if (downstream->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
|
|
adjustDownstreamBasicInfo(downstream->pDownstream[0], pBasic);
|
|
}
|
|
SStreamScanInfo* pScanInfo = downstream->info;
|
|
pScanInfo->basic.operatorFlag = pBasic->operatorFlag;
|
|
}
|
|
|
|
int32_t createSemiIntervalSliceOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo,
|
|
SReadHandle* pHandle, SOperatorInfo** ppOptInfo) {
|
|
int32_t code = TSDB_CODE_SUCCESS;
|
|
int32_t lino = 0;
|
|
code = createStreamIntervalSliceOperatorInfo(downstream, pPhyNode, pTaskInfo, pHandle, ppOptInfo);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
|
|
SStreamIntervalSliceOperatorInfo* pInfo = (SStreamIntervalSliceOperatorInfo*)(*ppOptInfo)->info;
|
|
pInfo->nbSup.numOfKeep = 0;
|
|
pInfo->nbSup.pWindowAggFn = doStreamSemiIntervalNonblockAggImpl;
|
|
setSemiOperatorFlag(&pInfo->basic);
|
|
adjustDownstreamBasicInfo(downstream, &pInfo->basic);
|
|
|
|
_end:
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
|
|
}
|
|
return code;
|
|
}
|
|
|
|
bool isDataDeletedStreamWindow(SStreamIntervalSliceOperatorInfo* pInfo, STimeWindow* pWin, uint64_t groupId) {
|
|
SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
|
|
if (pWin->skey < pInfo->nbSup.tsOfKeep) {
|
|
SWinKey key = {.ts = pWin->skey, .groupId = groupId};
|
|
return !(pAggSup->stateStore.streamStateCheck(pAggSup->pState, &key, isFinalOperator(&pInfo->basic), NULL));
|
|
}
|
|
return false;
|
|
}
|
|
|
|
static int32_t doStreamFinalntervalNonblockAggImpl(SOperatorInfo* pOperator, SSDataBlock* pBlock) {
|
|
int32_t code = TSDB_CODE_SUCCESS;
|
|
int32_t lino = 0;
|
|
SStreamIntervalSliceOperatorInfo* pInfo = (SStreamIntervalSliceOperatorInfo*)pOperator->info;
|
|
SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
|
|
SResultRowInfo* pResultRowInfo = &(pInfo->binfo.resultRowInfo);
|
|
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
|
|
SExprSupp* pSup = &pOperator->exprSupp;
|
|
int32_t numOfOutput = pSup->numOfExprs;
|
|
SResultRow* pResult = NULL;
|
|
int32_t forwardRows = 1;
|
|
SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, pInfo->primaryTsIndex);
|
|
TSKEY* tsCols = (int64_t*)pColDataInfo->pData;
|
|
int32_t startPos = 0;
|
|
uint64_t groupId = pBlock->info.id.groupId;
|
|
SInervalSlicePoint curPoint = {0};
|
|
SInervalSlicePoint prevPoint = {0};
|
|
|
|
if (pAggSup->pScanBlock->info.rows > 0) {
|
|
blockDataCleanup(pAggSup->pScanBlock);
|
|
}
|
|
pInfo->twAggSup.minTs = TMIN(pInfo->twAggSup.minTs, pBlock->info.window.ekey);
|
|
|
|
blockDataEnsureCapacity(pAggSup->pScanBlock, pBlock->info.rows);
|
|
TSKEY ts = getStartTsKey(&pBlock->info.window, tsCols);
|
|
STimeWindow curWin = getFinalTimeWindow(ts, &pInfo->interval);
|
|
while (startPos >= 0) {
|
|
if (!isHistoryOperator(&pInfo->basic) && isDataDeletedStreamWindow(pInfo, &curWin, groupId)) {
|
|
uint64_t uid = 0;
|
|
code = appendOneRowToSpecialBlockImpl(pAggSup->pScanBlock, &curWin.skey, &curWin.ekey, &curWin.skey, &curWin.skey,
|
|
&uid, &groupId, NULL, NULL);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
|
|
startPos = getNextQualifiedFinalWindow(&pInfo->interval, &curWin, &pBlock->info, tsCols, startPos);
|
|
} else if (isRecalculateOperator(&pInfo->basic) && !inSlidingWindow(&pInfo->interval, &curWin, &pBlock->info)) {
|
|
startPos = getNextQualifiedFinalWindow(&pInfo->interval, &curWin, &pBlock->info, tsCols, startPos);
|
|
} else {
|
|
break;
|
|
}
|
|
}
|
|
|
|
while (startPos >= 0) {
|
|
int32_t winCode = TSDB_CODE_SUCCESS;
|
|
code = getIntervalSliceCurStateBuf(&pInfo->streamAggSup, &pInfo->interval, pInfo->hasInterpoFunc, &curWin, groupId,
|
|
&curPoint, &prevPoint, &winCode);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
|
|
SWinKey key = {.ts = curPoint.winKey.win.skey, .groupId = groupId};
|
|
if (pInfo->destHasPrimaryKey && winCode == TSDB_CODE_SUCCESS) {
|
|
if (tSimpleHashGet(pAggSup->pResultRows, &key, sizeof(SWinKey)) == NULL) {
|
|
void* pTmp = taosArrayPush(pInfo->pDelWins, &key);
|
|
QUERY_CHECK_NULL(pTmp, code, lino, _end, terrno);
|
|
}
|
|
}
|
|
|
|
curPoint.pResPos->beUpdated = true;
|
|
code = tSimpleHashPut(pAggSup->pResultRows, &key, sizeof(SWinKey), &curPoint.pResPos, POINTER_BYTES);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
|
|
code =
|
|
setIntervalSliceOutputBuf(&pInfo->streamAggSup, &curPoint, pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &curWin, 1);
|
|
code = applyAggFunctionOnPartialTuples(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, startPos,
|
|
forwardRows, pBlock->info.rows, numOfOutput);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
|
|
int32_t prevEndPos = startPos;
|
|
startPos = getNextQualifiedFinalWindow(&pInfo->interval, &curWin, &pBlock->info, tsCols, prevEndPos);
|
|
}
|
|
|
|
if (!pInfo->destHasPrimaryKey && !isHistoryOperator(&pInfo->basic)) {
|
|
code = closeNonblockIntervalWindow(pAggSup->pResultRows, &pInfo->twAggSup, &pInfo->interval, pInfo->pUpdated,
|
|
pTaskInfo);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
} else if ((isHistoryOperator(&pInfo->basic) || isRecalculateOperator(&pInfo->basic)) &&
|
|
tSimpleHashGetSize(pAggSup->pResultRows) > pOperator->resultInfo.capacity * 10) {
|
|
code = copyNewResult(&pAggSup->pResultRows, pInfo->pUpdated, winPosCmprImpl);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
}
|
|
|
|
if (!isHistoryOperator(&pInfo->basic) && !isRecalculateOperator(&pInfo->basic)) {
|
|
code = checkAndSaveWinStateToDisc(0, pInfo->pUpdated, 0, pInfo->basic.pTsDataState, &pInfo->streamAggSup, &pInfo->interval);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
}
|
|
|
|
_end:
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
|
|
}
|
|
return code;
|
|
}
|
|
|
|
int32_t createFinalIntervalSliceOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo,
|
|
SReadHandle* pHandle, SOperatorInfo** ppOptInfo) {
|
|
int32_t code = TSDB_CODE_SUCCESS;
|
|
int32_t lino = 0;
|
|
code = createStreamIntervalSliceOperatorInfo(downstream, pPhyNode, pTaskInfo, pHandle, ppOptInfo);
|
|
QUERY_CHECK_CODE(code, lino, _end);
|
|
|
|
SStreamIntervalSliceOperatorInfo* pInfo = (SStreamIntervalSliceOperatorInfo*)(*ppOptInfo)->info;
|
|
pInfo->nbSup.pWindowAggFn = doStreamFinalntervalNonblockAggImpl;
|
|
pInfo->nbSup.numOfChild = pHandle->numOfVgroups;
|
|
pInfo->streamAggSup.pScanBlock->info.type = STREAM_RETRIEVE;
|
|
pInfo->nbSup.tsOfKeep = INT64_MIN;
|
|
pInfo->twAggSup.waterMark = 0;
|
|
setFinalOperatorFlag(&pInfo->basic);
|
|
adjustDownstreamBasicInfo(downstream, &pInfo->basic);
|
|
|
|
_end:
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
qError("%s failed at line %d since %s. task:%s", __func__, lino, tstrerror(code), GET_TASKID(pTaskInfo));
|
|
}
|
|
return code;
|
|
}
|