From 47cb043f4cfa6aaa8f02e2d4943139d603886455 Mon Sep 17 00:00:00 2001 From: 54liuyao <54liuyao@163.com> Date: Fri, 22 Jul 2022 16:55:14 +0800 Subject: [PATCH] feat(stream):optimize update scan range --- source/libs/executor/inc/executorimpl.h | 20 +- source/libs/executor/src/executorimpl.c | 3 +- source/libs/executor/src/scanoperator.c | 388 ++++++------------ source/libs/executor/src/timewindowoperator.c | 103 +++-- source/libs/function/src/builtinsimpl.c | 20 +- tests/script/tsim/stream/sliding.sim | 4 +- 6 files changed, 206 insertions(+), 332 deletions(-) diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index 21068c68a4..34aa74454a 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -52,11 +52,12 @@ typedef int32_t (*__block_search_fn_t)(char* data, int32_t num, int64_t key, int #define NEEDTO_COMPRESS_QUERY(size) ((size) > tsCompressColData ? 1 : 0) -#define START_TS_COLUMN_INDEX 0 -#define END_TS_COLUMN_INDEX 1 -#define UID_COLUMN_INDEX 2 -#define GROUPID_COLUMN_INDEX UID_COLUMN_INDEX -#define DELETE_GROUPID_COLUMN_INDEX 2 +#define START_TS_COLUMN_INDEX 0 +#define END_TS_COLUMN_INDEX 1 +#define UID_COLUMN_INDEX 2 +#define GROUPID_COLUMN_INDEX 3 +#define CALCULATE_START_TS_COLUMN_INDEX 4 +#define CALCULATE_END_TS_COLUMN_INDEX 5 enum { // when this task starts to execute, this status will set @@ -346,7 +347,6 @@ typedef enum EStreamScanMode { STREAM_SCAN_FROM_READERHANDLE = 1, STREAM_SCAN_FROM_RES, STREAM_SCAN_FROM_UPDATERES, - STREAM_SCAN_FROM_DATAREADER, // todo(liuyao) delete it STREAM_SCAN_FROM_DATAREADER_RETRIEVE, STREAM_SCAN_FROM_DATAREADER_RANGE, } EStreamScanMode; @@ -366,7 +366,7 @@ typedef struct SStreamAggSupporter { char* pKeyBuf; // window key buffer SDiskbasedBuf* pResultBuf; // query result buffer based on blocked-wised disk file int32_t resultRowSize; // the result buffer size for each result row, with the meta data size for each row - SArray* pScanWindow; + SSDataBlock* pScanBlock; } SStreamAggSupporter; typedef struct SessionWindowSupporter { @@ -419,7 +419,7 @@ typedef struct SStreamScanInfo { int32_t deleteDataIndex; STimeWindow updateWin; STimeWindowAggSupp twAggSup; - + SSDataBlock* pUpdateDataRes; // status for tmq // SSchemaWrapper schema; STqOffset offset; @@ -712,7 +712,6 @@ typedef struct SStreamStateAggOperatorInfo { SSDataBlock* pDelRes; SHashObj* pSeDeleted; void* pDelIterator; - SArray* pScanWindow; SArray* pChildren; // cache for children's result; bool ignoreExpiredData; } SStreamStateAggOperatorInfo; @@ -954,6 +953,7 @@ int32_t updateSessionWindowInfo(SResultWindowInfo* pWinInfo, TSKEY* pStartTs, TSKEY* pEndTs, int32_t rows, int32_t start, int64_t gap, SHashObj* pStDeleted); bool functionNeedToExecute(SqlFunctionCtx* pCtx); bool isCloseWindow(STimeWindow* pWin, STimeWindowAggSupp* pSup); +void appendOneRow(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, uint64_t* pUid); int32_t finalizeResultRowIntoResultDataBlock(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPosition, SqlFunctionCtx* pCtx, SExprInfo* pExprInfo, int32_t numOfExprs, const int32_t* rowCellOffset, @@ -970,7 +970,7 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_t tsColIndex); int32_t generateGroupIdMap(STableListInfo* pTableListInfo, SReadHandle* pHandle, SNodeList* groupKey); -SSDataBlock* createPullDataBlock(); +SSDataBlock* createSpecialDataBlock(EStreamType type); #ifdef __cplusplus } diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 44a78f2a0b..38a50dbbd7 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -5136,8 +5136,7 @@ int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, const char* pKey, SqlF } pSup->valueSize = size; - pSup->pScanWindow = taosArrayInit(4, sizeof(STimeWindow)); - + pSup->pScanBlock = createSpecialDataBlock(STREAM_CLEAR); int32_t pageSize = 4096; while (pageSize < pSup->resultRowSize * 4) { pageSize <<= 1u; diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index fc29eed455..f31c138cf0 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -25,7 +25,6 @@ #include "tdatablock.h" #include "tmsg.h" -#include "executorimpl.h" #include "query.h" #include "tcompare.h" #include "thash.h" @@ -812,6 +811,10 @@ static bool isSignleIntervalWindow(SStreamScanInfo* pInfo) { return pInfo->sessionSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL; } +static bool isSlidingWindow(SStreamScanInfo* pInfo) { + return isIntervalWindow(pInfo) && pInfo->interval.interval != pInfo->interval.sliding; +} + static uint64_t getGroupId(SOperatorInfo* pOperator, uint64_t uid) { uint64_t* groupId = taosHashGet(pOperator->pTaskInfo->tableqinfoList.map, &uid, sizeof(int64_t)); if (groupId) { @@ -834,17 +837,10 @@ static uint64_t getGroupId(SOperatorInfo* pOperator, uint64_t uid) { } static void setGroupId(SStreamScanInfo* pInfo, SSDataBlock* pBlock, int32_t groupColIndex, int32_t rowIndex) { + SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, groupColIndex); + uint64_t* groupCol = (uint64_t*)pColInfo->pData; ASSERT(rowIndex < pBlock->info.rows); - switch (pBlock->info.type) { - case STREAM_DELETE_DATA: - case STREAM_RETRIEVE: { - SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, groupColIndex); - uint64_t* groupCol = (uint64_t*)pColInfo->pData; - pInfo->groupId = groupCol[rowIndex]; - } break; - default: - break; - } + pInfo->groupId = groupCol[rowIndex]; } void resetTableScanInfo(STableScanInfo* pTableScanInfo, STimeWindow* pWin) { @@ -864,7 +860,17 @@ static bool prepareRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pBlock, int32_ SColumnInfoData* pEndTsCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX); TSKEY* endData = (TSKEY*)pEndTsCol->pData; STimeWindow win = {.skey = startData[*pRowIndex], .ekey = endData[*pRowIndex]}; + + SColumnInfoData* pCalStartTsCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX); + TSKEY* calStartData = (TSKEY*)pCalStartTsCol->pData; + SColumnInfoData* pCalEndTsCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX); + TSKEY* calEndData = (TSKEY*)pCalEndTsCol->pData; + setGroupId(pInfo, pBlock, GROUPID_COLUMN_INDEX, *pRowIndex); + if (isSlidingWindow(pInfo)) { + pInfo->updateWin.skey = calStartData[*pRowIndex]; + pInfo->updateWin.ekey = calEndData[*pRowIndex]; + } (*pRowIndex)++; for (; *pRowIndex < pBlock->info.rows; (*pRowIndex)++) { @@ -876,8 +882,8 @@ static bool prepareRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pBlock, int32_ win.skey = TMIN(win.skey, startData[*pRowIndex]); continue; } - ASSERT((win.skey > startData[*pRowIndex] && win.ekey < endData[*pRowIndex]) || - (isInTimeWindow(&win, startData[*pRowIndex], 0) || isInTimeWindow(&win, endData[*pRowIndex], 0))); + ASSERT(!(win.skey > startData[*pRowIndex] && win.ekey < endData[*pRowIndex]) || + !(isInTimeWindow(&win, startData[*pRowIndex], 0) || isInTimeWindow(&win, endData[*pRowIndex], 0))); break; } @@ -908,68 +914,6 @@ static STimeWindow getSlidingWindow(TSKEY* tsCol, SInterval* pInterval, SDataBlo win.ekey = endWin.ekey; } } -static bool prepareDataScan(SStreamScanInfo* pInfo, SSDataBlock* pSDB, int32_t tsColIndex, int32_t* pRowIndex) { - STimeWindow win = { - .skey = INT64_MIN, - .ekey = INT64_MAX, - }; - bool needRead = false; - if (!isStateWindow(pInfo) && (*pRowIndex) < pSDB->info.rows) { - SColumnInfoData* pColDataInfo = taosArrayGet(pSDB->pDataBlock, tsColIndex); - TSKEY* tsCols = (TSKEY*)pColDataInfo->pData; - SResultRowInfo dumyInfo; - dumyInfo.cur.pageId = -1; - if (isSessionWindow(pInfo)) { - SStreamAggSupporter* pAggSup = pInfo->sessionSup.pStreamAggSup; - int64_t gap = pInfo->sessionSup.gap; - int32_t winIndex = 0; - SResultWindowInfo* pCurWin = - getSessionTimeWindow(pAggSup, tsCols[*pRowIndex], INT64_MIN, pSDB->info.groupId, gap, &winIndex); - win = pCurWin->win; - setGroupId(pInfo, pSDB, GROUPID_COLUMN_INDEX, *pRowIndex); - (*pRowIndex) += updateSessionWindowInfo(pCurWin, tsCols, NULL, pSDB->info.rows, *pRowIndex, gap, NULL); - } else { - setGroupId(pInfo, pSDB, GROUPID_COLUMN_INDEX, *pRowIndex); - pInfo->updateWin.skey = tsCols[*pRowIndex]; - win = getSlidingWindow(tsCols, &pInfo->interval, &pSDB->info, pRowIndex); - pInfo->updateWin.ekey = tsCols[*pRowIndex - 1]; - // win = getActiveTimeWindow(NULL, &dumyInfo, tsCols[*pRowIndex], &pInfo->interval, TSDB_ORDER_ASC); - // (*pRowIndex) += - // getNumOfRowsInTimeWindow(&pSDB->info, tsCols, *pRowIndex, win.ekey, binarySearchForKey, NULL, - // TSDB_ORDER_ASC); - } - needRead = true; - } else if (isStateWindow(pInfo)) { - SArray* pWins = pInfo->sessionSup.pStreamAggSup->pScanWindow; - int32_t size = taosArrayGetSize(pWins); - if (pInfo->scanWinIndex < size) { - win = *(STimeWindow*)taosArrayGet(pWins, pInfo->scanWinIndex); - pInfo->scanWinIndex++; - needRead = true; - } else { - pInfo->scanWinIndex = 0; - taosArrayClear(pWins); - } - } - if (!needRead) { - return false; - } - resetTableScanInfo(pInfo->pTableScanOp->info, &win); - return true; -} - -static void copyOneRow(SSDataBlock* dest, SSDataBlock* source, int32_t sourceRowId) { - for (int32_t j = 0; j < taosArrayGetSize(source->pDataBlock); j++) { - SColumnInfoData* pDestCol = (SColumnInfoData*)taosArrayGet(dest->pDataBlock, j); - SColumnInfoData* pSourceCol = (SColumnInfoData*)taosArrayGet(source->pDataBlock, j); - if (colDataIsNull_s(pSourceCol, sourceRowId)) { - colDataAppendNULL(pDestCol, dest->info.rows); - } else { - colDataAppend(pDestCol, dest->info.rows, colDataGetData(pSourceCol, sourceRowId), false); - } - } - dest->info.rows++; -} static SSDataBlock* doRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pSDB, int32_t tsColIndex, int32_t* pRowIndex) { while (1) { @@ -982,29 +926,6 @@ static SSDataBlock* doRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pSDB, int32 if (!pResult) { blockDataCleanup(pSDB); *pRowIndex = 0; - STableScanInfo* pTableScanInfo = pInfo->pTableScanOp->info; - tsdbReaderClose(pTableScanInfo->dataReader); - pTableScanInfo->dataReader = NULL; - return NULL; - } - - if (pResult->info.groupId == pInfo->groupId) { - return pResult; - } - } -} - -static SSDataBlock* doDataScan(SStreamScanInfo* pInfo, SSDataBlock* pSDB, int32_t tsColIndex, int32_t* pRowIndex) { - while (1) { - SSDataBlock* pResult = NULL; - pResult = doTableScan(pInfo->pTableScanOp); - if (pResult == NULL) { - if (prepareDataScan(pInfo, pSDB, tsColIndex, pRowIndex)) { - // scan next window data - pResult = doTableScan(pInfo->pTableScanOp); - } - } - if (!pResult) { pInfo->updateWin = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX}; STableScanInfo* pTableScanInfo = pInfo->pTableScanOp->info; tsdbReaderClose(pTableScanInfo->dataReader); @@ -1017,77 +938,31 @@ static SSDataBlock* doDataScan(SStreamScanInfo* pInfo, SSDataBlock* pSDB, int32_ return pResult; } } - - /* Todo(liuyao) for partition by column - SSDataBlock* pBlock = createOneDataBlock(pResult, true); - blockDataCleanup(pResult); - for (int32_t i = 0; i < pBlock->info.rows; i++) { - uint64_t id = getGroupId(pInfo->pOperatorDumy, pBlock->info.uid); - if (id == pInfo->groupId) { - copyOneRow(pResult, pBlock, i); - } - } - return pResult; - */ } -static void generateIntervalTs(SStreamScanInfo* pInfo, SSDataBlock* pDelBlock, SOperatorInfo* pOperator, - SSDataBlock* pUpdateRes) { - if (pDelBlock->info.rows == 0) { - return; - } - blockDataCleanup(pUpdateRes); - blockDataEnsureCapacity(pUpdateRes, 64); - ASSERT(taosArrayGetSize(pDelBlock->pDataBlock) >= 3); - SColumnInfoData* pStartTsCol = taosArrayGet(pDelBlock->pDataBlock, START_TS_COLUMN_INDEX); - TSKEY* startData = (TSKEY*)pStartTsCol->pData; - SColumnInfoData* pEndTsCol = taosArrayGet(pDelBlock->pDataBlock, END_TS_COLUMN_INDEX); - TSKEY* endData = (TSKEY*)pEndTsCol->pData; - SColumnInfoData* pGpCol = taosArrayGet(pDelBlock->pDataBlock, UID_COLUMN_INDEX); - uint64_t* uidCol = (uint64_t*)pGpCol->pData; - - SColumnInfoData* pDestTsCol = taosArrayGet(pUpdateRes->pDataBlock, START_TS_COLUMN_INDEX); - SColumnInfoData* pDestGpCol = taosArrayGet(pUpdateRes->pDataBlock, GROUPID_COLUMN_INDEX); - for (int32_t i = pInfo->deleteDataIndex; - i < pDelBlock->info.rows && - i < pDelBlock->info.capacity - (endData[i] - startData[i]) / pInfo->interval.interval - 1; - i++) { - uint64_t groupId = getGroupId(pOperator, uidCol[i]); - for (TSKEY startTs = startData[i]; startTs <= endData[i];) { - colDataAppend(pDestTsCol, pUpdateRes->info.rows, (const char*)&startTs, false); - colDataAppend(pDestGpCol, pUpdateRes->info.rows, (const char*)&groupId, false); - pUpdateRes->info.rows++; - startTs = taosTimeAdd(startTs, pInfo->interval.interval, pInfo->interval.intervalUnit, pInfo->interval.precision); - } - pInfo->deleteDataIndex++; - } - if (pInfo->deleteDataIndex > 0 && pInfo->deleteDataIndex == pDelBlock->info.rows) { - blockDataCleanup(pDelBlock); - pInfo->deleteDataIndex = 0; +static int32_t generateSessionScanRange(SStreamScanInfo* pInfo, SSDataBlock* pSrcBlock, SSDataBlock* pDestBlock) { + if (pSrcBlock->info.rows == 0) { + return TSDB_CODE_SUCCESS; } -} - -static void generateScanRange(SStreamScanInfo* pInfo, SSDataBlock* pBlock, SOperatorInfo* pOperator, - SSDataBlock* pUpdateRes) { - if (pBlock->info.rows == 0) { - return; + blockDataCleanup(pDestBlock); + int32_t code = blockDataEnsureCapacity(pDestBlock, pSrcBlock->info.rows); + if (code != TSDB_CODE_SUCCESS) { + return code; } - blockDataCleanup(pUpdateRes); - blockDataEnsureCapacity(pUpdateRes, pBlock->info.rows); - ASSERT(taosArrayGetSize(pBlock->pDataBlock) >= 3); - SColumnInfoData* pStartTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX); + ASSERT(taosArrayGetSize(pSrcBlock->pDataBlock) >= 3); + SColumnInfoData* pStartTsCol = taosArrayGet(pSrcBlock->pDataBlock, START_TS_COLUMN_INDEX); TSKEY* startData = (TSKEY*)pStartTsCol->pData; - SColumnInfoData* pEndTsCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX); + SColumnInfoData* pEndTsCol = taosArrayGet(pSrcBlock->pDataBlock, END_TS_COLUMN_INDEX); TSKEY* endData = (TSKEY*)pEndTsCol->pData; - SColumnInfoData* pGpCol = taosArrayGet(pBlock->pDataBlock, UID_COLUMN_INDEX); - uint64_t* uidCol = (uint64_t*)pGpCol->pData; + SColumnInfoData* pUidCol = taosArrayGet(pSrcBlock->pDataBlock, UID_COLUMN_INDEX); + uint64_t* uidCol = (uint64_t*)pUidCol->pData; - SColumnInfoData* pDestStartCol = taosArrayGet(pUpdateRes->pDataBlock, START_TS_COLUMN_INDEX); - SColumnInfoData* pDestEndCol = taosArrayGet(pUpdateRes->pDataBlock, END_TS_COLUMN_INDEX); - SColumnInfoData* pDestGpCol = taosArrayGet(pUpdateRes->pDataBlock, GROUPID_COLUMN_INDEX); + SColumnInfoData* pDestStartCol = taosArrayGet(pDestBlock->pDataBlock, START_TS_COLUMN_INDEX); + SColumnInfoData* pDestEndCol = taosArrayGet(pDestBlock->pDataBlock, END_TS_COLUMN_INDEX); + SColumnInfoData* pDestGpCol = taosArrayGet(pDestBlock->pDataBlock, GROUPID_COLUMN_INDEX); int32_t dummy = 0; - for (int32_t i = 0; i < pBlock->info.rows; i++) { - uint64_t groupId = getGroupId(pOperator, uidCol[i]); + for (int32_t i = 0; i < pSrcBlock->info.rows; i++) { + uint64_t groupId = getGroupId(pInfo->pTableScanOp, uidCol[i]); // gap must be 0. SResultWindowInfo* pStartWin = getCurSessionWindow(pInfo->sessionSup.pStreamAggSup, startData[i], endData[i], groupId, 0, &dummy); @@ -1101,46 +976,75 @@ static void generateScanRange(SStreamScanInfo* pInfo, SSDataBlock* pBlock, SOper colDataAppend(pDestStartCol, i, (const char*)&pStartWin->win.skey, false); colDataAppend(pDestEndCol, i, (const char*)&pEndWin->win.ekey, false); colDataAppend(pDestGpCol, i, (const char*)&groupId, false); - pUpdateRes->info.rows++; - } -} -static void setUpdateData(SStreamScanInfo* pInfo, SSDataBlock* pBlock, SSDataBlock* pUpdateBlock) { - blockDataCleanup(pUpdateBlock); - int32_t size = taosArrayGetSize(pInfo->tsArray); - if (pInfo->tsArrayIndex < size) { - SColumnInfoData* pCol = (SColumnInfoData*)taosArrayGet(pUpdateBlock->pDataBlock, pInfo->primaryTsIndex); - ASSERT(pCol->info.type == TSDB_DATA_TYPE_TIMESTAMP); - blockDataEnsureCapacity(pUpdateBlock, size); - - int32_t rowId = *(int32_t*)taosArrayGet(pInfo->tsArray, pInfo->tsArrayIndex); - pInfo->groupId = getGroupId(pInfo->pTableScanOp, pBlock->info.uid); - int32_t i = 0; - for (; i < size; i++) { - rowId = *(int32_t*)taosArrayGet(pInfo->tsArray, i + pInfo->tsArrayIndex); - uint64_t id = getGroupId(pInfo->pTableScanOp, pBlock->info.uid); - if (pInfo->groupId != id) { - break; - } - copyOneRow(pUpdateBlock, pBlock, rowId); - } - pUpdateBlock->info.rows = i; - pInfo->tsArrayIndex += i; - pUpdateBlock->info.groupId = pInfo->groupId; - pUpdateBlock->info.type = STREAM_CLEAR; - blockDataUpdateTsWindow(pUpdateBlock, 0); + pDestBlock->info.rows++; } - // all rows have same group id - ASSERT(pInfo->tsArrayIndex >= size); - if (size > 0 && pInfo->tsArrayIndex == size) { - taosArrayClear(pInfo->tsArray); + return TSDB_CODE_SUCCESS; +} + +static int32_t generateIntervalScanRange(SStreamScanInfo* pInfo, SSDataBlock* pSrcBlock, SSDataBlock* pDestBlock) { + blockDataCleanup(pDestBlock); + int32_t rows = pSrcBlock->info.rows; + if (rows == 0) { + return TSDB_CODE_SUCCESS; + } + int32_t code = blockDataEnsureCapacity(pDestBlock, rows); + if (code != TSDB_CODE_SUCCESS) { + return code; } - if (size == 0) { - generateIntervalTs(pInfo, pInfo->pDeleteDataRes, pInfo->pTableScanOp, pUpdateBlock); + SColumnInfoData* pTsCol = (SColumnInfoData*)taosArrayGet(pSrcBlock->pDataBlock, START_TS_COLUMN_INDEX); + SColumnInfoData* pUidCol = taosArrayGet(pSrcBlock->pDataBlock, UID_COLUMN_INDEX); + uint64_t* uidCol = (uint64_t*)pUidCol->pData; + ASSERT(pTsCol->info.type == TSDB_DATA_TYPE_TIMESTAMP); + TSKEY* tsCol = (TSKEY*)pTsCol->pData; + SColumnInfoData* pStartTsCol = taosArrayGet(pDestBlock->pDataBlock, START_TS_COLUMN_INDEX); + SColumnInfoData* pEndTsCol = taosArrayGet(pDestBlock->pDataBlock, END_TS_COLUMN_INDEX); + SColumnInfoData* pGpCol = taosArrayGet(pDestBlock->pDataBlock, GROUPID_COLUMN_INDEX); + SColumnInfoData* pCalStartTsCol = taosArrayGet(pDestBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX); + SColumnInfoData* pCalEndTsCol = taosArrayGet(pDestBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX); + uint64_t groupId = getGroupId(pInfo->pTableScanOp, uidCol[0]); + for (int32_t i = 0; i < rows; ) { + colDataAppend(pCalStartTsCol, pDestBlock->info.rows, (const char*)(tsCol + i), false); + STimeWindow win = getSlidingWindow(tsCol, &pInfo->interval, &pSrcBlock->info, &i); + colDataAppend(pCalEndTsCol, pDestBlock->info.rows, (const char*)(tsCol + i - 1), false); + + colDataAppend(pStartTsCol, pDestBlock->info.rows, (const char*)(&win.skey), false); + colDataAppend(pEndTsCol, pDestBlock->info.rows, (const char*)(&win.ekey), false); + colDataAppend(pGpCol, pDestBlock->info.rows, (const char*)(&groupId), false); + pDestBlock->info.rows++; } + // all rows have same group id + pDestBlock->info.groupId = groupId; + return TSDB_CODE_SUCCESS; +} + +static int32_t generateScanRange(SStreamScanInfo* pInfo, SSDataBlock* pSrcBlock, SSDataBlock* pDestBlock) { + int32_t code = TSDB_CODE_SUCCESS; + if (isIntervalWindow(pInfo)) { + code = generateIntervalScanRange(pInfo, pSrcBlock, pDestBlock); + } else { + code = generateSessionScanRange(pInfo, pSrcBlock, pDestBlock); + } + pDestBlock->info.type = STREAM_CLEAR; + blockDataUpdateTsWindow(pDestBlock, 0); + return code; +} + +void appendOneRow(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, uint64_t* pUid) { + SColumnInfoData* pStartTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX); + SColumnInfoData* pEndTsCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX); + SColumnInfoData* pUidCol = taosArrayGet(pBlock->pDataBlock, UID_COLUMN_INDEX); + colDataAppend(pStartTsCol, pBlock->info.rows, (const char*)pStartTs, false); + colDataAppend(pEndTsCol, pBlock->info.rows, (const char*)pEndTs, false); + colDataAppend(pUidCol, pBlock->info.rows, (const char*)pUid, false); + pBlock->info.rows++; } static void checkUpdateData(SStreamScanInfo* pInfo, bool invertible, SSDataBlock* pBlock, bool out) { + if (out) { + blockDataCleanup(pInfo->pUpdateDataRes); + blockDataEnsureCapacity(pInfo->pUpdateDataRes, pBlock->info.rows); + } SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, pInfo->primaryTsIndex); ASSERT(pColDataInfo->info.type == TSDB_DATA_TYPE_TIMESTAMP); TSKEY* tsCol = (TSKEY*)pColDataInfo->pData; @@ -1151,9 +1055,13 @@ static void checkUpdateData(SStreamScanInfo* pInfo, bool invertible, SSDataBlock // must check update info first. bool update = updateInfoIsUpdated(pInfo->pUpdateInfo, pBlock->info.uid, tsCol[rowId]); if ((update || (isSignleIntervalWindow(pInfo) && isCloseWindow(&win, &pInfo->twAggSup))) && out) { - taosArrayPush(pInfo->tsArray, &rowId); + appendOneRow(pInfo->pUpdateDataRes, tsCol + rowId, tsCol + rowId, &pBlock->info.uid); } } + if (out) { + blockDataUpdateTsWindow(pInfo->pUpdateDataRes, 0); + pInfo->pUpdateDataRes->info.type = STREAM_CLEAR; + } } static void setBlockGroupId(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t uidColIndex) { @@ -1319,26 +1227,18 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { case STREAM_RETRIEVE: { pInfo->blockType = STREAM_INPUT__DATA_SUBMIT; pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RETRIEVE; - copyDataBlock(pInfo->pPullDataRes, pBlock); - pInfo->pullDataResIndex = 0; - prepareDataScan(pInfo, pInfo->pPullDataRes, START_TS_COLUMN_INDEX, &pInfo->pullDataResIndex); + copyDataBlock(pInfo->pUpdateRes, pBlock); + prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex); updateInfoAddCloseWindowSBF(pInfo->pUpdateInfo); } break; case STREAM_DELETE_DATA: { pInfo->blockType = STREAM_INPUT__DATA_SUBMIT; pInfo->updateResIndex = 0; - if (isIntervalWindow(pInfo)) { - copyDataBlock(pInfo->pDeleteDataRes, pBlock); - generateIntervalTs(pInfo, pInfo->pDeleteDataRes, pInfo->pTableScanOp, pInfo->pUpdateRes); - prepareDataScan(pInfo, pInfo->pUpdateRes, START_TS_COLUMN_INDEX, &pInfo->updateResIndex); - pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER; - } else { - generateScanRange(pInfo, pBlock, pInfo->pTableScanOp, pInfo->pUpdateRes); - prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex); - pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RANGE; - } - pInfo->pUpdateRes->info.type = STREAM_DELETE_DATA; - return pInfo->pUpdateRes; + generateScanRange(pInfo, pBlock, pInfo->pUpdateRes); + prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex); + copyDataBlock(pInfo->pDeleteDataRes, pInfo->pUpdateRes); + pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RANGE; + return pInfo->pDeleteDataRes; } break; default: break; @@ -1351,51 +1251,27 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE; return pInfo->pRes; } else if (pInfo->scanMode == STREAM_SCAN_FROM_UPDATERES) { - if (isStateWindow(pInfo)) { - pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE; - } else { - pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER; - prepareDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex); - } + generateScanRange(pInfo, pInfo->pUpdateDataRes, pInfo->pUpdateRes); + pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RANGE; + prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex); return pInfo->pUpdateRes; - } else if (pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER_RETRIEVE) { - SSDataBlock* pSDB = doDataScan(pInfo, pInfo->pPullDataRes, 0, &pInfo->pullDataResIndex); - if (pSDB != NULL) { - checkUpdateData(pInfo, true, pSDB, false); - pSDB->info.type = STREAM_PULL_DATA; - return pSDB; - } - pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER; - } else if (pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER) { - SSDataBlock* pSDB = doDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex); - if (pSDB) { - pSDB->info.type = STREAM_NORMAL; - checkUpdateData(pInfo, true, pSDB, false); - return pSDB; - } - setUpdateData(pInfo, pInfo->pRes, pInfo->pUpdateRes); - if (pInfo->pUpdateRes->info.rows > 0) { - prepareDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex); - return pInfo->pUpdateRes; - } - pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE; - } else if (pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER_RANGE) { + } else if (pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER_RANGE || pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER_RETRIEVE) { SSDataBlock* pSDB = doRangeScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex); if (pSDB) { - pSDB->info.type = STREAM_NORMAL; + pSDB->info.type = pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER_RANGE ? STREAM_NORMAL : STREAM_PULL_DATA; checkUpdateData(pInfo, true, pSDB, false); return pSDB; } pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE; - } else if (isStateWindow(pInfo)) { - pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER; - pInfo->updateResIndex = pInfo->pUpdateRes->info.rows; - if (prepareDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex)) { - blockDataCleanup(pInfo->pUpdateRes); - // return empty data blcok - return pInfo->pUpdateRes; - } - pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE; + } + + if (isStateWindow(pInfo) && pInfo->sessionSup.pStreamAggSup->pScanBlock->info.rows > 0) { + pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RANGE; + pInfo->updateResIndex = 0; + copyDataBlock(pInfo->pUpdateRes, pInfo->sessionSup.pStreamAggSup->pScanBlock); + blockDataCleanup(pInfo->sessionSup.pStreamAggSup->pScanBlock); + prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex); + return pInfo->pUpdateRes; } SDataBlockInfo* pBlockInfo = &pInfo->pRes->info; @@ -1454,15 +1330,14 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { } else if (pInfo->pUpdateInfo) { pInfo->tsArrayIndex = 0; checkUpdateData(pInfo, true, pInfo->pRes, true); - setUpdateData(pInfo, pInfo->pRes, pInfo->pUpdateRes); pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlockInfo->window.ekey); - if (pInfo->pUpdateRes->info.rows > 0) { - if (pInfo->pUpdateRes->info.type == STREAM_CLEAR) { + if (pInfo->pUpdateDataRes->info.rows > 0) { + if (pInfo->pUpdateDataRes->info.type == STREAM_CLEAR) { pInfo->updateResIndex = 0; pInfo->scanMode = STREAM_SCAN_FROM_UPDATERES; - } else if (pInfo->pUpdateRes->info.type == STREAM_INVERT) { + } else if (pInfo->pUpdateDataRes->info.type == STREAM_INVERT) { pInfo->scanMode = STREAM_SCAN_FROM_RES; - return pInfo->pUpdateRes; + return pInfo->pUpdateDataRes; } } } @@ -1628,17 +1503,18 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys } pInfo->pRes = createResDataBlock(pDescNode); - pInfo->pUpdateRes = createResDataBlock(pDescNode); + pInfo->pUpdateRes = createSpecialDataBlock(STREAM_CLEAR); pInfo->pCondition = pScanPhyNode->node.pConditions; pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE; pInfo->sessionSup = (SessionWindowSupporter){.pStreamAggSup = NULL, .gap = -1, .parentType = QUERY_NODE_PHYSICAL_PLAN}; pInfo->groupId = 0; - pInfo->pPullDataRes = createPullDataBlock(); + pInfo->pPullDataRes = createSpecialDataBlock(STREAM_RETRIEVE); pInfo->pStreamScanOp = pOperator; pInfo->deleteDataIndex = 0; - pInfo->pDeleteDataRes = createPullDataBlock(); + pInfo->pDeleteDataRes = createSpecialDataBlock(STREAM_DELETE_DATA); pInfo->updateWin = (STimeWindow){.skey = INT64_MAX, .ekey = INT64_MAX}; + pInfo->pUpdateDataRes = createSpecialDataBlock(STREAM_CLEAR); pOperator->name = "StreamScanOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN; diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c index b5966fc463..bc2c812e62 100644 --- a/source/libs/executor/src/timewindowoperator.c +++ b/source/libs/executor/src/timewindowoperator.c @@ -1373,8 +1373,10 @@ void doDeleteSpecifyIntervalWindow(SAggSupporter* pAggSup, SSDataBlock* pBlock, static void doClearWindows(SAggSupporter* pAggSup, SExprSupp* pSup1, SInterval* pInterval, int32_t numOfOutput, SSDataBlock* pBlock, SArray* pUpWins) { - SColumnInfoData* pTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX); - TSKEY* tsCols = (TSKEY*)pTsCol->pData; + 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; uint64_t* pGpDatas = NULL; if (pBlock->info.type == STREAM_RETRIEVE) { SColumnInfoData* pGpCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX); @@ -1382,22 +1384,18 @@ static void doClearWindows(SAggSupporter* pAggSup, SExprSupp* pSup1, SInterval* } int32_t step = 0; int32_t startPos = 0; - SResultRowInfo dumyInfo; - dumyInfo.cur.pageId = -1; - STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, tsCols[0], pInterval, TSDB_ORDER_ASC); - while (1) { - step = - getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, win.ekey, binarySearchForKey, NULL, TSDB_ORDER_ASC); - uint64_t winGpId = pGpDatas ? pGpDatas[startPos] : pBlock->info.groupId; - bool res = doClearWindow(pAggSup, pSup1, (char*)&win.skey, sizeof(TSKEY), winGpId, numOfOutput); - if (pUpWins && res) { - SWinRes winRes = {.ts = win.skey, .groupId = winGpId}; - taosArrayPush(pUpWins, &winRes); - } - int32_t prevEndPos = step - 1 + startPos; - startPos = getNextQualifiedWindow(pInterval, &win, &pBlock->info, tsCols, prevEndPos, TSDB_ORDER_ASC); - if (startPos < 0) { - break; + for (int32_t i = 0; i < pBlock->info.rows; i++) { + SResultRowInfo dumyInfo; + dumyInfo.cur.pageId = -1; + STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, startTsCols[i], pInterval, TSDB_ORDER_ASC); + while (win.ekey <= endTsCols[i]) { + uint64_t winGpId = pGpDatas ? pGpDatas[startPos] : pBlock->info.groupId; + bool res = doClearWindow(pAggSup, pSup1, (char*)&win.skey, sizeof(TSKEY), winGpId, numOfOutput); + if (pUpWins && res) { + SWinRes winRes = {.ts = win.skey, .groupId = winGpId}; + taosArrayPush(pUpWins, &winRes); + } + getNextTimeWindow(pInterval, pInterval->precision, TSDB_ORDER_ASC, &win); } } } @@ -1501,7 +1499,7 @@ static void doBuildDeleteResult(SArray* pWins, int32_t* index, SSDataBlock* pBlo } blockDataEnsureCapacity(pBlock, size - *index); SColumnInfoData* pTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX); - SColumnInfoData* pGroupCol = taosArrayGet(pBlock->pDataBlock, DELETE_GROUPID_COLUMN_INDEX); + SColumnInfoData* pGroupCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX); for (int32_t i = *index; i < size; i++) { SWinRes* pWin = taosArrayGet(pWins, i); colDataAppend(pTsCol, pBlock->info.rows, (const char*)&pWin->ts, false); @@ -1793,10 +1791,7 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pInfo->pRecycledPages = taosArrayInit(4, sizeof(int32_t)); pInfo->pDelWins = taosArrayInit(4, sizeof(SWinRes)); pInfo->delIndex = 0; - // pInfo->pDelRes = createPullDataBlock(); todo(liuyao) for delete - pInfo->pDelRes = createOneDataBlock(pInfo->binfo.pRes, false); // todo(liuyao) for delete - pInfo->pDelRes->info.type = STREAM_DELETE_RESULT; // todo(liuyao) for delete - + pInfo->pDelRes = createSpecialDataBlock(STREAM_DELETE_RESULT); initResultRowInfo(&pInfo->binfo.resultRowInfo); pOperator->name = "TimeIntervalAggOperator"; @@ -2598,14 +2593,6 @@ static void doHashInterval(SOperatorInfo* pOperatorInfo, SSDataBlock* pSDataBloc chId = getChildIndex(pSDataBlock); index = taosArraySearchIdx(chArray, &chId, compareInt32Val, TD_EQ); } - // if (index != -1 && pSDataBlock->info.type == STREAM_PULL_DATA) { - // qDebug("===stream===delete child id %d", chId); - // taosArrayRemove(chArray, index); - // if (taosArrayGetSize(chArray) == 0) { - // // pull data is over - // taosHashRemove(pInfo->pPullDataMap, &winRes, sizeof(SWinRes)); - // } - // } if (index == -1 || pSDataBlock->info.type == STREAM_PULL_DATA) { ignore = false; } @@ -2697,16 +2684,18 @@ static void doBuildPullDataBlock(SArray* array, int32_t* pIndex, SSDataBlock* pB } blockDataEnsureCapacity(pBlock, size - (*pIndex)); ASSERT(3 <= taosArrayGetSize(pBlock->pDataBlock)); + SColumnInfoData* pStartTs = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX); + SColumnInfoData* pEndTs = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX); + SColumnInfoData* pGroupId = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX); + SColumnInfoData* pCalStartTs = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX); + SColumnInfoData* pCalEndTs = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX); for (; (*pIndex) < size; (*pIndex)++) { SPullWindowInfo* pWin = taosArrayGet(array, (*pIndex)); - SColumnInfoData* pStartTs = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX); colDataAppend(pStartTs, pBlock->info.rows, (const char*)&pWin->window.skey, false); - - SColumnInfoData* pEndTs = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX); colDataAppend(pEndTs, pBlock->info.rows, (const char*)&pWin->window.ekey, false); - - SColumnInfoData* pGroupId = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX); colDataAppend(pGroupId, pBlock->info.rows, (const char*)&pWin->groupId, false); + colDataAppend(pCalStartTs, pBlock->info.rows, (const char*)&pWin->window.skey, false); + colDataAppend(pCalEndTs, pBlock->info.rows, (const char*)&pWin->window.ekey, false); pBlock->info.rows++; } if ((*pIndex) == size) { @@ -2825,7 +2814,8 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) { continue; } removeResults(pUpWins, pUpdated); - copyUpdateDataBlock(pInfo->pUpdateRes, pBlock, pInfo->primaryTsIndex); + copyDataBlock(pInfo->pUpdateRes, pBlock); + // copyUpdateDataBlock(pInfo->pUpdateRes, pBlock, pInfo->primaryTsIndex); pInfo->returnUpdate = true; taosArrayDestroy(pUpWins); break; @@ -2933,12 +2923,12 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) { return NULL; } -SSDataBlock* createPullDataBlock() { +SSDataBlock* createSpecialDataBlock(EStreamType type) { SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock)); pBlock->info.hasVarCol = false; pBlock->info.groupId = 0; pBlock->info.rows = 0; - pBlock->info.type = STREAM_RETRIEVE; + pBlock->info.type = type; pBlock->info.rowSize = sizeof(TSKEY) + sizeof(TSKEY) + sizeof(uint64_t); pBlock->pDataBlock = taosArrayInit(3, sizeof(SColumnInfoData)); @@ -2952,6 +2942,14 @@ SSDataBlock* createPullDataBlock() { infoData.info.type = TSDB_DATA_TYPE_UBIGINT; infoData.info.bytes = sizeof(uint64_t); + // uid + taosArrayPush(pBlock->pDataBlock, &infoData); + // group id + taosArrayPush(pBlock->pDataBlock, &infoData); + + // calculate start ts + taosArrayPush(pBlock->pDataBlock, &infoData); + // calculate end ts taosArrayPush(pBlock->pDataBlock, &infoData); return pBlock; @@ -3019,8 +3017,7 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, goto _error; } } - pInfo->pUpdateRes = createResDataBlock(pPhyNode->pOutputDataBlockDesc); - pInfo->pUpdateRes->info.type = STREAM_CLEAR; + pInfo->pUpdateRes = createSpecialDataBlock(STREAM_CLEAR); blockDataEnsureCapacity(pInfo->pUpdateRes, 128); pInfo->returnUpdate = false; @@ -3042,11 +3039,9 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, pInfo->pullIndex = 0; _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY); pInfo->pPullDataMap = taosHashInit(64, hashFn, false, HASH_NO_LOCK); - pInfo->pPullDataRes = createPullDataBlock(); + pInfo->pPullDataRes = createSpecialDataBlock(STREAM_RETRIEVE); pInfo->ignoreExpiredData = pIntervalPhyNode->window.igExpired; - // pInfo->pDelRes = createPullDataBlock(); // todo(liuyao) for delete - pInfo->pDelRes = createOneDataBlock(pInfo->binfo.pRes, false); // todo(liuyao) for delete - pInfo->pDelRes->info.type = STREAM_DELETE_RESULT; // todo(liuyao) for delete + pInfo->pDelRes = createSpecialDataBlock(STREAM_DELETE_RESULT); pInfo->delIndex = 0; pInfo->pDelWins = taosArrayInit(4, sizeof(SWinRes)); @@ -3061,7 +3056,9 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, pOperator->fpSet = createOperatorFpSet(NULL, doStreamFinalIntervalAgg, NULL, NULL, destroyStreamFinalIntervalOperatorInfo, aggEncodeResultRow, aggDecodeResultRow, NULL); - + if (pPhyNode->type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL) { + initIntervalDownStream(downstream, pPhyNode->type); + } code = appendDownstream(pOperator, &downstream, 1); if (code != TSDB_CODE_SUCCESS) { goto _error; @@ -3086,6 +3083,7 @@ void destroyStreamAggSupporter(SStreamAggSupporter* pSup) { } taosHashCleanup(pSup->pResultRows); destroyDiskbasedBuf(pSup->pResultBuf); + blockDataDestroy(pSup->pScanBlock); } void destroyStreamSessionAggOperatorInfo(void* param, int32_t numOfOutput) { @@ -3200,7 +3198,7 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY); pInfo->pStDeleted = taosHashInit(64, hashFn, true, HASH_NO_LOCK); pInfo->pDelIterator = NULL; - // pInfo->pDelRes = createPullDataBlock(); + // pInfo->pDelRes = createSpecialDataBlock(STREAM_DELETE_RESULT); pInfo->pDelRes = createOneDataBlock(pInfo->binfo.pRes, false); // todo(liuyao) for delete pInfo->pDelRes->info.type = STREAM_DELETE_RESULT; // todo(liuyao) for delete pInfo->pChildren = NULL; @@ -3559,7 +3557,7 @@ static void doDeleteTimeWindows(SStreamAggSupporter* pAggSup, SSDataBlock* pBloc TSKEY* startDatas = (TSKEY*)pStartTsCol->pData; SColumnInfoData* pEndTsCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX); TSKEY* endDatas = (TSKEY*)pEndTsCol->pData; - SColumnInfoData* pGroupCol = taosArrayGet(pBlock->pDataBlock, UID_COLUMN_INDEX); + SColumnInfoData* pGroupCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX); uint64_t* gpDatas = (uint64_t*)pGroupCol->pData; for (int32_t i = 0; i < pBlock->info.rows; i++) { int32_t winIndex = 0; @@ -4255,7 +4253,6 @@ static void doClearStateWindows(SStreamAggSupporter* pAggSup, SSDataBlock* pBloc step = updateStateWindowInfo(pAggSup->pCurWins, winIndex, tsCol, pKeyColInfo, pBlock->info.rows, i, &allEqual, pSeDeleted); ASSERT(isTsInWindow(pCurWin, tsCol[i]) || isEqualStateKey(pCurWin, pKeyData)); - taosArrayPush(pAggSup->pScanWindow, &pCurWin->winInfo.win); taosHashRemove(pSeUpdated, &pCurWin->winInfo.pos, sizeof(SResultRowPosition)); deleteWindow(pAggSup->pCurWins, winIndex); } @@ -4280,8 +4277,9 @@ static void doStreamStateAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl } else { return; } - + SStreamAggSupporter* pAggSup = &pInfo->streamAggSup; + blockDataEnsureCapacity(pAggSup->pScanBlock, pSDataBlock->info.rows); SColumnInfoData* pKeyColInfo = taosArrayGet(pSDataBlock->pDataBlock, pInfo->stateCol.slotId); for (int32_t i = 0; i < pSDataBlock->info.rows; i += winRows) { if (pInfo->ignoreExpiredData && isOverdue(tsCols[i], &pInfo->twAggSup)) { @@ -4296,7 +4294,8 @@ static void doStreamStateAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl winRows = updateStateWindowInfo(pAggSup->pCurWins, winIndex, tsCols, pKeyColInfo, pSDataBlock->info.rows, i, &allEqual, pInfo->pSeDeleted); if (!allEqual) { - taosArrayPush(pAggSup->pScanWindow, &pCurWin->winInfo.win); + appendOneRow(pAggSup->pScanBlock, &pCurWin->winInfo.win.skey, &pCurWin->winInfo.win.ekey, + &pSDataBlock->info.groupId); taosHashRemove(pSeUpdated, &pCurWin->winInfo.pos, sizeof(SResultRowPosition)); deleteWindow(pAggSup->pCurWins, winIndex); continue; @@ -4460,7 +4459,7 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY); pInfo->pSeDeleted = taosHashInit(64, hashFn, true, HASH_NO_LOCK); pInfo->pDelIterator = NULL; - // pInfo->pDelRes = createPullDataBlock(); // todo(liuyao) for delete + // pInfo->pDelRes = createSpecialDataBlock(STREAM_DELETE_RESULT); pInfo->pDelRes = createOneDataBlock(pInfo->binfo.pRes, false); // todo(liuyao) for delete pInfo->pDelRes->info.type = STREAM_DELETE_RESULT; // todo(liuyao) for delete pInfo->pChildren = NULL; diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c index 05f84df7f8..2ea9652a4a 100644 --- a/source/libs/function/src/builtinsimpl.c +++ b/source/libs/function/src/builtinsimpl.c @@ -476,16 +476,16 @@ int32_t functionFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) { int32_t firstCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) { SResultRowEntryInfo* pDResInfo = GET_RES_INFO(pDestCtx); - char* pDBuf = GET_ROWCELL_INTERBUF(pDResInfo); + SFirstLastRes* pDBuf = GET_ROWCELL_INTERBUF(pDResInfo); int32_t type = pDestCtx->input.pData[0]->info.type; int32_t bytes = pDestCtx->input.pData[0]->info.bytes; SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx); - char* pSBuf = GET_ROWCELL_INTERBUF(pSResInfo); + SFirstLastRes* pSBuf = GET_ROWCELL_INTERBUF(pSResInfo); - if (pSResInfo->numOfRes != 0 && (pDResInfo->numOfRes == 0 || *(TSKEY*)(pDBuf + bytes) > *(TSKEY*)(pSBuf + bytes))) { - memcpy(pDBuf, pSBuf, bytes); - *(TSKEY*)(pDBuf + bytes) = *(TSKEY*)(pSBuf + bytes); + if (pSResInfo->numOfRes != 0 && (pDResInfo->numOfRes == 0 || pDBuf->ts > pSBuf->ts)) { + memcpy(pDBuf->buf, pSBuf->buf, bytes); + pDBuf->ts = pSBuf->ts; pDResInfo->numOfRes = 1; } return TSDB_CODE_SUCCESS; @@ -2994,16 +2994,16 @@ int32_t firstLastPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) { // todo rewrite: int32_t lastCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) { SResultRowEntryInfo* pDResInfo = GET_RES_INFO(pDestCtx); - char* pDBuf = GET_ROWCELL_INTERBUF(pDResInfo); + SFirstLastRes* pDBuf = GET_ROWCELL_INTERBUF(pDResInfo); int32_t type = pDestCtx->input.pData[0]->info.type; int32_t bytes = pDestCtx->input.pData[0]->info.bytes; SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx); - char* pSBuf = GET_ROWCELL_INTERBUF(pSResInfo); + SFirstLastRes* pSBuf = GET_ROWCELL_INTERBUF(pSResInfo); - if (pSResInfo->numOfRes != 0 && (pDResInfo->numOfRes == 0 || *(TSKEY*)(pDBuf + bytes) < *(TSKEY*)(pSBuf + bytes))) { - memcpy(pDBuf, pSBuf, bytes); - *(TSKEY*)(pDBuf + bytes) = *(TSKEY*)(pSBuf + bytes); + if (pSResInfo->numOfRes != 0 && (pDResInfo->numOfRes == 0 || pDBuf->ts < pSBuf->ts)) { + memcpy(pDBuf->buf, pSBuf->buf, bytes); + pDBuf->ts = pSBuf->ts; pDResInfo->numOfRes = 1; } return TSDB_CODE_SUCCESS; diff --git a/tests/script/tsim/stream/sliding.sim b/tests/script/tsim/stream/sliding.sim index 4364b56d44..8ebadbfb50 100644 --- a/tests/script/tsim/stream/sliding.sim +++ b/tests/script/tsim/stream/sliding.sim @@ -367,7 +367,7 @@ if $data32 != 8 then endi #$loop_all = 0 -#looptest: +#=looptest: sql drop database IF EXISTS test2; sql drop stream IF EXISTS streams21; @@ -511,6 +511,6 @@ endi $loop_all = $loop_all + 1 print ============loop_all=$loop_all -#goto looptest +#=goto looptest system sh/stop_dnodes.sh \ No newline at end of file -- GitLab