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

adj function of stream operator result

This commit is contained in:
54liuyao 2024-07-15 15:32:11 +08:00
parent 4efd774ce8
commit 3cdb957f22
10 changed files with 368 additions and 214 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/libs/executor/storageapi.h
View File

@ -394,7 +394,7 @@ typedef struct SStateStore {
int32_t (*updateInfoInitP)(SInterval* pInterval, int64_t watermark, bool igUp, int8_t pkType, int32_t pkLen, SUpdateInfo** ppInfo); int32_t (*updateInfoInitP)(SInterval* pInterval, int64_t watermark, bool igUp, int8_t pkType, int32_t pkLen, SUpdateInfo** ppInfo);
void (*updateInfoAddCloseWindowSBF)(SUpdateInfo* pInfo); void (*updateInfoAddCloseWindowSBF)(SUpdateInfo* pInfo);
void (*updateInfoDestoryColseWinSBF)(SUpdateInfo* pInfo); void (*updateInfoDestoryColseWinSBF)(SUpdateInfo* pInfo);
int32_t (*updateInfoSerialize)(void* buf, int32_t bufLen, const SUpdateInfo* pInfo); int32_t (*updateInfoSerialize)(void* buf, int32_t bufLen, const SUpdateInfo* pInfo, int32_t* pLen);
int32_t (*updateInfoDeserialize)(void* buf, int32_t bufLen, SUpdateInfo* pInfo); int32_t (*updateInfoDeserialize)(void* buf, int32_t bufLen, SUpdateInfo* pInfo);
SStreamStateCur* (*streamStateSessionSeekKeyNext)(SStreamState* pState, const SSessionKey* key); SStreamStateCur* (*streamStateSessionSeekKeyNext)(SStreamState* pState, const SSessionKey* key);

2
include/libs/stream/tstreamUpdate.h
View File

@ -33,7 +33,7 @@ bool updateInfoIsTableInserted(SUpdateInfo* pInfo, int64_t tbUid);
void updateInfoDestroy(SUpdateInfo* pInfo); void updateInfoDestroy(SUpdateInfo* pInfo);
void updateInfoAddCloseWindowSBF(SUpdateInfo* pInfo); void updateInfoAddCloseWindowSBF(SUpdateInfo* pInfo);
void updateInfoDestoryColseWinSBF(SUpdateInfo* pInfo); void updateInfoDestoryColseWinSBF(SUpdateInfo* pInfo);
int32_t updateInfoSerialize(void* buf, int32_t bufLen, const SUpdateInfo* pInfo); int32_t updateInfoSerialize(void* buf, int32_t bufLen, const SUpdateInfo* pInfo, int32_t* pLen);
int32_t updateInfoDeserialize(void* buf, int32_t bufLen, SUpdateInfo* pInfo); int32_t updateInfoDeserialize(void* buf, int32_t bufLen, SUpdateInfo* pInfo);
void windowSBfDelete(SUpdateInfo* pInfo, uint64_t count); void windowSBfDelete(SUpdateInfo* pInfo, uint64_t count);
int32_t windowSBfAdd(SUpdateInfo* pInfo, uint64_t count); int32_t windowSBfAdd(SUpdateInfo* pInfo, uint64_t count);

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

@ -929,7 +929,7 @@ int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, SExprSupp* pExpSup, i
SStreamState* pState, int32_t keySize, int16_t keyType, SStateStore* pStore, SStreamState* pState, int32_t keySize, int16_t keyType, SStateStore* pStore,
SReadHandle* pHandle, STimeWindowAggSupp* pTwAggSup, const char* taskIdStr, SReadHandle* pHandle, STimeWindowAggSupp* pTwAggSup, const char* taskIdStr,
SStorageAPI* pApi, int32_t tsIndex); SStorageAPI* pApi, int32_t tsIndex);
void initDownStream(struct SOperatorInfo* downstream, SStreamAggSupporter* pAggSup, uint16_t type, int32_t tsColIndex, int32_t initDownStream(struct SOperatorInfo* downstream, SStreamAggSupporter* pAggSup, uint16_t type, int32_t tsColIndex,
STimeWindowAggSupp* pTwSup, struct SSteamOpBasicInfo* pBasic); STimeWindowAggSupp* pTwSup, struct SSteamOpBasicInfo* pBasic);
void getMaxTsWins(const SArray* pAllWins, SArray* pMaxWins); void getMaxTsWins(const SArray* pAllWins, SArray* pMaxWins);
void initGroupResInfoFromArrayList(SGroupResInfo* pGroupResInfo, SArray* pArrayList); void initGroupResInfoFromArrayList(SGroupResInfo* pGroupResInfo, SArray* pArrayList);

1
source/libs/executor/inc/streamexecutorInt.h
View File

@ -20,6 +20,7 @@ extern "C" {
#endif #endif
#include "executorInt.h" #include "executorInt.h"
#include "tutil.h"
void setStreamOperatorState(SSteamOpBasicInfo* pBasicInfo, EStreamType type); void setStreamOperatorState(SSteamOpBasicInfo* pBasicInfo, EStreamType type);
bool needSaveStreamOperatorInfo(SSteamOpBasicInfo* pBasicInfo); bool needSaveStreamOperatorInfo(SSteamOpBasicInfo* pBasicInfo);

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

@ -17,6 +17,7 @@
#include "function.h" #include "function.h"
#include "os.h" #include "os.h"
#include "tname.h" #include "tname.h"
#include "tutil.h"
#include "tdatablock.h" #include "tdatablock.h"
#include "tmsg.h" #include "tmsg.h"
@ -1357,11 +1358,13 @@ static void destroyStreamPartitionOperatorInfo(void* param) {
taosMemoryFreeClear(param); taosMemoryFreeClear(param);
} }
void initParDownStream(SOperatorInfo* downstream, SPartitionBySupporter* pParSup, SExprSupp* pExpr, SExprSupp* pTbnameExpr) { int32_t initParDownStream(SOperatorInfo* downstream, SPartitionBySupporter* pParSup, SExprSupp* pExpr, SExprSupp* pTbnameExpr) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SStorageAPI* pAPI = &downstream->pTaskInfo->storageAPI; SStorageAPI* pAPI = &downstream->pTaskInfo->storageAPI;
if (downstream->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) { if (downstream->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
return; return code;
} }
SStreamScanInfo* pScanInfo = downstream->info; SStreamScanInfo* pScanInfo = downstream->info;
@ -1369,8 +1372,9 @@ void initParDownStream(SOperatorInfo* downstream, SPartitionBySupporter* pParSup
pScanInfo->pPartScalarSup = pExpr; pScanInfo->pPartScalarSup = pExpr;
pScanInfo->pPartTbnameSup = pTbnameExpr; pScanInfo->pPartTbnameSup = pTbnameExpr;
if (!pScanInfo->pUpdateInfo) { if (!pScanInfo->pUpdateInfo) {
pAPI->stateStore.updateInfoInit(60000, TSDB_TIME_PRECISION_MILLI, 0, pScanInfo->igCheckUpdate, pScanInfo->pkColType, pScanInfo->pkColLen, &pScanInfo->pUpdateInfo); code = pAPI->stateStore.updateInfoInit(60000, TSDB_TIME_PRECISION_MILLI, 0, pScanInfo->igCheckUpdate, pScanInfo->pkColType, pScanInfo->pkColLen, &pScanInfo->pUpdateInfo);
} }
return code;
} }
SSDataBlock* buildCreateTableBlock(SExprSupp* tbName, SExprSupp* tag) { SSDataBlock* buildCreateTableBlock(SExprSupp* tbName, SExprSupp* tag) {
@ -1421,6 +1425,7 @@ void freePartItem(void* ptr) {
SOperatorInfo* createStreamPartitionOperatorInfo(SOperatorInfo* downstream, SStreamPartitionPhysiNode* pPartNode, SOperatorInfo* createStreamPartitionOperatorInfo(SOperatorInfo* downstream, SStreamPartitionPhysiNode* pPartNode,
SExecTaskInfo* pTaskInfo) { SExecTaskInfo* pTaskInfo) {
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SStreamPartitionOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamPartitionOperatorInfo)); SStreamPartitionOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamPartitionOperatorInfo));
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
if (pInfo == NULL || pOperator == NULL) { if (pInfo == NULL || pOperator == NULL) {
@ -1511,14 +1516,19 @@ SOperatorInfo* createStreamPartitionOperatorInfo(SOperatorInfo* downstream, SStr
destroyStreamPartitionOperatorInfo, optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL); destroyStreamPartitionOperatorInfo, optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
setOperatorStreamStateFn(pOperator, streamOpReleaseState, streamOpReloadState); setOperatorStreamStateFn(pOperator, streamOpReleaseState, streamOpReloadState);
initParDownStream(downstream, &pInfo->partitionSup, &pInfo->scalarSup, &pInfo->tbnameCalSup); code = initParDownStream(downstream, &pInfo->partitionSup, &pInfo->scalarSup, &pInfo->tbnameCalSup);
TSDB_CHECK_CODE(code, lino, _error);
code = appendDownstream(pOperator, &downstream, 1); code = appendDownstream(pOperator, &downstream, 1);
TSDB_CHECK_CODE(code, lino, _error);
return pOperator; return pOperator;
_error: _error:
pTaskInfo->code = code; pTaskInfo->code = code;
destroyStreamPartitionOperatorInfo(pInfo); destroyStreamPartitionOperatorInfo(pInfo);
taosMemoryFreeClear(pOperator); taosMemoryFreeClear(pOperator);
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
return NULL; return NULL;
} }

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

@ -2477,27 +2477,54 @@ static void doCheckUpdate(SStreamScanInfo* pInfo, TSKEY endKey, SSDataBlock* pBl
} }
} }
int32_t streamScanOperatorEncode(SStreamScanInfo* pInfo, void** pBuff) { int32_t streamScanOperatorEncode(SStreamScanInfo* pInfo, void** pBuff, int32_t* pLen) {
int32_t len = pInfo->stateStore.updateInfoSerialize(NULL, 0, pInfo->pUpdateInfo); int32_t code = TSDB_CODE_SUCCESS;
len += encodeSTimeWindowAggSupp(NULL, &pInfo->twAggSup); int32_t lino = 0;
*pBuff = taosMemoryCalloc(1, len); int32_t len = 0;
void* buf = *pBuff; code = pInfo->stateStore.updateInfoSerialize(NULL, 0, pInfo->pUpdateInfo, &len);
encodeSTimeWindowAggSupp(&buf, &pInfo->twAggSup); TSDB_CHECK_CODE(code, lino, _end);
pInfo->stateStore.updateInfoSerialize(buf, len, pInfo->pUpdateInfo);
return len; len += encodeSTimeWindowAggSupp(NULL, &pInfo->twAggSup);
*pBuff = taosMemoryCalloc(1, len);
if (!(*pBuff)) {
code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _end);
}
void* buf = *pBuff;
encodeSTimeWindowAggSupp(&buf, &pInfo->twAggSup);
int32_t tmp = 0;
code = pInfo->stateStore.updateInfoSerialize(buf, len, pInfo->pUpdateInfo, &tmp);
TSDB_CHECK_CODE(code, lino, _end);
*pLen = len;
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
} }
void streamScanOperatorSaveCheckpoint(SStreamScanInfo* pInfo) { void streamScanOperatorSaveCheckpoint(SStreamScanInfo* pInfo) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
if (!pInfo->pState) { if (!pInfo->pState) {
return; return;
} }
if (needSaveStreamOperatorInfo(&pInfo->basic)) { if (needSaveStreamOperatorInfo(&pInfo->basic)) {
void* pBuf = NULL; void* pBuf = NULL;
int32_t len = streamScanOperatorEncode(pInfo, &pBuf); int32_t len = 0;
code = streamScanOperatorEncode(pInfo, &pBuf, &len);
TSDB_CHECK_CODE(code, lino, _end);
pInfo->stateStore.streamStateSaveInfo(pInfo->pState, STREAM_SCAN_OP_CHECKPOINT_NAME, strlen(STREAM_SCAN_OP_CHECKPOINT_NAME), pBuf, len); pInfo->stateStore.streamStateSaveInfo(pInfo->pState, STREAM_SCAN_OP_CHECKPOINT_NAME, strlen(STREAM_SCAN_OP_CHECKPOINT_NAME), pBuf, len);
taosMemoryFree(pBuf); taosMemoryFree(pBuf);
saveStreamOperatorStateComplete(&pInfo->basic); saveStreamOperatorStateComplete(&pInfo->basic);
} }
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
} }
// other properties are recovered from the execution plan // other properties are recovered from the execution plan
@ -3094,17 +3121,35 @@ static void destroyStreamScanOperatorInfo(void* param) {
} }
void streamScanReleaseState(SOperatorInfo* pOperator) { void streamScanReleaseState(SOperatorInfo* pOperator) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SStreamScanInfo* pInfo = pOperator->info; SStreamScanInfo* pInfo = pOperator->info;
void* pBuff = NULL;
if (!pInfo->pState) { if (!pInfo->pState) {
return; return;
} }
if (!pInfo->pUpdateInfo) { if (!pInfo->pUpdateInfo) {
return; return;
} }
int32_t len = pInfo->stateStore.updateInfoSerialize(NULL, 0, pInfo->pUpdateInfo); int32_t len = 0;
void* pBuff = taosMemoryCalloc(1, len); code = pInfo->stateStore.updateInfoSerialize(NULL, 0, pInfo->pUpdateInfo, &len);
pInfo->stateStore.updateInfoSerialize(pBuff, len, pInfo->pUpdateInfo); TSDB_CHECK_CODE(code, lino, _end);
pBuff = taosMemoryCalloc(1, len);
if (!pBuff) {
code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _end);
}
int32_t tmp = 0;
code = pInfo->stateStore.updateInfoSerialize(pBuff, len, pInfo->pUpdateInfo, &tmp);
TSDB_CHECK_CODE(code, lino, _end);
pInfo->stateStore.streamStateSaveInfo(pInfo->pState, STREAM_SCAN_OP_STATE_NAME, strlen(STREAM_SCAN_OP_STATE_NAME), pBuff, len); pInfo->stateStore.streamStateSaveInfo(pInfo->pState, STREAM_SCAN_OP_STATE_NAME, strlen(STREAM_SCAN_OP_STATE_NAME), pBuff, len);
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
taosMemoryFree(pBuff); taosMemoryFree(pBuff);
} }

90
source/libs/executor/src/streamcountwindowoperator.c
View File

@ -68,9 +68,7 @@ void destroyStreamCountAggOperatorInfo(void* param) {
taosMemoryFreeClear(param); taosMemoryFreeClear(param);
} }
bool isSlidingCountWindow(SStreamAggSupporter* pAggSup) { bool isSlidingCountWindow(SStreamAggSupporter* pAggSup) { return pAggSup->windowCount != pAggSup->windowSliding; }
return pAggSup->windowCount != pAggSup->windowSliding;
}
void setCountOutputBuf(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId, SCountWindowInfo* pCurWin, void setCountOutputBuf(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId, SCountWindowInfo* pCurWin,
SBuffInfo* pBuffInfo) { SBuffInfo* pBuffInfo) {
@ -89,13 +87,14 @@ void setCountOutputBuf(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId,
ASSERT(pBuffInfo->pCur); ASSERT(pBuffInfo->pCur);
pAggSup->stateStore.streamStateCurNext(pAggSup->pState, pBuffInfo->pCur); pAggSup->stateStore.streamStateCurNext(pAggSup->pState, pBuffInfo->pCur);
code = pAggSup->stateStore.streamStateSessionGetKVByCur(pBuffInfo->pCur, &pCurWin->winInfo.sessionWin, code = pAggSup->stateStore.streamStateSessionGetKVByCur(pBuffInfo->pCur, &pCurWin->winInfo.sessionWin,
(void**)&pCurWin->winInfo.pStatePos, &size); (void**)&pCurWin->winInfo.pStatePos, &size);
if (code == TSDB_CODE_FAILED) { if (code == TSDB_CODE_FAILED) {
pAggSup->stateStore.streamStateCountWinAdd(pAggSup->pState, &pCurWin->winInfo.sessionWin, pAggSup->stateStore.streamStateCountWinAdd(pAggSup->pState, &pCurWin->winInfo.sessionWin,
(void**)&pCurWin->winInfo.pStatePos, &size); (void**)&pCurWin->winInfo.pStatePos, &size);
} }
} else { } else {
pBuffInfo->pCur = pAggSup->stateStore.streamStateCountSeekKeyPrev(pAggSup->pState, &pCurWin->winInfo.sessionWin, pAggSup->windowCount); pBuffInfo->pCur = pAggSup->stateStore.streamStateCountSeekKeyPrev(pAggSup->pState, &pCurWin->winInfo.sessionWin,
pAggSup->windowCount);
code = pAggSup->stateStore.streamStateSessionGetKVByCur(pBuffInfo->pCur, &pCurWin->winInfo.sessionWin, code = pAggSup->stateStore.streamStateSessionGetKVByCur(pBuffInfo->pCur, &pCurWin->winInfo.sessionWin,
(void**)&pCurWin->winInfo.pStatePos, &size); (void**)&pCurWin->winInfo.pStatePos, &size);
if (code == TSDB_CODE_FAILED) { if (code == TSDB_CODE_FAILED) {
@ -107,15 +106,16 @@ void setCountOutputBuf(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId,
pBuffInfo->rebuildWindow = true; pBuffInfo->rebuildWindow = true;
} }
} else { } else {
code = pAggSup->stateStore.streamStateCountWinAddIfNotExist( code = pAggSup->stateStore.streamStateCountWinAddIfNotExist(pAggSup->pState, &pCurWin->winInfo.sessionWin,
pAggSup->pState, &pCurWin->winInfo.sessionWin, pAggSup->windowCount, (void**)&pCurWin->winInfo.pStatePos, &size); pAggSup->windowCount,
(void**)&pCurWin->winInfo.pStatePos, &size);
} }
if (code == TSDB_CODE_SUCCESS) { if (code == TSDB_CODE_SUCCESS) {
pCurWin->winInfo.isOutput = true; pCurWin->winInfo.isOutput = true;
} }
pCurWin->pWindowCount= pCurWin->pWindowCount =
(COUNT_TYPE*) ((char*)pCurWin->winInfo.pStatePos->pRowBuff + (pAggSup->resultRowSize - sizeof(COUNT_TYPE))); (COUNT_TYPE*)((char*)pCurWin->winInfo.pStatePos->pRowBuff + (pAggSup->resultRowSize - sizeof(COUNT_TYPE)));
if (*pCurWin->pWindowCount == pAggSup->windowCount) { if (*pCurWin->pWindowCount == pAggSup->windowCount) {
pBuffInfo->rebuildWindow = true; pBuffInfo->rebuildWindow = true;
@ -133,7 +133,7 @@ static int32_t updateCountWindowInfo(SStreamAggSupporter* pAggSup, SCountWindowI
int32_t start, int32_t rows, int32_t maxRows, SSHashObj* pStUpdated, int32_t start, int32_t rows, int32_t maxRows, SSHashObj* pStUpdated,
SSHashObj* pStDeleted, bool* pRebuild) { SSHashObj* pStDeleted, bool* pRebuild) {
SSessionKey sWinKey = pWinInfo->winInfo.sessionWin; SSessionKey sWinKey = pWinInfo->winInfo.sessionWin;
int32_t num = 0; int32_t num = 0;
for (int32_t i = start; i < rows; i++) { for (int32_t i = start; i < rows; i++) {
if (pTs[i] < pWinInfo->winInfo.sessionWin.win.ekey) { if (pTs[i] < pWinInfo->winInfo.sessionWin.win.ekey) {
num++; num++;
@ -181,7 +181,7 @@ void getCountWinRange(SStreamAggSupporter* pAggSup, const SSessionKey* pKey, ESt
pCur = pAggSup->stateStore.streamStateSessionSeekKeyCurrentNext(pAggSup->pState, pKey); pCur = pAggSup->stateStore.streamStateSessionSeekKeyCurrentNext(pAggSup->pState, pKey);
} }
SSessionKey tmpKey = {.groupId = pKey->groupId, .win.ekey = INT64_MIN, .win.skey = INT64_MIN}; SSessionKey tmpKey = {.groupId = pKey->groupId, .win.ekey = INT64_MIN, .win.skey = INT64_MIN};
int32_t code = pAggSup->stateStore.streamStateSessionGetKVByCur(pCur, &tmpKey, NULL, 0); int32_t code = pAggSup->stateStore.streamStateSessionGetKVByCur(pCur, &tmpKey, NULL, 0);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
pAggSup->stateStore.streamStateFreeCur(pCur); pAggSup->stateStore.streamStateFreeCur(pCur);
return; return;
@ -239,7 +239,7 @@ static void doStreamCountAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl
TSKEY* startTsCols = (int64_t*)pStartTsCol->pData; TSKEY* startTsCols = (int64_t*)pStartTsCol->pData;
blockDataEnsureCapacity(pAggSup->pScanBlock, rows * 2); blockDataEnsureCapacity(pAggSup->pScanBlock, rows * 2);
SStreamStateCur* pCur = NULL; SStreamStateCur* pCur = NULL;
COUNT_TYPE slidingRows = 0; COUNT_TYPE slidingRows = 0;
for (int32_t i = 0; i < rows;) { for (int32_t i = 0; i < rows;) {
if (pInfo->ignoreExpiredData && if (pInfo->ignoreExpiredData &&
@ -269,7 +269,7 @@ static void doStreamCountAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl
} }
getCountWinRange(pAggSup, &curWin.winInfo.sessionWin, STREAM_DELETE_DATA, &range); getCountWinRange(pAggSup, &curWin.winInfo.sessionWin, STREAM_DELETE_DATA, &range);
range.win.skey = TMIN(startTsCols[i], range.win.skey); range.win.skey = TMIN(startTsCols[i], range.win.skey);
range.win.ekey = TMAX(startTsCols[rows-1], range.win.ekey); range.win.ekey = TMAX(startTsCols[rows - 1], range.win.ekey);
uint64_t uid = 0; uint64_t uid = 0;
appendDataToSpecialBlock(pAggSup->pScanBlock, &range.win.skey, &range.win.ekey, &uid, &range.groupId, NULL); appendDataToSpecialBlock(pAggSup->pScanBlock, &range.win.skey, &range.win.ekey, &uid, &range.groupId, NULL);
break; break;
@ -289,8 +289,7 @@ static void doStreamCountAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl
if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE && pStUpdated) { if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE && pStUpdated) {
code = saveResult(curWin.winInfo, pStUpdated); code = saveResult(curWin.winInfo, pStUpdated);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
qError("%s do stream count aggregate impl, set result error, code %s", GET_TASKID(pTaskInfo), qError("%s do stream count aggregate impl, set result error, code %s", GET_TASKID(pTaskInfo), tstrerror(code));
tstrerror(code));
break; break;
} }
} }
@ -397,13 +396,14 @@ void* doStreamCountDecodeOpState(void* buf, int32_t len, SOperatorInfo* pOperato
int32_t mapSize = 0; int32_t mapSize = 0;
buf = taosDecodeFixedI32(buf, &mapSize); buf = taosDecodeFixedI32(buf, &mapSize);
for (int32_t i = 0; i < mapSize; i++) { for (int32_t i = 0; i < mapSize; i++) {
SSessionKey key = {0}; SSessionKey key = {0};
SCountWindowInfo curWin = {0}; SCountWindowInfo curWin = {0};
buf = decodeSSessionKey(buf, &key); buf = decodeSSessionKey(buf, &key);
SBuffInfo buffInfo = {.rebuildWindow = false, .winBuffOp = NONE_WINDOW, .pCur = NULL}; SBuffInfo buffInfo = {.rebuildWindow = false, .winBuffOp = NONE_WINDOW, .pCur = NULL};
setCountOutputBuf(&pInfo->streamAggSup, key.win.skey, key.groupId, &curWin, &buffInfo); setCountOutputBuf(&pInfo->streamAggSup, key.win.skey, key.groupId, &curWin, &buffInfo);
buf = decodeSResultWindowInfo(buf, &curWin.winInfo, pInfo->streamAggSup.resultRowSize); buf = decodeSResultWindowInfo(buf, &curWin.winInfo, pInfo->streamAggSup.resultRowSize);
tSimpleHashPut(pInfo->streamAggSup.pResultRows, &key, sizeof(SSessionKey), &curWin.winInfo, sizeof(SResultWindowInfo)); tSimpleHashPut(pInfo->streamAggSup.pResultRows, &key, sizeof(SSessionKey), &curWin.winInfo,
sizeof(SResultWindowInfo));
} }
// 2.twAggSup // 2.twAggSup
@ -441,9 +441,9 @@ void doResetCountWindows(SStreamAggSupporter* pAggSup, SSDataBlock* pBlock) {
uint64_t* gpDatas = (uint64_t*)pGroupCol->pData; uint64_t* gpDatas = (uint64_t*)pGroupCol->pData;
SRowBuffPos* pPos = NULL; SRowBuffPos* pPos = NULL;
int32_t size = 0; int32_t size = 0;
for (int32_t i = 0; i < pBlock->info.rows; i++) { for (int32_t i = 0; i < pBlock->info.rows; i++) {
SSessionKey key = {.groupId = gpDatas[i], .win.skey = startDatas[i], .win.ekey = endDatas[i]}; SSessionKey key = {.groupId = gpDatas[i], .win.skey = startDatas[i], .win.ekey = endDatas[i]};
SStreamStateCur* pCur = NULL; SStreamStateCur* pCur = NULL;
if (isSlidingCountWindow(pAggSup)) { if (isSlidingCountWindow(pAggSup)) {
pCur = pAggSup->stateStore.streamStateCountSeekKeyPrev(pAggSup->pState, &key, pAggSup->windowCount); pCur = pAggSup->stateStore.streamStateCountSeekKeyPrev(pAggSup->pState, &key, pAggSup->windowCount);
@ -452,7 +452,7 @@ void doResetCountWindows(SStreamAggSupporter* pAggSup, SSDataBlock* pBlock) {
} }
while (1) { while (1) {
SSessionKey tmpKey = {.groupId = gpDatas[i], .win.skey = INT64_MIN, .win.ekey = INT64_MIN}; SSessionKey tmpKey = {.groupId = gpDatas[i], .win.skey = INT64_MIN, .win.ekey = INT64_MIN};
int32_t code = pAggSup->stateStore.streamStateSessionGetKVByCur(pCur, &tmpKey, (void **)&pPos, &size); int32_t code = pAggSup->stateStore.streamStateSessionGetKVByCur(pCur, &tmpKey, (void**)&pPos, &size);
if (code != TSDB_CODE_SUCCESS || tmpKey.win.skey > endDatas[i]) { if (code != TSDB_CODE_SUCCESS || tmpKey.win.skey > endDatas[i]) {
pAggSup->stateStore.streamStateFreeCur(pCur); pAggSup->stateStore.streamStateFreeCur(pCur);
break; break;
@ -482,7 +482,7 @@ void doDeleteCountWindows(SStreamAggSupporter* pAggSup, SSDataBlock* pBlock, SAr
SSessionKey key = {.win.skey = startDatas[i], .win.ekey = endDatas[i], .groupId = gpDatas[i]}; SSessionKey key = {.win.skey = startDatas[i], .win.ekey = endDatas[i], .groupId = gpDatas[i]};
while (1) { while (1) {
SSessionKey curWin = {0}; SSessionKey curWin = {0};
int32_t code = pAggSup->stateStore.streamStateCountGetKeyByRange(pAggSup->pState, &key, &curWin); int32_t code = pAggSup->stateStore.streamStateCountGetKeyByRange(pAggSup->pState, &key, &curWin);
if (code == TSDB_CODE_FAILED) { if (code == TSDB_CODE_FAILED) {
break; break;
} }
@ -511,11 +511,11 @@ void deleteCountWinState(SStreamAggSupporter* pAggSup, SSDataBlock* pBlock, SSHa
} }
static SSDataBlock* doStreamCountAgg(SOperatorInfo* pOperator) { static SSDataBlock* doStreamCountAgg(SOperatorInfo* pOperator) {
SExprSupp* pSup = &pOperator->exprSupp; SExprSupp* pSup = &pOperator->exprSupp;
SStreamCountAggOperatorInfo* pInfo = pOperator->info; SStreamCountAggOperatorInfo* pInfo = pOperator->info;
SOptrBasicInfo* pBInfo = &pInfo->binfo; SOptrBasicInfo* pBInfo = &pInfo->binfo;
SStreamAggSupporter* pAggSup = &pInfo->streamAggSup; SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
qDebug("stask:%s %s status: %d", GET_TASKID(pTaskInfo), getStreamOpName(pOperator->operatorType), pOperator->status); qDebug("stask:%s %s status: %d", GET_TASKID(pTaskInfo), getStreamOpName(pOperator->operatorType), pOperator->status);
if (pOperator->status == OP_EXEC_DONE) { if (pOperator->status == OP_EXEC_DONE) {
return NULL; return NULL;
@ -635,7 +635,7 @@ void streamCountReloadState(SOperatorInfo* pOperator) {
int32_t code = pAggSup->stateStore.streamStateGetInfo(pAggSup->pState, STREAM_COUNT_OP_STATE_NAME, int32_t code = pAggSup->stateStore.streamStateGetInfo(pAggSup->pState, STREAM_COUNT_OP_STATE_NAME,
strlen(STREAM_COUNT_OP_STATE_NAME), &pBuf, &size); strlen(STREAM_COUNT_OP_STATE_NAME), &pBuf, &size);
TSKEY ts = *(TSKEY*)pBuf; TSKEY ts = *(TSKEY*)pBuf;
pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, ts); pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, ts);
taosMemoryFree(pBuf); taosMemoryFree(pBuf);
@ -650,11 +650,13 @@ SOperatorInfo* createStreamCountAggOperatorInfo(SOperatorInfo* downstream, SPhys
SExecTaskInfo* pTaskInfo, SReadHandle* pHandle) { SExecTaskInfo* pTaskInfo, SReadHandle* pHandle) {
SCountWinodwPhysiNode* pCountNode = (SCountWinodwPhysiNode*)pPhyNode; SCountWinodwPhysiNode* pCountNode = (SCountWinodwPhysiNode*)pPhyNode;
int32_t numOfCols = 0; int32_t numOfCols = 0;
int32_t code = TSDB_CODE_OUT_OF_MEMORY; int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SStreamCountAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamCountAggOperatorInfo)); SStreamCountAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamCountAggOperatorInfo));
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
if (pInfo == NULL || pOperator == NULL) { if (pInfo == NULL || pOperator == NULL) {
goto _error; code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _error);
} }
pOperator->pTaskInfo = pTaskInfo; pOperator->pTaskInfo = pTaskInfo;
@ -664,18 +666,14 @@ SOperatorInfo* createStreamCountAggOperatorInfo(SOperatorInfo* downstream, SPhys
int32_t numOfScalar = 0; int32_t numOfScalar = 0;
SExprInfo* pScalarExprInfo = createExprInfo(pCountNode->window.pExprs, NULL, &numOfScalar); SExprInfo* pScalarExprInfo = createExprInfo(pCountNode->window.pExprs, NULL, &numOfScalar);
code = initExprSupp(&pInfo->scalarSupp, pScalarExprInfo, numOfScalar, &pTaskInfo->storageAPI.functionStore); code = initExprSupp(&pInfo->scalarSupp, pScalarExprInfo, numOfScalar, &pTaskInfo->storageAPI.functionStore);
if (code != TSDB_CODE_SUCCESS) { TSDB_CHECK_CODE(code, lino, _error);
goto _error;
}
} }
SExprSupp* pExpSup = &pOperator->exprSupp; SExprSupp* pExpSup = &pOperator->exprSupp;
SExprInfo* pExprInfo = createExprInfo(pCountNode->window.pFuncs, NULL, &numOfCols); SExprInfo* pExprInfo = createExprInfo(pCountNode->window.pFuncs, NULL, &numOfCols);
SSDataBlock* pResBlock = createDataBlockFromDescNode(pPhyNode->pOutputDataBlockDesc); SSDataBlock* pResBlock = createDataBlockFromDescNode(pPhyNode->pOutputDataBlockDesc);
code = initBasicInfoEx(&pInfo->binfo, pExpSup, pExprInfo, numOfCols, pResBlock, &pTaskInfo->storageAPI.functionStore); code = initBasicInfoEx(&pInfo->binfo, pExpSup, pExprInfo, numOfCols, pResBlock, &pTaskInfo->storageAPI.functionStore);
if (code != TSDB_CODE_SUCCESS) { TSDB_CHECK_CODE(code, lino, _error);
goto _error;
}
pInfo->twAggSup = (STimeWindowAggSupp){ pInfo->twAggSup = (STimeWindowAggSupp){
.waterMark = pCountNode->window.watermark, .waterMark = pCountNode->window.watermark,
@ -686,12 +684,11 @@ SOperatorInfo* createStreamCountAggOperatorInfo(SOperatorInfo* downstream, SPhys
}; };
pInfo->primaryTsIndex = ((SColumnNode*)pCountNode->window.pTspk)->slotId; pInfo->primaryTsIndex = ((SColumnNode*)pCountNode->window.pTspk)->slotId;
code = initStreamAggSupporter(&pInfo->streamAggSup, pExpSup, numOfCols, 0, code = initStreamAggSupporter(&pInfo->streamAggSup, pExpSup, numOfCols, 0, pTaskInfo->streamInfo.pState,
pTaskInfo->streamInfo.pState, sizeof(COUNT_TYPE), 0, &pTaskInfo->storageAPI.stateStore, pHandle, sizeof(COUNT_TYPE), 0, &pTaskInfo->storageAPI.stateStore, pHandle, &pInfo->twAggSup,
&pInfo->twAggSup, GET_TASKID(pTaskInfo), &pTaskInfo->storageAPI, pInfo->primaryTsIndex); GET_TASKID(pTaskInfo), &pTaskInfo->storageAPI, pInfo->primaryTsIndex);
if (code != TSDB_CODE_SUCCESS) { TSDB_CHECK_CODE(code, lino, _error);
goto _error;
}
pInfo->streamAggSup.windowCount = pCountNode->windowCount; pInfo->streamAggSup.windowCount = pCountNode->windowCount;
pInfo->streamAggSup.windowSliding = pCountNode->windowSliding; pInfo->streamAggSup.windowSliding = pCountNode->windowSliding;
@ -709,7 +706,8 @@ SOperatorInfo* createStreamCountAggOperatorInfo(SOperatorInfo* downstream, SPhys
pInfo->dataVersion = 0; pInfo->dataVersion = 0;
pInfo->historyWins = taosArrayInit(4, sizeof(SSessionKey)); pInfo->historyWins = taosArrayInit(4, sizeof(SSessionKey));
if (!pInfo->historyWins) { if (!pInfo->historyWins) {
goto _error; code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _error);
} }
pInfo->pCheckpointRes = createSpecialDataBlock(STREAM_CHECKPOINT); pInfo->pCheckpointRes = createSpecialDataBlock(STREAM_CHECKPOINT);
@ -735,8 +733,12 @@ SOperatorInfo* createStreamCountAggOperatorInfo(SOperatorInfo* downstream, SPhys
setOperatorStreamStateFn(pOperator, streamCountReleaseState, streamCountReloadState); setOperatorStreamStateFn(pOperator, streamCountReleaseState, streamCountReloadState);
if (downstream) { if (downstream) {
initDownStream(downstream, &pInfo->streamAggSup, pOperator->operatorType, pInfo->primaryTsIndex, &pInfo->twAggSup, &pInfo->basic); code = initDownStream(downstream, &pInfo->streamAggSup, pOperator->operatorType, pInfo->primaryTsIndex,
&pInfo->twAggSup, &pInfo->basic);
TSDB_CHECK_CODE(code, lino, _error);
code = appendDownstream(pOperator, &downstream, 1); code = appendDownstream(pOperator, &downstream, 1);
TSDB_CHECK_CODE(code, lino, _error);
} }
return pOperator; return pOperator;
@ -747,6 +749,6 @@ _error:
taosMemoryFreeClear(pOperator); taosMemoryFreeClear(pOperator);
pTaskInfo->code = code; pTaskInfo->code = code;
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
return NULL; return NULL;
} }

112
source/libs/executor/src/streameventwindowoperator.c
View File

@ -28,9 +28,9 @@
#include "tlog.h" #include "tlog.h"
#include "ttime.h" #include "ttime.h"
#define IS_NORMAL_EVENT_OP(op) ((op)->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_EVENT) #define IS_NORMAL_EVENT_OP(op) ((op)->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_EVENT)
#define STREAM_EVENT_OP_STATE_NAME "StreamEventHistoryState" #define STREAM_EVENT_OP_STATE_NAME "StreamEventHistoryState"
#define STREAM_EVENT_OP_CHECKPOINT_NAME "StreamEventOperator_Checkpoint" #define STREAM_EVENT_OP_CHECKPOINT_NAME "StreamEventOperator_Checkpoint"
typedef struct SEventWinfowFlag { typedef struct SEventWinfowFlag {
bool startFlag; bool startFlag;
@ -86,10 +86,11 @@ void destroyStreamEventOperatorInfo(void* param) {
void setEventWindowFlag(SStreamAggSupporter* pAggSup, SEventWindowInfo* pWinInfo) { void setEventWindowFlag(SStreamAggSupporter* pAggSup, SEventWindowInfo* pWinInfo) {
char* pFlagInfo = (char*)pWinInfo->winInfo.pStatePos->pRowBuff + (pAggSup->resultRowSize - pAggSup->stateKeySize); char* pFlagInfo = (char*)pWinInfo->winInfo.pStatePos->pRowBuff + (pAggSup->resultRowSize - pAggSup->stateKeySize);
pWinInfo->pWinFlag = (SEventWinfowFlag*) pFlagInfo; pWinInfo->pWinFlag = (SEventWinfowFlag*)pFlagInfo;
} }
void setEventWindowInfo(SStreamAggSupporter* pAggSup, SSessionKey* pKey, SRowBuffPos* pPos, SEventWindowInfo* pWinInfo) { void setEventWindowInfo(SStreamAggSupporter* pAggSup, SSessionKey* pKey, SRowBuffPos* pPos,
SEventWindowInfo* pWinInfo) {
pWinInfo->winInfo.sessionWin = *pKey; pWinInfo->winInfo.sessionWin = *pKey;
pWinInfo->winInfo.pStatePos = pPos; pWinInfo->winInfo.pStatePos = pPos;
setEventWindowFlag(pAggSup, pWinInfo); setEventWindowFlag(pAggSup, pWinInfo);
@ -122,22 +123,23 @@ void setEventOutputBuf(SStreamAggSupporter* pAggSup, TSKEY* pTs, uint64_t groupI
pCurWin->winInfo.sessionWin.groupId = groupId; pCurWin->winInfo.sessionWin.groupId = groupId;
pCurWin->winInfo.sessionWin.win.skey = ts; pCurWin->winInfo.sessionWin.win.skey = ts;
pCurWin->winInfo.sessionWin.win.ekey = ts; pCurWin->winInfo.sessionWin.win.ekey = ts;
SStreamStateCur* pCur = pAggSup->stateStore.streamStateSessionSeekKeyCurrentPrev(pAggSup->pState, &pCurWin->winInfo.sessionWin); SStreamStateCur* pCur =
SSessionKey leftWinKey = {.groupId = groupId}; pAggSup->stateStore.streamStateSessionSeekKeyCurrentPrev(pAggSup->pState, &pCurWin->winInfo.sessionWin);
void* pVal = NULL; SSessionKey leftWinKey = {.groupId = groupId};
int32_t len = 0; void* pVal = NULL;
int32_t len = 0;
code = pAggSup->stateStore.streamStateSessionGetKVByCur(pCur, &leftWinKey, &pVal, &len); code = pAggSup->stateStore.streamStateSessionGetKVByCur(pCur, &leftWinKey, &pVal, &len);
if (code == TSDB_CODE_SUCCESS && inWinRange(&pAggSup->winRange, &leftWinKey.win) ) { if (code == TSDB_CODE_SUCCESS && inWinRange(&pAggSup->winRange, &leftWinKey.win)) {
bool inWin = isInTimeWindow(&leftWinKey.win, ts, 0); bool inWin = isInTimeWindow(&leftWinKey.win, ts, 0);
setEventWindowInfo(pAggSup, &leftWinKey, pVal, pCurWin); setEventWindowInfo(pAggSup, &leftWinKey, pVal, pCurWin);
if(inWin || (pCurWin->pWinFlag->startFlag && !pCurWin->pWinFlag->endFlag) ) { if (inWin || (pCurWin->pWinFlag->startFlag && !pCurWin->pWinFlag->endFlag)) {
pCurWin->winInfo.isOutput = !isWindowIncomplete(pCurWin); pCurWin->winInfo.isOutput = !isWindowIncomplete(pCurWin);
goto _end; goto _end;
} }
} }
pAggSup->stateStore.streamStateFreeCur(pCur); pAggSup->stateStore.streamStateFreeCur(pCur);
pCur = pAggSup->stateStore.streamStateSessionSeekKeyNext(pAggSup->pState, &pCurWin->winInfo.sessionWin); pCur = pAggSup->stateStore.streamStateSessionSeekKeyNext(pAggSup->pState, &pCurWin->winInfo.sessionWin);
SSessionKey rightWinKey = {.groupId = groupId}; SSessionKey rightWinKey = {.groupId = groupId};
code = pAggSup->stateStore.streamStateSessionGetKVByCur(pCur, &rightWinKey, &pVal, &len); code = pAggSup->stateStore.streamStateSessionGetKVByCur(pCur, &rightWinKey, &pVal, &len);
bool inWin = isInTimeWindow(&rightWinKey.win, ts, 0); bool inWin = isInTimeWindow(&rightWinKey.win, ts, 0);
if (code == TSDB_CODE_SUCCESS && inWinRange(&pAggSup->winRange, &rightWinKey.win) && (inWin || (start && !end))) { if (code == TSDB_CODE_SUCCESS && inWinRange(&pAggSup->winRange, &rightWinKey.win) && (inWin || (start && !end))) {
@ -149,7 +151,7 @@ void setEventOutputBuf(SStreamAggSupporter* pAggSup, TSKEY* pTs, uint64_t groupI
} }
} }
SSessionKey winKey = {.win.skey = ts, .win.ekey = ts, .groupId = groupId}; SSessionKey winKey = {.win.skey = ts, .win.ekey = ts, .groupId = groupId};
pAggSup->stateStore.streamStateSessionAllocWinBuffByNextPosition(pAggSup->pState, pCur, &winKey, &pVal, &len); pAggSup->stateStore.streamStateSessionAllocWinBuffByNextPosition(pAggSup->pState, pCur, &winKey, &pVal, &len);
setEventWindowInfo(pAggSup, &winKey, pVal, pCurWin); setEventWindowInfo(pAggSup, &winKey, pVal, pCurWin);
pCurWin->pWinFlag->startFlag = start; pCurWin->pWinFlag->startFlag = start;
@ -164,7 +166,7 @@ _end:
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
SET_SESSION_WIN_KEY_INVALID(pNextWinKey); SET_SESSION_WIN_KEY_INVALID(pNextWinKey);
} }
if(pCurWin->winInfo.pStatePos->needFree) { if (pCurWin->winInfo.pStatePos->needFree) {
pAggSup->stateStore.streamStateSessionDel(pAggSup->pState, &pCurWin->winInfo.sessionWin); pAggSup->stateStore.streamStateSessionDel(pAggSup->pState, &pCurWin->winInfo.sessionWin);
} }
pAggSup->stateStore.streamStateFreeCur(pCur); pAggSup->stateStore.streamStateFreeCur(pCur);
@ -173,14 +175,14 @@ _end:
} }
int32_t updateEventWindowInfo(SStreamAggSupporter* pAggSup, SEventWindowInfo* pWinInfo, SSessionKey* pNextWinKey, int32_t updateEventWindowInfo(SStreamAggSupporter* pAggSup, SEventWindowInfo* pWinInfo, SSessionKey* pNextWinKey,
TSKEY* pTsData, bool* starts, bool* ends, int32_t rows, int32_t start, SSHashObj* pResultRows, TSKEY* pTsData, bool* starts, bool* ends, int32_t rows, int32_t start,
SSHashObj* pStUpdated, SSHashObj* pStDeleted, bool* pRebuild) { SSHashObj* pResultRows, SSHashObj* pStUpdated, SSHashObj* pStDeleted, bool* pRebuild) {
*pRebuild = false; *pRebuild = false;
if (!pWinInfo->pWinFlag->startFlag && !(starts[start]) ) { if (!pWinInfo->pWinFlag->startFlag && !(starts[start])) {
return 1; return 1;
} }
TSKEY maxTs = INT64_MAX; TSKEY maxTs = INT64_MAX;
STimeWindow* pWin = &pWinInfo->winInfo.sessionWin.win; STimeWindow* pWin = &pWinInfo->winInfo.sessionWin.win;
if (pWinInfo->pWinFlag->endFlag) { if (pWinInfo->pWinFlag->endFlag) {
maxTs = pWin->ekey + 1; maxTs = pWin->ekey + 1;
@ -219,7 +221,7 @@ int32_t updateEventWindowInfo(SStreamAggSupporter* pAggSup, SEventWindowInfo* pW
memcpy(pWinInfo->winInfo.pStatePos->pKey, &pWinInfo->winInfo.sessionWin, sizeof(SSessionKey)); memcpy(pWinInfo->winInfo.pStatePos->pKey, &pWinInfo->winInfo.sessionWin, sizeof(SSessionKey));
if (ends[i]) { if (ends[i]) {
if (pWinInfo->pWinFlag->endFlag && pWin->skey <= pTsData[i] && pTsData[i] < pWin->ekey ) { if (pWinInfo->pWinFlag->endFlag && pWin->skey <= pTsData[i] && pTsData[i] < pWin->ekey) {
*pRebuild = true; *pRebuild = true;
} }
return i + 1 - start; return i + 1 - start;
@ -230,26 +232,28 @@ int32_t updateEventWindowInfo(SStreamAggSupporter* pAggSup, SEventWindowInfo* pW
static int32_t compactEventWindow(SOperatorInfo* pOperator, SEventWindowInfo* pCurWin, SSHashObj* pStUpdated, static int32_t compactEventWindow(SOperatorInfo* pOperator, SEventWindowInfo* pCurWin, SSHashObj* pStUpdated,
SSHashObj* pStDeleted, bool addGap) { SSHashObj* pStDeleted, bool addGap) {
SExprSupp* pSup = &pOperator->exprSupp; SExprSupp* pSup = &pOperator->exprSupp;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SStorageAPI* pAPI = &pOperator->pTaskInfo->storageAPI; SStorageAPI* pAPI = &pOperator->pTaskInfo->storageAPI;
int32_t winNum = 0; int32_t winNum = 0;
SStreamEventAggOperatorInfo* pInfo = pOperator->info; SStreamEventAggOperatorInfo* pInfo = pOperator->info;
SResultRow* pCurResult = NULL; SResultRow* pCurResult = NULL;
int32_t numOfOutput = pOperator->exprSupp.numOfExprs; int32_t numOfOutput = pOperator->exprSupp.numOfExprs;
SStreamAggSupporter* pAggSup = &pInfo->streamAggSup; SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
while (1) { while (1) {
if (!pCurWin->pWinFlag->startFlag || pCurWin->pWinFlag->endFlag) { if (!pCurWin->pWinFlag->startFlag || pCurWin->pWinFlag->endFlag) {
break; break;
} }
SEventWindowInfo nextWinInfo = {0}; SEventWindowInfo nextWinInfo = {0};
getNextSessionWinInfo(pAggSup, pStUpdated, &pCurWin->winInfo, &nextWinInfo.winInfo); getNextSessionWinInfo(pAggSup, pStUpdated, &pCurWin->winInfo, &nextWinInfo.winInfo);
if (!IS_VALID_SESSION_WIN(nextWinInfo.winInfo) || !inWinRange(&pAggSup->winRange, &nextWinInfo.winInfo.sessionWin.win)) { if (!IS_VALID_SESSION_WIN(nextWinInfo.winInfo) ||
!inWinRange(&pAggSup->winRange, &nextWinInfo.winInfo.sessionWin.win)) {
releaseOutputBuf(pAggSup->pState, nextWinInfo.winInfo.pStatePos, &pAggSup->pSessionAPI->stateStore); releaseOutputBuf(pAggSup->pState, nextWinInfo.winInfo.pStatePos, &pAggSup->pSessionAPI->stateStore);
break; break;
} }
setEventWindowFlag(pAggSup, &nextWinInfo); setEventWindowFlag(pAggSup, &nextWinInfo);
compactTimeWindow(pSup, pAggSup, &pInfo->twAggSup, pTaskInfo, &pCurWin->winInfo, &nextWinInfo.winInfo, pStUpdated, pStDeleted, false); compactTimeWindow(pSup, pAggSup, &pInfo->twAggSup, pTaskInfo, &pCurWin->winInfo, &nextWinInfo.winInfo, pStUpdated,
pStDeleted, false);
pCurWin->pWinFlag->endFlag = nextWinInfo.pWinFlag->endFlag; pCurWin->pWinFlag->endFlag = nextWinInfo.pWinFlag->endFlag;
winNum++; winNum++;
} }
@ -275,7 +279,7 @@ static void doStreamEventAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl
int32_t winRows = 0; int32_t winRows = 0;
SStreamAggSupporter* pAggSup = &pInfo->streamAggSup; SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
SColumnInfoData* pColStart = NULL; SColumnInfoData* pColStart = NULL;
SColumnInfoData* pColEnd = NULL; SColumnInfoData* pColEnd = NULL;
pInfo->dataVersion = TMAX(pInfo->dataVersion, pSDataBlock->info.version); pInfo->dataVersion = TMAX(pInfo->dataVersion, pSDataBlock->info.version);
pAggSup->winRange = pTaskInfo->streamInfo.fillHistoryWindow; pAggSup->winRange = pTaskInfo->streamInfo.fillHistoryWindow;
@ -290,7 +294,8 @@ static void doStreamEventAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl
return; return;
} }
SFilterColumnParam paramStart = {.numOfCols = taosArrayGetSize(pSDataBlock->pDataBlock), .pDataBlock = pSDataBlock->pDataBlock}; SFilterColumnParam paramStart = {.numOfCols = taosArrayGetSize(pSDataBlock->pDataBlock),
.pDataBlock = pSDataBlock->pDataBlock};
code = filterSetDataFromSlotId(pInfo->pStartCondInfo, &paramStart); code = filterSetDataFromSlotId(pInfo->pStartCondInfo, &paramStart);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
qError("set data from start slotId error."); qError("set data from start slotId error.");
@ -299,7 +304,8 @@ static void doStreamEventAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl
int32_t statusStart = 0; int32_t statusStart = 0;
filterExecute(pInfo->pStartCondInfo, pSDataBlock, &pColStart, NULL, paramStart.numOfCols, &statusStart); filterExecute(pInfo->pStartCondInfo, pSDataBlock, &pColStart, NULL, paramStart.numOfCols, &statusStart);
SFilterColumnParam paramEnd = {.numOfCols = taosArrayGetSize(pSDataBlock->pDataBlock), .pDataBlock = pSDataBlock->pDataBlock}; SFilterColumnParam paramEnd = {.numOfCols = taosArrayGetSize(pSDataBlock->pDataBlock),
.pDataBlock = pSDataBlock->pDataBlock};
code = filterSetDataFromSlotId(pInfo->pEndCondInfo, &paramEnd); code = filterSetDataFromSlotId(pInfo->pEndCondInfo, &paramEnd);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
qError("set data from end slotId error."); qError("set data from end slotId error.");
@ -320,20 +326,23 @@ static void doStreamEventAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl
int32_t winIndex = 0; int32_t winIndex = 0;
bool allEqual = true; bool allEqual = true;
SEventWindowInfo curWin = {0}; SEventWindowInfo curWin = {0};
SSessionKey nextWinKey = {0}; SSessionKey nextWinKey = {0};
setEventOutputBuf(pAggSup, tsCols, groupId, (bool*)pColStart->pData, (bool*)pColEnd->pData, i, rows, &curWin, &nextWinKey); setEventOutputBuf(pAggSup, tsCols, groupId, (bool*)pColStart->pData, (bool*)pColEnd->pData, i, rows, &curWin,
&nextWinKey);
setSessionWinOutputInfo(pSeUpdated, &curWin.winInfo); setSessionWinOutputInfo(pSeUpdated, &curWin.winInfo);
bool rebuild = false; bool rebuild = false;
winRows = updateEventWindowInfo(pAggSup, &curWin, &nextWinKey, tsCols, (bool*)pColStart->pData, (bool*)pColEnd->pData, rows, i, winRows =
pAggSup->pResultRows, pSeUpdated, pStDeleted, &rebuild); updateEventWindowInfo(pAggSup, &curWin, &nextWinKey, tsCols, (bool*)pColStart->pData, (bool*)pColEnd->pData,
rows, i, pAggSup->pResultRows, pSeUpdated, pStDeleted, &rebuild);
ASSERT(winRows >= 1); ASSERT(winRows >= 1);
if (rebuild) { if (rebuild) {
uint64_t uid = 0; uint64_t uid = 0;
appendDataToSpecialBlock(pAggSup->pScanBlock, &curWin.winInfo.sessionWin.win.skey, appendDataToSpecialBlock(pAggSup->pScanBlock, &curWin.winInfo.sessionWin.win.skey,
&curWin.winInfo.sessionWin.win.ekey, &uid, &groupId, NULL); &curWin.winInfo.sessionWin.win.ekey, &uid, &groupId, NULL);
tSimpleHashRemove(pSeUpdated, &curWin.winInfo.sessionWin, sizeof(SSessionKey)); tSimpleHashRemove(pSeUpdated, &curWin.winInfo.sessionWin, sizeof(SSessionKey));
doDeleteEventWindow(pAggSup, pSeUpdated, &curWin.winInfo.sessionWin); doDeleteEventWindow(pAggSup, pSeUpdated, &curWin.winInfo.sessionWin);
if (pInfo->destHasPrimaryKey && curWin.winInfo.isOutput && IS_NORMAL_EVENT_OP(pOperator) && !isWindowIncomplete(&curWin)) { if (pInfo->destHasPrimaryKey && curWin.winInfo.isOutput && IS_NORMAL_EVENT_OP(pOperator) &&
!isWindowIncomplete(&curWin)) {
saveDeleteRes(pInfo->pPkDeleted, curWin.winInfo.sessionWin); saveDeleteRes(pInfo->pPkDeleted, curWin.winInfo.sessionWin);
} }
releaseOutputBuf(pAggSup->pState, curWin.winInfo.pStatePos, &pAPI->stateStore); releaseOutputBuf(pAggSup->pState, curWin.winInfo.pStatePos, &pAPI->stateStore);
@ -603,7 +612,8 @@ void streamEventReleaseState(SOperatorInfo* pOperator) {
char* pBuff = taosMemoryCalloc(1, resSize); char* pBuff = taosMemoryCalloc(1, resSize);
memcpy(pBuff, pInfo->historyWins->pData, winSize); memcpy(pBuff, pInfo->historyWins->pData, winSize);
memcpy(pBuff + winSize, &pInfo->twAggSup.maxTs, sizeof(TSKEY)); memcpy(pBuff + winSize, &pInfo->twAggSup.maxTs, sizeof(TSKEY));
qDebug("===stream=== event window operator relase state. save result count:%d", (int32_t)taosArrayGetSize(pInfo->historyWins)); qDebug("===stream=== event window operator relase state. save result count:%d",
(int32_t)taosArrayGetSize(pInfo->historyWins));
pInfo->streamAggSup.stateStore.streamStateSaveInfo(pInfo->streamAggSup.pState, STREAM_EVENT_OP_STATE_NAME, pInfo->streamAggSup.stateStore.streamStateSaveInfo(pInfo->streamAggSup.pState, STREAM_EVENT_OP_STATE_NAME,
strlen(STREAM_EVENT_OP_STATE_NAME), pBuff, resSize); strlen(STREAM_EVENT_OP_STATE_NAME), pBuff, resSize);
pInfo->streamAggSup.stateStore.streamStateCommit(pInfo->streamAggSup.pState); pInfo->streamAggSup.stateStore.streamStateCommit(pInfo->streamAggSup.pState);
@ -647,7 +657,7 @@ void streamEventReloadState(SOperatorInfo* pOperator) {
qDebug("===stream=== reload state. try process result %" PRId64 ", %" PRIu64 ", index:%d", pSeKeyBuf[i].win.skey, qDebug("===stream=== reload state. try process result %" PRId64 ", %" PRIu64 ", index:%d", pSeKeyBuf[i].win.skey,
pSeKeyBuf[i].groupId, i); pSeKeyBuf[i].groupId, i);
getSessionWindowInfoByKey(pAggSup, pSeKeyBuf + i, &curInfo.winInfo); getSessionWindowInfoByKey(pAggSup, pSeKeyBuf + i, &curInfo.winInfo);
//event window has been deleted // event window has been deleted
if (!IS_VALID_SESSION_WIN(curInfo.winInfo)) { if (!IS_VALID_SESSION_WIN(curInfo.winInfo)) {
continue; continue;
} }
@ -659,7 +669,7 @@ void streamEventReloadState(SOperatorInfo* pOperator) {
compactEventWindow(pOperator, &curInfo, pInfo->pSeUpdated, pInfo->pSeDeleted, false); compactEventWindow(pOperator, &curInfo, pInfo->pSeUpdated, pInfo->pSeDeleted, false);
qDebug("===stream=== reload state. save result %" PRId64 ", %" PRIu64, curInfo.winInfo.sessionWin.win.skey, qDebug("===stream=== reload state. save result %" PRId64 ", %" PRIu64, curInfo.winInfo.sessionWin.win.skey,
curInfo.winInfo.sessionWin.groupId); curInfo.winInfo.sessionWin.groupId);
if (IS_VALID_SESSION_WIN(curInfo.winInfo)) { if (IS_VALID_SESSION_WIN(curInfo.winInfo)) {
saveSessionOutputBuf(pAggSup, &curInfo.winInfo); saveSessionOutputBuf(pAggSup, &curInfo.winInfo);
} }
@ -693,6 +703,7 @@ SOperatorInfo* createStreamEventAggOperatorInfo(SOperatorInfo* downstream, SPhys
SStreamEventWinodwPhysiNode* pEventNode = (SStreamEventWinodwPhysiNode*)pPhyNode; SStreamEventWinodwPhysiNode* pEventNode = (SStreamEventWinodwPhysiNode*)pPhyNode;
int32_t tsSlotId = ((SColumnNode*)pEventNode->window.pTspk)->slotId; int32_t tsSlotId = ((SColumnNode*)pEventNode->window.pTspk)->slotId;
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SStreamEventAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamEventAggOperatorInfo)); SStreamEventAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamEventAggOperatorInfo));
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
if (pInfo == NULL || pOperator == NULL) { if (pInfo == NULL || pOperator == NULL) {
@ -783,20 +794,18 @@ SOperatorInfo* createStreamEventAggOperatorInfo(SOperatorInfo* downstream, SPhys
pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamEventAgg, NULL, destroyStreamEventOperatorInfo, pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamEventAgg, NULL, destroyStreamEventOperatorInfo,
optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL); optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
setOperatorStreamStateFn(pOperator, streamEventReleaseState, streamEventReloadState); setOperatorStreamStateFn(pOperator, streamEventReleaseState, streamEventReloadState);
initDownStream(downstream, &pInfo->streamAggSup, pOperator->operatorType, pInfo->primaryTsIndex, &pInfo->twAggSup, &pInfo->basic); code = initDownStream(downstream, &pInfo->streamAggSup, pOperator->operatorType, pInfo->primaryTsIndex,
&pInfo->twAggSup, &pInfo->basic);
TSDB_CHECK_CODE(code, lino, _error);
code = appendDownstream(pOperator, &downstream, 1); code = appendDownstream(pOperator, &downstream, 1);
if (code != TSDB_CODE_SUCCESS) { TSDB_CHECK_CODE(code, lino, _error);
goto _error;
}
code = filterInitFromNode((SNode*)pEventNode->pStartCond, &pInfo->pStartCondInfo, 0); code = filterInitFromNode((SNode*)pEventNode->pStartCond, &pInfo->pStartCondInfo, 0);
if (code != TSDB_CODE_SUCCESS) { TSDB_CHECK_CODE(code, lino, _error);
goto _error;
}
code = filterInitFromNode((SNode*)pEventNode->pEndCond, &pInfo->pEndCondInfo, 0); code = filterInitFromNode((SNode*)pEventNode->pEndCond, &pInfo->pEndCondInfo, 0);
if (code != TSDB_CODE_SUCCESS) { TSDB_CHECK_CODE(code, lino, _error);
goto _error;
}
return pOperator; return pOperator;
@ -804,5 +813,6 @@ _error:
destroyStreamEventOperatorInfo(pInfo); destroyStreamEventOperatorInfo(pInfo);
taosMemoryFreeClear(pOperator); taosMemoryFreeClear(pOperator);
pTaskInfo->code = code; pTaskInfo->code = code;
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
return NULL; return NULL;
} }

185
source/libs/executor/src/streamtimewindowoperator.c
View File

@ -28,14 +28,18 @@
#include "tlog.h" #include "tlog.h"
#include "ttime.h" #include "ttime.h"
#define IS_FINAL_INTERVAL_OP(op) ((op)->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL) #define IS_FINAL_INTERVAL_OP(op) ((op)->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL)
#define IS_MID_INTERVAL_OP(op) ((op)->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_MID_INTERVAL) #define IS_MID_INTERVAL_OP(op) ((op)->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_MID_INTERVAL)
#define IS_NORMAL_INTERVAL_OP(op) ((op)->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL || (op)->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL) #define IS_NORMAL_INTERVAL_OP(op) \
((op)->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL || \
(op)->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL)
#define IS_FINAL_SESSION_OP(op) ((op)->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION) #define IS_FINAL_SESSION_OP(op) ((op)->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION)
#define IS_NORMAL_SESSION_OP(op) ((op)->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION || (op)->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION) #define IS_NORMAL_SESSION_OP(op) \
((op)->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION || \
(op)->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION)
#define IS_NORMAL_STATE_OP(op) ((op)->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE) #define IS_NORMAL_STATE_OP(op) ((op)->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE)
#define DEAULT_DELETE_MARK INT64_MAX #define DEAULT_DELETE_MARK INT64_MAX
#define STREAM_INTERVAL_OP_STATE_NAME "StreamIntervalHistoryState" #define STREAM_INTERVAL_OP_STATE_NAME "StreamIntervalHistoryState"
@ -263,14 +267,16 @@ static void doDeleteWindows(SOperatorInfo* pOperator, SInterval* pInterval, SSDa
if (chIds) { if (chIds) {
int32_t childId = getChildIndex(pBlock); int32_t childId = getChildIndex(pBlock);
if (pInvalidWins) { if (pInvalidWins) {
qDebug("===stream===save invalid delete window:%" PRId64 ",groupId:%" PRId64 ",chId:%d", winRes.ts, winRes.groupId, childId); qDebug("===stream===save invalid delete window:%" PRId64 ",groupId:%" PRId64 ",chId:%d", winRes.ts,
winRes.groupId, childId);
taosHashPut(pInvalidWins, &winRes, sizeof(SWinKey), NULL, 0); taosHashPut(pInvalidWins, &winRes, sizeof(SWinKey), NULL, 0);
} }
SArray* chArray = *(void**)chIds; SArray* chArray = *(void**)chIds;
int32_t index = taosArraySearchIdx(chArray, &childId, compareInt32Val, TD_EQ); int32_t index = taosArraySearchIdx(chArray, &childId, compareInt32Val, TD_EQ);
if (index != -1) { if (index != -1) {
qDebug("===stream===try push delete window:%" PRId64 ",groupId:%" PRId64 ",chId:%d ,continue", win.skey, winGpId, childId); qDebug("===stream===try push delete window:%" PRId64 ",groupId:%" PRId64 ",chId:%d ,continue", win.skey,
winGpId, childId);
getNextTimeWindow(pInterval, &win, TSDB_ORDER_ASC); getNextTimeWindow(pInterval, &win, TSDB_ORDER_ASC);
continue; continue;
} }
@ -390,7 +396,7 @@ static void doBuildDeleteResult(SStreamIntervalOperatorInfo* pInfo, SArray* pWin
} }
void destroyFlusedPos(void* pRes) { void destroyFlusedPos(void* pRes) {
SRowBuffPos* pPos = (SRowBuffPos*) pRes; SRowBuffPos* pPos = (SRowBuffPos*)pRes;
if (!pPos->needFree && !pPos->pRowBuff) { if (!pPos->needFree && !pPos->pRowBuff) {
taosMemoryFreeClear(pPos->pKey); taosMemoryFreeClear(pPos->pKey);
taosMemoryFree(pPos); taosMemoryFree(pPos);
@ -626,17 +632,17 @@ static void doBuildPullDataBlock(SArray* array, int32_t* pIndex, SSDataBlock* pB
blockDataUpdateTsWindow(pBlock, 0); blockDataUpdateTsWindow(pBlock, 0);
} }
static bool processPullOver(SSDataBlock* pBlock, SHashObj* pMap, SHashObj* pFinalMap, SInterval* pInterval, SArray* pPullWins, static bool processPullOver(SSDataBlock* pBlock, SHashObj* pMap, SHashObj* pFinalMap, SInterval* pInterval,
int32_t numOfCh, SOperatorInfo* pOperator) { SArray* pPullWins, int32_t numOfCh, SOperatorInfo* pOperator) {
SStreamIntervalOperatorInfo* pInfo = pOperator->info; SStreamIntervalOperatorInfo* pInfo = pOperator->info;
SColumnInfoData* pStartCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX); SColumnInfoData* pStartCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);
TSKEY* tsData = (TSKEY*)pStartCol->pData; TSKEY* tsData = (TSKEY*)pStartCol->pData;
SColumnInfoData* pEndCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX); SColumnInfoData* pEndCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);
TSKEY* tsEndData = (TSKEY*)pEndCol->pData; TSKEY* tsEndData = (TSKEY*)pEndCol->pData;
SColumnInfoData* pGroupCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX); SColumnInfoData* pGroupCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
uint64_t* groupIdData = (uint64_t*)pGroupCol->pData; uint64_t* groupIdData = (uint64_t*)pGroupCol->pData;
int32_t chId = getChildIndex(pBlock); int32_t chId = getChildIndex(pBlock);
bool res = false; bool res = false;
for (int32_t i = 0; i < pBlock->info.rows; i++) { for (int32_t i = 0; i < pBlock->info.rows; i++) {
TSKEY winTs = tsData[i]; TSKEY winTs = tsData[i];
while (winTs <= tsEndData[i]) { while (winTs <= tsEndData[i]) {
@ -652,7 +658,7 @@ static bool processPullOver(SSDataBlock* pBlock, SHashObj* pMap, SHashObj* pFina
// pull data is over // pull data is over
taosArrayDestroy(chArray); taosArrayDestroy(chArray);
taosHashRemove(pMap, &winRes, sizeof(SWinKey)); taosHashRemove(pMap, &winRes, sizeof(SWinKey));
res =true; res = true;
qDebug("===stream===retrive pull data over.window %" PRId64, winRes.ts); qDebug("===stream===retrive pull data over.window %" PRId64, winRes.ts);
void* pFinalCh = taosHashGet(pFinalMap, &winRes, sizeof(SWinKey)); void* pFinalCh = taosHashGet(pFinalMap, &winRes, sizeof(SWinKey));
@ -665,7 +671,8 @@ static bool processPullOver(SSDataBlock* pBlock, SHashObj* pMap, SHashObj* pFina
.calWin.skey = nextWin.skey, .calWin.skey = nextWin.skey,
.calWin.ekey = nextWin.skey}; .calWin.ekey = nextWin.skey};
// add pull data request // add pull data request
qDebug("===stream===prepare final retrive for delete window:%" PRId64 ",groupId:%" PRId64 ", size:%d", winRes.ts, winRes.groupId, numOfCh); qDebug("===stream===prepare final retrive for delete window:%" PRId64 ",groupId:%" PRId64 ", size:%d",
winRes.ts, winRes.groupId, numOfCh);
if (IS_MID_INTERVAL_OP(pOperator)) { if (IS_MID_INTERVAL_OP(pOperator)) {
SStreamIntervalOperatorInfo* pInfo = (SStreamIntervalOperatorInfo*)pOperator->info; SStreamIntervalOperatorInfo* pInfo = (SStreamIntervalOperatorInfo*)pOperator->info;
taosArrayPush(pInfo->pMidPullDatas, &winRes); taosArrayPush(pInfo->pMidPullDatas, &winRes);
@ -673,7 +680,7 @@ static bool processPullOver(SSDataBlock* pBlock, SHashObj* pMap, SHashObj* pFina
taosArrayPush(pInfo->pDelWins, &winRes); taosArrayPush(pInfo->pDelWins, &winRes);
addPullWindow(pMap, &winRes, numOfCh); addPullWindow(pMap, &winRes, numOfCh);
if (pInfo->destHasPrimaryKey) { if (pInfo->destHasPrimaryKey) {
tSimpleHashPut(pInfo->pDeletedMap,&winRes, sizeof(SWinKey), NULL, 0); tSimpleHashPut(pInfo->pDeletedMap, &winRes, sizeof(SWinKey), NULL, 0);
} }
qDebug("===stream===prepare final retrive for delete %" PRId64 ", size:%d", winRes.ts, numOfCh); qDebug("===stream===prepare final retrive for delete %" PRId64 ", size:%d", winRes.ts, numOfCh);
} }
@ -692,7 +699,7 @@ static void addRetriveWindow(SArray* wins, SStreamIntervalOperatorInfo* pInfo, i
for (int32_t i = 0; i < size; i++) { for (int32_t i = 0; i < size; i++) {
SWinKey* winKey = taosArrayGet(wins, i); SWinKey* winKey = taosArrayGet(wins, i);
STimeWindow nextWin = getFinalTimeWindow(winKey->ts, &pInfo->interval); STimeWindow nextWin = getFinalTimeWindow(winKey->ts, &pInfo->interval);
void* chIds = taosHashGet(pInfo->pPullDataMap, winKey, sizeof(SWinKey)); void* chIds = taosHashGet(pInfo->pPullDataMap, winKey, sizeof(SWinKey));
if (!chIds) { if (!chIds) {
SPullWindowInfo pull = { SPullWindowInfo pull = {
.window = nextWin, .groupId = winKey->groupId, .calWin.skey = nextWin.skey, .calWin.ekey = nextWin.skey}; .window = nextWin, .groupId = winKey->groupId, .calWin.skey = nextWin.skey, .calWin.ekey = nextWin.skey};
@ -700,7 +707,7 @@ static void addRetriveWindow(SArray* wins, SStreamIntervalOperatorInfo* pInfo, i
if (savePullWindow(&pull, pInfo->pPullWins) == TSDB_CODE_SUCCESS) { if (savePullWindow(&pull, pInfo->pPullWins) == TSDB_CODE_SUCCESS) {
addPullWindow(pInfo->pPullDataMap, winKey, pInfo->numOfChild); addPullWindow(pInfo->pPullDataMap, winKey, pInfo->numOfChild);
if (pInfo->destHasPrimaryKey) { if (pInfo->destHasPrimaryKey) {
tSimpleHashPut(pInfo->pDeletedMap,winKey, sizeof(SWinKey), NULL, 0); tSimpleHashPut(pInfo->pDeletedMap, winKey, sizeof(SWinKey), NULL, 0);
} }
qDebug("===stream===prepare retrive for delete %" PRId64 ", size:%d", winKey->ts, pInfo->numOfChild); qDebug("===stream===prepare retrive for delete %" PRId64 ", size:%d", winKey->ts, pInfo->numOfChild);
} }
@ -753,7 +760,8 @@ int32_t buildDataBlockFromGroupRes(SOperatorInfo* pOperator, void* pState, SSDat
if (pBlock->info.id.groupId == 0) { if (pBlock->info.id.groupId == 0) {
pBlock->info.id.groupId = groupId; pBlock->info.id.groupId = groupId;
void* tbname = NULL; void* tbname = NULL;
if (pAPI->stateStore.streamStateGetParName(pTaskInfo->streamInfo.pState, pBlock->info.id.groupId, &tbname, false) < 0) { if (pAPI->stateStore.streamStateGetParName(pTaskInfo->streamInfo.pState, pBlock->info.id.groupId, &tbname,
false) < 0) {
pBlock->info.parTbName[0] = 0; pBlock->info.parTbName[0] = 0;
} else { } else {
memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN); memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN);
@ -831,9 +839,7 @@ static int32_t getNextQualifiedFinalWindow(SInterval* pInterval, STimeWindow* pN
return startPos; return startPos;
} }
bool hasSrcPrimaryKeyCol(SSteamOpBasicInfo* pInfo) { bool hasSrcPrimaryKeyCol(SSteamOpBasicInfo* pInfo) { return pInfo->primaryPkIndex != -1; }
return pInfo->primaryPkIndex != -1;
}
static void doStreamIntervalAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBlock, uint64_t groupId, static void doStreamIntervalAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBlock, uint64_t groupId,
SSHashObj* pUpdatedMap, SSHashObj* pDeletedMap) { SSHashObj* pUpdatedMap, SSHashObj* pDeletedMap) {
@ -854,23 +860,22 @@ static void doStreamIntervalAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDat
SColumnInfoData* pColDataInfo = taosArrayGet(pSDataBlock->pDataBlock, pInfo->primaryTsIndex); SColumnInfoData* pColDataInfo = taosArrayGet(pSDataBlock->pDataBlock, pInfo->primaryTsIndex);
tsCols = (int64_t*)pColDataInfo->pData; tsCols = (int64_t*)pColDataInfo->pData;
void* pPkVal = NULL; void* pPkVal = NULL;
int32_t pkLen = 0; int32_t pkLen = 0;
SColumnInfoData* pPkColDataInfo = NULL; SColumnInfoData* pPkColDataInfo = NULL;
if (hasSrcPrimaryKeyCol(&pInfo->basic)) { if (hasSrcPrimaryKeyCol(&pInfo->basic)) {
pPkColDataInfo = taosArrayGet(pSDataBlock->pDataBlock, pInfo->basic.primaryPkIndex); pPkColDataInfo = taosArrayGet(pSDataBlock->pDataBlock, pInfo->basic.primaryPkIndex);
} }
if (pSDataBlock->info.window.skey != tsCols[0] || pSDataBlock->info.window.ekey != tsCols[endRowId]) { if (pSDataBlock->info.window.skey != tsCols[0] || pSDataBlock->info.window.ekey != tsCols[endRowId]) {
qError("table uid %" PRIu64 " data block timestamp range may not be calculated! minKey %" PRId64 qError("table uid %" PRIu64 " data block timestamp range may not be calculated! minKey %" PRId64 ",maxKey %" PRId64,
",maxKey %" PRId64, pSDataBlock->info.id.uid, pSDataBlock->info.window.skey, pSDataBlock->info.window.ekey);
pSDataBlock->info.id.uid, pSDataBlock->info.window.skey, pSDataBlock->info.window.ekey);
blockDataUpdateTsWindow(pSDataBlock, pInfo->primaryTsIndex); blockDataUpdateTsWindow(pSDataBlock, pInfo->primaryTsIndex);
// timestamp of the data is incorrect // timestamp of the data is incorrect
if (pSDataBlock->info.window.skey <= 0 || pSDataBlock->info.window.ekey <= 0) { if (pSDataBlock->info.window.skey <= 0 || pSDataBlock->info.window.ekey <= 0) {
qError("table uid %" PRIu64 " data block timestamp is out of range! minKey %" PRId64 ",maxKey %" PRId64, qError("table uid %" PRIu64 " data block timestamp is out of range! minKey %" PRId64 ",maxKey %" PRId64,
pSDataBlock->info.id.uid, pSDataBlock->info.window.skey, pSDataBlock->info.window.ekey); pSDataBlock->info.id.uid, pSDataBlock->info.window.skey, pSDataBlock->info.window.ekey);
} }
} }
@ -918,7 +923,7 @@ static void doStreamIntervalAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDat
if (savePullWindow(&pull, pInfo->pPullWins) == TSDB_CODE_SUCCESS) { if (savePullWindow(&pull, pInfo->pPullWins) == TSDB_CODE_SUCCESS) {
addPullWindow(pInfo->pPullDataMap, &winRes, pInfo->numOfChild); addPullWindow(pInfo->pPullDataMap, &winRes, pInfo->numOfChild);
if (pInfo->destHasPrimaryKey) { if (pInfo->destHasPrimaryKey) {
tSimpleHashPut(pInfo->pDeletedMap,&winRes, sizeof(SWinKey), NULL, 0); tSimpleHashPut(pInfo->pDeletedMap, &winRes, sizeof(SWinKey), NULL, 0);
} }
} }
} else { } else {
@ -964,7 +969,7 @@ static void doStreamIntervalAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDat
}; };
if (pInfo->destHasPrimaryKey && code == TSDB_CODE_SUCCESS && IS_NORMAL_INTERVAL_OP(pOperator)) { if (pInfo->destHasPrimaryKey && code == TSDB_CODE_SUCCESS && IS_NORMAL_INTERVAL_OP(pOperator)) {
tSimpleHashPut(pDeletedMap,&key, sizeof(SWinKey), NULL, 0); tSimpleHashPut(pDeletedMap, &key, sizeof(SWinKey), NULL, 0);
} }
if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE && pUpdatedMap) { if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE && pUpdatedMap) {
@ -1367,7 +1372,7 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
pInfo->binfo.pRes->info.type = pBlock->info.type; pInfo->binfo.pRes->info.type = pBlock->info.type;
} else if (pBlock->info.type == STREAM_DELETE_DATA || pBlock->info.type == STREAM_DELETE_RESULT || } else if (pBlock->info.type == STREAM_DELETE_DATA || pBlock->info.type == STREAM_DELETE_RESULT ||
pBlock->info.type == STREAM_CLEAR) { pBlock->info.type == STREAM_CLEAR) {
SArray* delWins = taosArrayInit(8, sizeof(SWinKey)); SArray* delWins = taosArrayInit(8, sizeof(SWinKey));
SHashObj* finalMap = IS_FINAL_INTERVAL_OP(pOperator) ? pInfo->pFinalPullDataMap : NULL; SHashObj* finalMap = IS_FINAL_INTERVAL_OP(pOperator) ? pInfo->pFinalPullDataMap : NULL;
doDeleteWindows(pOperator, &pInfo->interval, pBlock, delWins, pInfo->pUpdatedMap, finalMap); doDeleteWindows(pOperator, &pInfo->interval, pBlock, delWins, pInfo->pUpdatedMap, finalMap);
if (IS_FINAL_INTERVAL_OP(pOperator)) { if (IS_FINAL_INTERVAL_OP(pOperator)) {
@ -1401,7 +1406,7 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
getAllIntervalWindow(pInfo->aggSup.pResultRowHashTable, pInfo->pUpdatedMap); getAllIntervalWindow(pInfo->aggSup.pResultRowHashTable, pInfo->pUpdatedMap);
continue; continue;
} else if (pBlock->info.type == STREAM_RETRIEVE) { } else if (pBlock->info.type == STREAM_RETRIEVE) {
if(!IS_FINAL_INTERVAL_OP(pOperator)) { if (!IS_FINAL_INTERVAL_OP(pOperator)) {
pInfo->recvRetrive = true; pInfo->recvRetrive = true;
copyDataBlock(pInfo->pMidRetriveRes, pBlock); copyDataBlock(pInfo->pMidRetriveRes, pBlock);
pInfo->pMidRetriveRes->info.type = STREAM_MID_RETRIEVE; pInfo->pMidRetriveRes->info.type = STREAM_MID_RETRIEVE;
@ -1527,8 +1532,8 @@ void streamIntervalReloadState(SOperatorInfo* pOperator) {
SStreamIntervalOperatorInfo* pInfo = pOperator->info; SStreamIntervalOperatorInfo* pInfo = pOperator->info;
if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL && if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL &&
pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_MID_INTERVAL) { pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_MID_INTERVAL) {
int32_t size = 0; int32_t size = 0;
void* pBuf = NULL; void* pBuf = NULL;
int32_t code = pInfo->stateStore.streamStateGetInfo(pInfo->pState, STREAM_INTERVAL_OP_STATE_NAME, int32_t code = pInfo->stateStore.streamStateGetInfo(pInfo->pState, STREAM_INTERVAL_OP_STATE_NAME,
strlen(STREAM_INTERVAL_OP_STATE_NAME), &pBuf, &size); strlen(STREAM_INTERVAL_OP_STATE_NAME), &pBuf, &size);
if (code == 0) { if (code == 0) {
@ -1602,7 +1607,7 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
pAPI->stateStore.streamStateSetNumber(pInfo->pState, -1, pInfo->primaryTsIndex); pAPI->stateStore.streamStateSetNumber(pInfo->pState, -1, pInfo->primaryTsIndex);
code = initAggSup(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str, code = initAggSup(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str,
pInfo->pState, &pTaskInfo->storageAPI.functionStore); pInfo->pState, &pTaskInfo->storageAPI.functionStore);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
goto _error; goto _error;
} }
@ -1757,27 +1762,37 @@ void initDummyFunction(SqlFunctionCtx* pDummy, SqlFunctionCtx* pCtx, int32_t num
} }
} }
void initDownStream(SOperatorInfo* downstream, SStreamAggSupporter* pAggSup, uint16_t type, int32_t tsColIndex, int32_t initDownStream(SOperatorInfo* downstream, SStreamAggSupporter* pAggSup, uint16_t type, int32_t tsColIndex,
STimeWindowAggSupp* pTwSup, struct SSteamOpBasicInfo* pBasic) { STimeWindowAggSupp* pTwSup, struct SSteamOpBasicInfo* pBasic) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
if (downstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION) { if (downstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION) {
SStreamPartitionOperatorInfo* pScanInfo = downstream->info; SStreamPartitionOperatorInfo* pScanInfo = downstream->info;
pScanInfo->tsColIndex = tsColIndex; pScanInfo->tsColIndex = tsColIndex;
} }
if (downstream->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) { if (downstream->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
initDownStream(downstream->pDownstream[0], pAggSup, type, tsColIndex, pTwSup, pBasic); code = initDownStream(downstream->pDownstream[0], pAggSup, type, tsColIndex, pTwSup, pBasic);
return; return code;
} }
SStreamScanInfo* pScanInfo = downstream->info; SStreamScanInfo* pScanInfo = downstream->info;
pScanInfo->windowSup = (SWindowSupporter){.pStreamAggSup = pAggSup, .gap = pAggSup->gap, .parentType = type}; pScanInfo->windowSup = (SWindowSupporter){.pStreamAggSup = pAggSup, .gap = pAggSup->gap, .parentType = type};
pScanInfo->pState = pAggSup->pState; pScanInfo->pState = pAggSup->pState;
if (!pScanInfo->pUpdateInfo) { if (!pScanInfo->pUpdateInfo) {
pAggSup->stateStore.updateInfoInit(60000, TSDB_TIME_PRECISION_MILLI, pTwSup->waterMark, code = pAggSup->stateStore.updateInfoInit(60000, TSDB_TIME_PRECISION_MILLI, pTwSup->waterMark,
pScanInfo->igCheckUpdate, pScanInfo->pkColType, pScanInfo->pkColLen, &pScanInfo->pUpdateInfo); pScanInfo->igCheckUpdate, pScanInfo->pkColType, pScanInfo->pkColLen,
&pScanInfo->pUpdateInfo);
TSDB_CHECK_CODE(code, lino, _end);
} }
pScanInfo->twAggSup = *pTwSup; pScanInfo->twAggSup = *pTwSup;
pAggSup->pUpdateInfo = pScanInfo->pUpdateInfo; pAggSup->pUpdateInfo = pScanInfo->pUpdateInfo;
pBasic->primaryPkIndex = pScanInfo->primaryKeyIndex; pBasic->primaryPkIndex = pScanInfo->primaryKeyIndex;
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
} }
static TSKEY sesionTs(void* pKey) { static TSKEY sesionTs(void* pKey) {
@ -2150,8 +2165,8 @@ static void doStreamSessionAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSData
TSKEY* endTsCols = (int64_t*)pEndTsCol->pData; TSKEY* endTsCols = (int64_t*)pEndTsCol->pData;
void* pPkVal = NULL; void* pPkVal = NULL;
int32_t pkLen = 0; int32_t pkLen = 0;
SColumnInfoData* pPkColDataInfo = NULL; SColumnInfoData* pPkColDataInfo = NULL;
if (hasSrcPrimaryKeyCol(&pInfo->basic)) { if (hasSrcPrimaryKeyCol(&pInfo->basic)) {
pPkColDataInfo = taosArrayGet(pSDataBlock->pDataBlock, pInfo->basic.primaryPkIndex); pPkColDataInfo = taosArrayGet(pSDataBlock->pDataBlock, pInfo->basic.primaryPkIndex);
@ -2456,8 +2471,8 @@ int32_t buildSessionResultDataBlock(SOperatorInfo* pOperator, void* pState, SSDa
pBlock->info.id.groupId = pKey->groupId; pBlock->info.id.groupId = pKey->groupId;
void* tbname = NULL; void* tbname = NULL;
if (pAPI->stateStore.streamStateGetParName((void*)pTaskInfo->streamInfo.pState, pBlock->info.id.groupId, if (pAPI->stateStore.streamStateGetParName((void*)pTaskInfo->streamInfo.pState, pBlock->info.id.groupId, &tbname,
&tbname, false) < 0) { false) < 0) {
pBlock->info.parTbName[0] = 0; pBlock->info.parTbName[0] = 0;
} else { } else {
memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN); memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN);
@ -2937,7 +2952,7 @@ void streamSessionSemiReloadState(SOperatorInfo* pOperator) {
int32_t size = 0; int32_t size = 0;
void* pBuf = NULL; void* pBuf = NULL;
int32_t code = pAggSup->stateStore.streamStateGetInfo(pAggSup->pState, STREAM_SESSION_OP_STATE_NAME, int32_t code = pAggSup->stateStore.streamStateGetInfo(pAggSup->pState, STREAM_SESSION_OP_STATE_NAME,
strlen(STREAM_SESSION_OP_STATE_NAME), &pBuf, &size); strlen(STREAM_SESSION_OP_STATE_NAME), &pBuf, &size);
int32_t num = (size - sizeof(TSKEY)) / sizeof(SSessionKey); int32_t num = (size - sizeof(TSKEY)) / sizeof(SSessionKey);
SSessionKey* pSeKeyBuf = (SSessionKey*)pBuf; SSessionKey* pSeKeyBuf = (SSessionKey*)pBuf;
ASSERT(size == num * sizeof(SSessionKey) + sizeof(TSKEY)); ASSERT(size == num * sizeof(SSessionKey) + sizeof(TSKEY));
@ -2970,7 +2985,7 @@ void streamSessionReloadState(SOperatorInfo* pOperator) {
int32_t size = 0; int32_t size = 0;
void* pBuf = NULL; void* pBuf = NULL;
int32_t code = pAggSup->stateStore.streamStateGetInfo(pAggSup->pState, STREAM_SESSION_OP_STATE_NAME, int32_t code = pAggSup->stateStore.streamStateGetInfo(pAggSup->pState, STREAM_SESSION_OP_STATE_NAME,
strlen(STREAM_SESSION_OP_STATE_NAME), &pBuf, &size); strlen(STREAM_SESSION_OP_STATE_NAME), &pBuf, &size);
int32_t num = (size - sizeof(TSKEY)) / sizeof(SSessionKey); int32_t num = (size - sizeof(TSKEY)) / sizeof(SSessionKey);
SSessionKey* pSeKeyBuf = (SSessionKey*)pBuf; SSessionKey* pSeKeyBuf = (SSessionKey*)pBuf;
ASSERT(size == num * sizeof(SSessionKey) + sizeof(TSKEY)); ASSERT(size == num * sizeof(SSessionKey) + sizeof(TSKEY));
@ -3020,6 +3035,7 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh
SSessionWinodwPhysiNode* pSessionNode = (SSessionWinodwPhysiNode*)pPhyNode; SSessionWinodwPhysiNode* pSessionNode = (SSessionWinodwPhysiNode*)pPhyNode;
int32_t numOfCols = 0; int32_t numOfCols = 0;
int32_t code = TSDB_CODE_OUT_OF_MEMORY; int32_t code = TSDB_CODE_OUT_OF_MEMORY;
int32_t lino = 0;
SStreamSessionAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamSessionAggOperatorInfo)); SStreamSessionAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamSessionAggOperatorInfo));
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
if (pInfo == NULL || pOperator == NULL) { if (pInfo == NULL || pOperator == NULL) {
@ -3114,8 +3130,12 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh
setOperatorStreamStateFn(pOperator, streamSessionReleaseState, streamSessionReloadState); setOperatorStreamStateFn(pOperator, streamSessionReleaseState, streamSessionReloadState);
if (downstream) { if (downstream) {
initDownStream(downstream, &pInfo->streamAggSup, pOperator->operatorType, pInfo->primaryTsIndex, &pInfo->twAggSup, &pInfo->basic); code = initDownStream(downstream, &pInfo->streamAggSup, pOperator->operatorType, pInfo->primaryTsIndex,
&pInfo->twAggSup, &pInfo->basic);
TSDB_CHECK_CODE(code, lino, _error);
code = appendDownstream(pOperator, &downstream, 1); code = appendDownstream(pOperator, &downstream, 1);
TSDB_CHECK_CODE(code, lino, _error);
} }
return pOperator; return pOperator;
@ -3126,6 +3146,7 @@ _error:
taosMemoryFreeClear(pOperator); taosMemoryFreeClear(pOperator);
pTaskInfo->code = code; pTaskInfo->code = code;
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
return NULL; return NULL;
} }
@ -3576,7 +3597,7 @@ static void doStreamStateAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl
if (!allEqual) { if (!allEqual) {
uint64_t uid = 0; uint64_t uid = 0;
appendDataToSpecialBlock(pAggSup->pScanBlock, &curWin.winInfo.sessionWin.win.skey, appendDataToSpecialBlock(pAggSup->pScanBlock, &curWin.winInfo.sessionWin.win.skey,
&curWin.winInfo.sessionWin.win.ekey, &uid, &groupId, NULL); &curWin.winInfo.sessionWin.win.ekey, &uid, &groupId, NULL);
tSimpleHashRemove(pSeUpdated, &curWin.winInfo.sessionWin, sizeof(SSessionKey)); tSimpleHashRemove(pSeUpdated, &curWin.winInfo.sessionWin, sizeof(SSessionKey));
doDeleteSessionWindow(pAggSup, &curWin.winInfo.sessionWin); doDeleteSessionWindow(pAggSup, &curWin.winInfo.sessionWin);
releaseOutputBuf(pAggSup->pState, curWin.winInfo.pStatePos, &pAPI->stateStore); releaseOutputBuf(pAggSup->pState, curWin.winInfo.pStatePos, &pAPI->stateStore);
@ -3683,9 +3704,9 @@ void* doStreamStateDecodeOpState(void* buf, int32_t len, SOperatorInfo* pOperato
SSessionKey key = {0}; SSessionKey key = {0};
SResultWindowInfo winfo = {0}; SResultWindowInfo winfo = {0};
buf = decodeSSessionKey(buf, &key); buf = decodeSSessionKey(buf, &key);
pAggSup->stateStore.streamStateStateAddIfNotExist(pAggSup->pState, &winfo.sessionWin, NULL, pAggSup->stateStore.streamStateStateAddIfNotExist(pAggSup->pState, &winfo.sessionWin, NULL, pAggSup->stateKeySize,
pAggSup->stateKeySize, compareStateKey, compareStateKey, (void**)&winfo.pStatePos,
(void**)&winfo.pStatePos, &pAggSup->resultRowSize); &pAggSup->resultRowSize);
buf = decodeSResultWindowInfo(buf, &winfo, pInfo->streamAggSup.resultRowSize); buf = decodeSResultWindowInfo(buf, &winfo, pInfo->streamAggSup.resultRowSize);
tSimpleHashPut(pInfo->streamAggSup.pResultRows, &key, sizeof(SSessionKey), &winfo, sizeof(SResultWindowInfo)); tSimpleHashPut(pInfo->streamAggSup.pResultRows, &key, sizeof(SSessionKey), &winfo, sizeof(SResultWindowInfo));
} }
@ -3879,7 +3900,7 @@ void streamStateReloadState(SOperatorInfo* pOperator) {
int32_t size = 0; int32_t size = 0;
void* pBuf = NULL; void* pBuf = NULL;
int32_t code = pAggSup->stateStore.streamStateGetInfo(pAggSup->pState, STREAM_STATE_OP_STATE_NAME, int32_t code = pAggSup->stateStore.streamStateGetInfo(pAggSup->pState, STREAM_STATE_OP_STATE_NAME,
strlen(STREAM_STATE_OP_STATE_NAME), &pBuf, &size); strlen(STREAM_STATE_OP_STATE_NAME), &pBuf, &size);
int32_t num = (size - sizeof(TSKEY)) / sizeof(SSessionKey); int32_t num = (size - sizeof(TSKEY)) / sizeof(SSessionKey);
qDebug("===stream=== reload state. get result count:%d", num); qDebug("===stream=== reload state. get result count:%d", num);
SSessionKey* pSeKeyBuf = (SSessionKey*)pBuf; SSessionKey* pSeKeyBuf = (SSessionKey*)pBuf;
@ -3943,12 +3964,13 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys
int32_t tsSlotId = ((SColumnNode*)pStateNode->window.pTspk)->slotId; int32_t tsSlotId = ((SColumnNode*)pStateNode->window.pTspk)->slotId;
SColumnNode* pColNode = (SColumnNode*)(pStateNode->pStateKey); SColumnNode* pColNode = (SColumnNode*)(pStateNode->pStateKey);
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SStreamStateAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamStateAggOperatorInfo)); SStreamStateAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamStateAggOperatorInfo));
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
if (pInfo == NULL || pOperator == NULL) { if (pInfo == NULL || pOperator == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
goto _error; TSDB_CHECK_CODE(code, lino, _error);
} }
pInfo->stateCol = extractColumnFromColumnNode(pColNode); pInfo->stateCol = extractColumnFromColumnNode(pColNode);
@ -3957,9 +3979,7 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys
int32_t numOfScalar = 0; int32_t numOfScalar = 0;
SExprInfo* pScalarExprInfo = createExprInfo(pStateNode->window.pExprs, NULL, &numOfScalar); SExprInfo* pScalarExprInfo = createExprInfo(pStateNode->window.pExprs, NULL, &numOfScalar);
code = initExprSupp(&pInfo->scalarSupp, pScalarExprInfo, numOfScalar, &pTaskInfo->storageAPI.functionStore); code = initExprSupp(&pInfo->scalarSupp, pScalarExprInfo, numOfScalar, &pTaskInfo->storageAPI.functionStore);
if (code != TSDB_CODE_SUCCESS) { TSDB_CHECK_CODE(code, lino, _error);
goto _error;
}
} }
pInfo->twAggSup = (STimeWindowAggSupp){ pInfo->twAggSup = (STimeWindowAggSupp){
@ -3986,9 +4006,8 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys
code = initStreamAggSupporter(&pInfo->streamAggSup, pExpSup, numOfCols, 0, pTaskInfo->streamInfo.pState, keySize, code = initStreamAggSupporter(&pInfo->streamAggSup, pExpSup, numOfCols, 0, pTaskInfo->streamInfo.pState, keySize,
type, &pTaskInfo->storageAPI.stateStore, pHandle, &pInfo->twAggSup, type, &pTaskInfo->storageAPI.stateStore, pHandle, &pInfo->twAggSup,
GET_TASKID(pTaskInfo), &pTaskInfo->storageAPI, pInfo->primaryTsIndex); GET_TASKID(pTaskInfo), &pTaskInfo->storageAPI, pInfo->primaryTsIndex);
if (code != TSDB_CODE_SUCCESS) { TSDB_CHECK_CODE(code, lino, _error);
goto _error;
}
_hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY); _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
pInfo->pSeDeleted = tSimpleHashInit(64, hashFn); pInfo->pSeDeleted = tSimpleHashInit(64, hashFn);
pInfo->pDelIterator = NULL; pInfo->pDelIterator = NULL;
@ -4001,7 +4020,8 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys
pInfo->dataVersion = 0; pInfo->dataVersion = 0;
pInfo->historyWins = taosArrayInit(4, sizeof(SSessionKey)); pInfo->historyWins = taosArrayInit(4, sizeof(SSessionKey));
if (!pInfo->historyWins) { if (!pInfo->historyWins) {
goto _error; code = TSDB_CODE_OUT_OF_MEMORY;
TSDB_CHECK_CODE(code, lino, _error);
} }
if (pHandle) { if (pHandle) {
pInfo->isHistoryOp = pHandle->fillHistory; pInfo->isHistoryOp = pHandle->fillHistory;
@ -4028,17 +4048,20 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys
pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamStateAgg, NULL, destroyStreamStateOperatorInfo, pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamStateAgg, NULL, destroyStreamStateOperatorInfo,
optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL); optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
setOperatorStreamStateFn(pOperator, streamStateReleaseState, streamStateReloadState); setOperatorStreamStateFn(pOperator, streamStateReleaseState, streamStateReloadState);
initDownStream(downstream, &pInfo->streamAggSup, pOperator->operatorType, pInfo->primaryTsIndex, &pInfo->twAggSup, &pInfo->basic); code = initDownStream(downstream, &pInfo->streamAggSup, pOperator->operatorType, pInfo->primaryTsIndex,
&pInfo->twAggSup, &pInfo->basic);
TSDB_CHECK_CODE(code, lino, _error);
code = appendDownstream(pOperator, &downstream, 1); code = appendDownstream(pOperator, &downstream, 1);
if (code != TSDB_CODE_SUCCESS) { TSDB_CHECK_CODE(code, lino, _error);
goto _error;
}
return pOperator; return pOperator;
_error: _error:
destroyStreamStateOperatorInfo(pInfo); destroyStreamStateOperatorInfo(pInfo);
taosMemoryFreeClear(pOperator); taosMemoryFreeClear(pOperator);
pTaskInfo->code = code; pTaskInfo->code = code;
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
return NULL; return NULL;
} }
@ -4205,11 +4228,12 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhys
.precision = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->node.resType.precision, .precision = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->node.resType.precision,
}; };
pInfo->twAggSup = (STimeWindowAggSupp){.waterMark = pIntervalPhyNode->window.watermark, pInfo->twAggSup =
.calTrigger = pIntervalPhyNode->window.triggerType, (STimeWindowAggSupp){.waterMark = pIntervalPhyNode->window.watermark,
.maxTs = INT64_MIN, .calTrigger = pIntervalPhyNode->window.triggerType,
.minTs = INT64_MAX, .maxTs = INT64_MIN,
.deleteMark = getDeleteMark(&pIntervalPhyNode->window, pIntervalPhyNode->interval)}; .minTs = INT64_MAX,
.deleteMark = getDeleteMark(&pIntervalPhyNode->window, pIntervalPhyNode->interval)};
ASSERTS(pInfo->twAggSup.calTrigger != STREAM_TRIGGER_MAX_DELAY, "trigger type should not be max delay"); ASSERTS(pInfo->twAggSup.calTrigger != STREAM_TRIGGER_MAX_DELAY, "trigger type should not be max delay");
@ -4337,7 +4361,7 @@ static void doStreamMidIntervalAggImpl(SOperatorInfo* pOperator, SSDataBlock* pS
.ts = nextWin.skey, .ts = nextWin.skey,
.groupId = groupId, .groupId = groupId,
}; };
void* chIds = taosHashGet(pInfo->pPullDataMap, &key, sizeof(SWinKey)); void* chIds = taosHashGet(pInfo->pPullDataMap, &key, sizeof(SWinKey));
int32_t index = -1; int32_t index = -1;
SArray* chArray = NULL; SArray* chArray = NULL;
int32_t chId = 0; int32_t chId = 0;
@ -4409,8 +4433,8 @@ static void doStreamMidIntervalAggImpl(SOperatorInfo* pOperator, SSDataBlock* pS
static void addMidRetriveWindow(SArray* wins, SHashObj* pMidPullMap, int32_t numOfChild) { static void addMidRetriveWindow(SArray* wins, SHashObj* pMidPullMap, int32_t numOfChild) {
int32_t size = taosArrayGetSize(wins); int32_t size = taosArrayGetSize(wins);
for (int32_t i = 0; i < size; i++) { for (int32_t i = 0; i < size; i++) {
SWinKey* winKey = taosArrayGet(wins, i); SWinKey* winKey = taosArrayGet(wins, i);
void* chIds = taosHashGet(pMidPullMap, winKey, sizeof(SWinKey)); void* chIds = taosHashGet(pMidPullMap, winKey, sizeof(SWinKey));
if (!chIds) { if (!chIds) {
addPullWindow(pMidPullMap, winKey, numOfChild); addPullWindow(pMidPullMap, winKey, numOfChild);
qDebug("===stream===prepare mid operator retrive for delete %" PRId64 ", size:%d", winKey->ts, numOfChild); qDebug("===stream===prepare mid operator retrive for delete %" PRId64 ", size:%d", winKey->ts, numOfChild);
@ -4613,5 +4637,6 @@ static SSDataBlock* doStreamMidIntervalAgg(SOperatorInfo* pOperator) {
void setStreamOperatorCompleted(SOperatorInfo* pOperator) { void setStreamOperatorCompleted(SOperatorInfo* pOperator) {
setOperatorCompleted(pOperator); setOperatorCompleted(pOperator);
qDebug("stask:%s %s status: %d. set completed", GET_TASKID(pOperator->pTaskInfo), getStreamOpName(pOperator->operatorType), pOperator->status); qDebug("stask:%s %s status: %d. set completed", GET_TASKID(pOperator->pTaskInfo),
getStreamOpName(pOperator->operatorType), pOperator->status);
} }

101
source/libs/stream/src/streamUpdate.c
View File

@ -304,6 +304,7 @@ TSKEY updateInfoFillBlockData(SUpdateInfo* pInfo, SSDataBlock* pBlock, int32_t p
len = colDataGetRowLength(pPkDataInfo, i); len = colDataGetRowLength(pPkDataInfo, i);
} }
int32_t buffLen = getKeyBuff(ts, tbUid, pPkVal, len, pInfo->pKeyBuff); int32_t buffLen = getKeyBuff(ts, tbUid, pPkVal, len, pInfo->pKeyBuff);
// we don't care whether the data is updated or not
tScalableBfPut(pSBf, pInfo->pKeyBuff, buffLen); tScalableBfPut(pSBf, pInfo->pKeyBuff, buffLen);
} }
} }
@ -413,59 +414,119 @@ void updateInfoDestoryColseWinSBF(SUpdateInfo* pInfo) {
pInfo->pCloseWinSBF = NULL; pInfo->pCloseWinSBF = NULL;
} }
int32_t updateInfoSerialize(void* buf, int32_t bufLen, const SUpdateInfo* pInfo) { int32_t updateInfoSerialize(void* buf, int32_t bufLen, const SUpdateInfo* pInfo, int32_t* pLen) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
if (!pInfo) { if (!pInfo) {
return 0; return TSDB_CODE_SUCCESS;
} }
SEncoder encoder = {0}; SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen); tEncoderInit(&encoder, buf, bufLen);
if (tStartEncode(&encoder) < 0) return -1; if (tStartEncode(&encoder) < 0) {
code = TSDB_CODE_FAILED;
TSDB_CHECK_CODE(code, lino, _end);
}
int32_t size = taosArrayGetSize(pInfo->pTsBuckets); int32_t size = taosArrayGetSize(pInfo->pTsBuckets);
if (tEncodeI32(&encoder, size) < 0) return -1; if (tEncodeI32(&encoder, size) < 0) {
code = TSDB_CODE_FAILED;
TSDB_CHECK_CODE(code, lino, _end);
}
for (int32_t i = 0; i < size; i++) { for (int32_t i = 0; i < size; i++) {
TSKEY* pTs = (TSKEY*)taosArrayGet(pInfo->pTsBuckets, i); TSKEY* pTs = (TSKEY*)taosArrayGet(pInfo->pTsBuckets, i);
if (tEncodeI64(&encoder, *pTs) < 0) return -1; if (tEncodeI64(&encoder, *pTs) < 0) {
code = TSDB_CODE_FAILED;
TSDB_CHECK_CODE(code, lino, _end);
}
} }
if (tEncodeU64(&encoder, pInfo->numBuckets) < 0) return -1; if (tEncodeU64(&encoder, pInfo->numBuckets) < 0) {
code = TSDB_CODE_FAILED;
TSDB_CHECK_CODE(code, lino, _end);
}
int32_t sBfSize = taosArrayGetSize(pInfo->pTsSBFs); int32_t sBfSize = taosArrayGetSize(pInfo->pTsSBFs);
if (tEncodeI32(&encoder, sBfSize) < 0) return -1; if (tEncodeI32(&encoder, sBfSize) < 0) {
code = TSDB_CODE_FAILED;
TSDB_CHECK_CODE(code, lino, _end);
}
for (int32_t i = 0; i < sBfSize; i++) { for (int32_t i = 0; i < sBfSize; i++) {
SScalableBf* pSBf = taosArrayGetP(pInfo->pTsSBFs, i); SScalableBf* pSBf = taosArrayGetP(pInfo->pTsSBFs, i);
if (tScalableBfEncode(pSBf, &encoder) < 0) return -1; if (tScalableBfEncode(pSBf, &encoder) < 0) {
code = TSDB_CODE_FAILED;
TSDB_CHECK_CODE(code, lino, _end);
}
} }
if (tEncodeU64(&encoder, pInfo->numSBFs) < 0) return -1; if (tEncodeU64(&encoder, pInfo->numSBFs) < 0) {
if (tEncodeI64(&encoder, pInfo->interval) < 0) return -1; code = TSDB_CODE_FAILED;
if (tEncodeI64(&encoder, pInfo->watermark) < 0) return -1; TSDB_CHECK_CODE(code, lino, _end);
if (tEncodeI64(&encoder, pInfo->minTS) < 0) return -1; }
if (tEncodeI64(&encoder, pInfo->interval) < 0) {
code = TSDB_CODE_FAILED;
TSDB_CHECK_CODE(code, lino, _end);
}
if (tEncodeI64(&encoder, pInfo->watermark) < 0) {
code = TSDB_CODE_FAILED;
TSDB_CHECK_CODE(code, lino, _end);
}
if (tEncodeI64(&encoder, pInfo->minTS) < 0) {
code = TSDB_CODE_FAILED;
TSDB_CHECK_CODE(code, lino, _end);
}
if (tScalableBfEncode(pInfo->pCloseWinSBF, &encoder) < 0) return -1; if (tScalableBfEncode(pInfo->pCloseWinSBF, &encoder) < 0) {
code = TSDB_CODE_FAILED;
TSDB_CHECK_CODE(code, lino, _end);
}
int32_t mapSize = taosHashGetSize(pInfo->pMap); int32_t mapSize = taosHashGetSize(pInfo->pMap);
if (tEncodeI32(&encoder, mapSize) < 0) return -1; if (tEncodeI32(&encoder, mapSize) < 0) {
code = TSDB_CODE_FAILED;
TSDB_CHECK_CODE(code, lino, _end);
}
void* pIte = NULL; void* pIte = NULL;
size_t keyLen = 0; size_t keyLen = 0;
while ((pIte = taosHashIterate(pInfo->pMap, pIte)) != NULL) { while ((pIte = taosHashIterate(pInfo->pMap, pIte)) != NULL) {
void* key = taosHashGetKey(pIte, &keyLen); void* key = taosHashGetKey(pIte, &keyLen);
if (tEncodeU64(&encoder, *(uint64_t*)key) < 0) return -1; if (tEncodeU64(&encoder, *(uint64_t*)key) < 0) {
code = TSDB_CODE_FAILED;
TSDB_CHECK_CODE(code, lino, _end);
}
int32_t valueSize = taosHashGetValueSize(pIte); int32_t valueSize = taosHashGetValueSize(pIte);
if (tEncodeBinary(&encoder, (const uint8_t*)pIte, valueSize) < 0) return -1; if (tEncodeBinary(&encoder, (const uint8_t*)pIte, valueSize) < 0) {
code = TSDB_CODE_FAILED;
TSDB_CHECK_CODE(code, lino, _end);
}
} }
if (tEncodeU64(&encoder, pInfo->maxDataVersion) < 0) return -1; if (tEncodeU64(&encoder, pInfo->maxDataVersion) < 0) {
code = TSDB_CODE_FAILED;
TSDB_CHECK_CODE(code, lino, _end);
}
if (tEncodeI32(&encoder, pInfo->pkColLen) < 0) return -1; if (tEncodeI32(&encoder, pInfo->pkColLen) < 0) {
if (tEncodeI8(&encoder, pInfo->pkColType) < 0) return -1; code = TSDB_CODE_FAILED;
TSDB_CHECK_CODE(code, lino, _end);
}
if (tEncodeI8(&encoder, pInfo->pkColType) < 0) {
code = TSDB_CODE_FAILED;
TSDB_CHECK_CODE(code, lino, _end);
}
tEndEncode(&encoder); tEndEncode(&encoder);
int32_t tlen = encoder.pos; int32_t tlen = encoder.pos;
tEncoderClear(&encoder); tEncoderClear(&encoder);
return tlen; *pLen = tlen;
_end:
if (code != TSDB_CODE_SUCCESS) {
uError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
return code;
} }
int32_t updateInfoDeserialize(void* buf, int32_t bufLen, SUpdateInfo* pInfo) { int32_t updateInfoDeserialize(void* buf, int32_t bufLen, SUpdateInfo* pInfo) {