From fac36002c1bd834c2396bb7ece3bebc307188216 Mon Sep 17 00:00:00 2001 From: liuyao <54liuyao@163.com> Date: Thu, 20 Mar 2025 10:34:31 +0800 Subject: [PATCH] ci(stream): add ci and fix issue (#30270) * ci(stream): add ci and fix issue * ci: adj test * ci: add test and fix issue * ci: adj test * ci(stream): add test for fill operator and fix issue * fix: fix issue * fix: fix issue * fix: adjust defalut rec value --- source/libs/executor/inc/executorInt.h | 25 +- source/libs/executor/inc/streaminterval.h | 1 + source/libs/executor/inc/streamsession.h | 1 + source/libs/executor/src/projectoperator.c | 4 + source/libs/executor/src/streamclient.c | 67 ++++- .../executor/src/streamfillnonblockoperator.c | 53 +++- source/libs/executor/src/streamfilloperator.c | 15 +- .../src/streamintervalnonblockoperator.c | 23 +- .../src/streamintervalsliceoperator.c | 6 +- source/libs/executor/src/streamscanoperator.c | 37 ++- .../executor/src/streamtimesliceoperator.c | 10 +- source/libs/parser/src/parTranslater.c | 9 +- source/libs/stream/src/streamSessionState.c | 7 +- source/libs/stream/src/tstreamFileState.c | 14 +- tests/parallel_test/longtimeruning_cases.task | 2 + tests/pytest/util/common.py | 4 +- .../tsim/stream/nonblockIntervalBasic.sim | 16 +- .../continuous_window_close_interval.py | 217 ++++++++++++++++ .../continuous_window_close_interval_basic.py | 242 ++++++++++++++++++ 19 files changed, 684 insertions(+), 69 deletions(-) create mode 100644 tests/system-test/8-stream/continuous_window_close_interval.py create mode 100644 tests/system-test/8-stream/continuous_window_close_interval_basic.py diff --git a/source/libs/executor/inc/executorInt.h b/source/libs/executor/inc/executorInt.h index e6be2b34a8..dc46374729 100644 --- a/source/libs/executor/inc/executorInt.h +++ b/source/libs/executor/inc/executorInt.h @@ -510,18 +510,19 @@ typedef struct SStreamFillSupporter { } SStreamFillSupporter; typedef struct SStreamRecParam { - char pSql[2048]; - int32_t sqlCapcity; - char pUrl[TSDB_EP_LEN + 17]; // "http://localhost:6041/rest/sql" - char pAuth[512 + 22]; // Authorization: Basic token - char pStbFullName[TSDB_TABLE_FNAME_LEN]; - char pWstartName[TSDB_COL_NAME_LEN]; - char pWendName[TSDB_COL_NAME_LEN]; - char pGroupIdName[TSDB_COL_NAME_LEN]; - char pIsWindowFilledName[TSDB_COL_NAME_LEN]; - void* pIteData; - int32_t iter; - TSKEY gap; + char pSql[2048]; + int32_t sqlCapcity; + char pUrl[TSDB_EP_LEN + 17]; // "http://localhost:6041/rest/sql" + char pAuth[512 + 22]; // Authorization: Basic token + char pStbFullName[TSDB_TABLE_FNAME_LEN]; + char pWstartName[TSDB_COL_NAME_LEN]; + char pWendName[TSDB_COL_NAME_LEN]; + char pGroupIdName[TSDB_COL_NAME_LEN]; + char pIsWindowFilledName[TSDB_COL_NAME_LEN]; + void* pIteData; + int32_t iter; + TSKEY gap; + SSHashObj* pColIdMap; } SStreamRecParam; typedef struct SStreamScanInfo { diff --git a/source/libs/executor/inc/streaminterval.h b/source/libs/executor/inc/streaminterval.h index 7fe42c5fe1..5f7e79bd3f 100644 --- a/source/libs/executor/inc/streaminterval.h +++ b/source/libs/executor/inc/streaminterval.h @@ -71,6 +71,7 @@ void setDeleteFillValueInfo(TSKEY start, TSKEY end, SStreamFillSupporter* void doStreamFillRange(SStreamFillInfo* pFillInfo, SStreamFillSupporter* pFillSup, SSDataBlock* pRes); int32_t initFillSupRowInfo(SStreamFillSupporter* pFillSup, SSDataBlock* pRes); void getStateKeepInfo(SNonBlockAggSupporter* pNbSup, bool isRecOp, int32_t* pNumRes, TSKEY* pTsRes); +int32_t initStreamFillOperatorColumnMapInfo(SExprSupp* pExprSup, SOperatorInfo* pOperator); #ifdef __cplusplus } diff --git a/source/libs/executor/inc/streamsession.h b/source/libs/executor/inc/streamsession.h index 8ae1c4eb3a..714175de61 100644 --- a/source/libs/executor/inc/streamsession.h +++ b/source/libs/executor/inc/streamsession.h @@ -81,6 +81,7 @@ void setEventWindowInfo(SStreamAggSupporter* pAggSup, SSessionKey* pKey, SRowBuf // stream client int32_t streamClientGetResultRange(SStreamRecParam* pParam, SSHashObj* pRangeMap, SArray* pRangeRes); int32_t streamClientGetFillRange(SStreamRecParam* pParam, SWinKey* pKey, SArray* pRangeRes, void* pEmptyRow, int32_t size, int32_t* pOffsetInfo, int32_t numOfCols); +int32_t streamClientCheckCfg(SStreamRecParam* pParam); #ifdef __cplusplus } diff --git a/source/libs/executor/src/projectoperator.c b/source/libs/executor/src/projectoperator.c index b8b9860e26..c471225ad3 100644 --- a/source/libs/executor/src/projectoperator.c +++ b/source/libs/executor/src/projectoperator.c @@ -18,6 +18,7 @@ #include "functionMgt.h" #include "operator.h" #include "querytask.h" +#include "streaminterval.h" #include "taoserror.h" #include "tdatablock.h" @@ -162,6 +163,9 @@ int32_t createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhysiNode* code = setRowTsColumnOutputInfo(pOperator->exprSupp.pCtx, numOfCols, &pInfo->pPseudoColInfo); TSDB_CHECK_CODE(code, lino, _error); + code = initStreamFillOperatorColumnMapInfo(&pOperator->exprSupp, downstream); + TSDB_CHECK_CODE(code, lino, _error); + setOperatorInfo(pOperator, "ProjectOperator", QUERY_NODE_PHYSICAL_PLAN_PROJECT, false, OP_NOT_OPENED, pInfo, pTaskInfo); pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doProjectOperation, NULL, destroyProjectOperatorInfo, diff --git a/source/libs/executor/src/streamclient.c b/source/libs/executor/src/streamclient.c index 920dd57e7d..29f048ba99 100644 --- a/source/libs/executor/src/streamclient.c +++ b/source/libs/executor/src/streamclient.c @@ -98,7 +98,7 @@ static int32_t doProcessSql(SStreamRecParam* pParam, SJson** ppJsonResult) { curlRes = curl_easy_setopt(pCurl, CURLOPT_POSTFIELDS, pParam->pSql); QUERY_CHECK_CONDITION(curlRes == CURLE_OK, code, lino, _end, TSDB_CODE_FAILED); - qTrace("===stream=== sql:%s", pParam->pSql); + qDebug("===stream=== sql:%s", pParam->pSql); curlRes = curl_easy_setopt(pCurl, CURLOPT_FOLLOWLOCATION, 1L); QUERY_CHECK_CONDITION(curlRes == CURLE_OK, code, lino, _end, TSDB_CODE_FAILED); @@ -110,7 +110,11 @@ static int32_t doProcessSql(SStreamRecParam* pParam, SJson** ppJsonResult) { QUERY_CHECK_CONDITION(curlRes == CURLE_OK, code, lino, _end, TSDB_CODE_FAILED); curlRes = curl_easy_perform(pCurl); - QUERY_CHECK_CONDITION(curlRes == CURLE_OK, code, lino, _end, TSDB_CODE_FAILED); + if (curlRes != CURLE_OK) { + qError("error: unable to request data from %s.since %s. res code:%d", pParam->pUrl, curl_easy_strerror(curlRes), + (int32_t)curlRes); + QUERY_CHECK_CONDITION(curlRes == CURLE_OK, code, lino, _end, TSDB_CODE_FAILED); + } _end: if (pHeaders != NULL) { @@ -222,8 +226,16 @@ static int32_t jsonToDataCell(const SJson* pJson, SResultCellData* pCell) { return code; } +static int32_t getColumnIndex(SSHashObj* pMap, int32_t colId) { + void* pVal = tSimpleHashGet(pMap, &colId, sizeof(int32_t)); + if (pVal == NULL) { + return -1; + } + return *(int32_t*)pVal; +} + static int32_t doTransformFillResult(const SJson* pJsonResult, SArray* pRangeRes, void* pEmptyRow, int32_t size, - int32_t* pOffsetInfo, int32_t numOfCols) { + int32_t* pOffsetInfo, int32_t numOfCols, SSHashObj* pMap) { int32_t code = TSDB_CODE_SUCCESS; int32_t lino = 0; @@ -239,13 +251,21 @@ static int32_t doTransformFillResult(const SJson* pJsonResult, SArray* pRangeRes SSliceRowData* pRowData = taosMemoryCalloc(1, sizeof(TSKEY) + size); pRowData->key = INT64_MIN; memcpy(pRowData->pRowVal, pEmptyRow, size); - for (int32_t j = 0; j < cols && j < numOfCols; ++j) { - SJson* pJsonCell = tjsonGetArrayItem(pRow, j); - QUERY_CHECK_NULL(pJsonCell, code, lino, _end, TSDB_CODE_FAILED); - + int32_t colOffset = 0; + for (int32_t j = 0; j < numOfCols; ++j) { SResultCellData* pDataCell = getSliceResultCell((SResultCellData*)pRowData->pRowVal, j, pOffsetInfo); QUERY_CHECK_NULL(pDataCell, code, lino, _end, TSDB_CODE_FAILED); + int32_t colIndex = getColumnIndex(pMap, j); + if (colIndex == -1 || colIndex >= cols) { + qDebug("invalid result columm index:%d", colIndex); + pDataCell->isNull = true; + continue; + } + + SJson* pJsonCell = tjsonGetArrayItem(pRow, colIndex); + QUERY_CHECK_NULL(pJsonCell, code, lino, _end, TSDB_CODE_FAILED); + code = jsonToDataCell(pJsonCell, pDataCell); QUERY_CHECK_CODE(code, lino, _end); } @@ -278,7 +298,7 @@ int32_t streamClientGetFillRange(SStreamRecParam* pParam, SWinKey* pKey, SArray* SJson* pJsRes = NULL; code = doProcessSql(pParam, &pJsRes); QUERY_CHECK_CODE(code, lino, _end); - code = doTransformFillResult(pJsRes, pRangeRes, pEmptyRow, size, pOffsetInfo, numOfCols); + code = doTransformFillResult(pJsRes, pRangeRes, pEmptyRow, size, pOffsetInfo, numOfCols, pParam->pColIdMap); QUERY_CHECK_CODE(code, lino, _end); _end: @@ -288,6 +308,33 @@ _end: return code; } +int32_t streamClientCheckCfg(SStreamRecParam* pParam) { + int32_t code = TSDB_CODE_SUCCESS; + int32_t lino = 0; + + const char* pTestSql = "select name, ntables, status from information_schema.ins_databases;"; + (void)memset(pParam->pSql, 0, pParam->sqlCapcity); + tstrncpy(pParam->pSql, pTestSql, pParam->sqlCapcity); + + SJson* pJsRes = NULL; + code = doProcessSql(pParam, &pJsRes); + QUERY_CHECK_CODE(code, lino, _end); + SJson* jArray = tjsonGetObjectItem(pJsRes, "data"); + QUERY_CHECK_NULL(jArray, code, lino, _end, TSDB_CODE_FAILED); + + int32_t rows = tjsonGetArraySize(jArray); + if (rows < 2) { + code = TSDB_CODE_INVALID_CFG_VALUE; + qError("invalid taos adapter config value"); + } + +_end: + if (code != TSDB_CODE_SUCCESS) { + qError("%s failed at line %d since %s", __func__, lino, tstrerror(code)); + } + return code; +} + #else int32_t streamClientGetResultRange(SStreamRecParam* pParam, SSHashObj* pRangeMap, SArray* pRangeRes) { @@ -297,4 +344,8 @@ int32_t streamClientGetFillRange(SStreamRecParam* pParam, SWinKey* pKey, SArray* return TSDB_CODE_FAILED; } +int32_t streamClientCheckCfg(SStreamRecParam* pParam) { + return TSDB_CODE_FAILED; +} + #endif \ No newline at end of file diff --git a/source/libs/executor/src/streamfillnonblockoperator.c b/source/libs/executor/src/streamfillnonblockoperator.c index e9f33a5110..53c20c41ae 100644 --- a/source/libs/executor/src/streamfillnonblockoperator.c +++ b/source/libs/executor/src/streamfillnonblockoperator.c @@ -182,6 +182,20 @@ void doBuildNonblockFillResult(SOperatorInfo* pOperator, SStreamFillSupporter* p } } + if (pBlock->info.rows > 0) { + SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; + void* tbname = NULL; + int32_t winCode = TSDB_CODE_SUCCESS; + code = pInfo->stateStore.streamStateGetParName(pTaskInfo->streamInfo.pState, pBlock->info.id.groupId, &tbname, + false, &winCode); + QUERY_CHECK_CODE(code, lino, _end); + if (winCode != TSDB_CODE_SUCCESS) { + pBlock->info.parTbName[0] = 0; + } else { + memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN); + } + } + _end: if (code != TSDB_CODE_SUCCESS) { qError("%s failed at line %d since %s", __func__, lino, tstrerror(code)); @@ -402,8 +416,6 @@ int32_t doStreamNonblockFillNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) case STREAM_INVALID: { code = doApplyStreamScalarCalculation(pOperator, pBlock, pInfo->pSrcBlock); QUERY_CHECK_CODE(code, lino, _end); - - memcpy(pInfo->pSrcBlock->info.parTbName, pBlock->info.parTbName, TSDB_TABLE_NAME_LEN); pInfo->srcRowIndex = -1; } break; case STREAM_CHECKPOINT: { @@ -476,4 +488,39 @@ void destroyStreamNonblockFillOperatorInfo(void* param) { SStreamFillOperatorInfo* pInfo = (SStreamFillOperatorInfo*)param; resetTimeSlicePrevAndNextWindow(pInfo->pFillSup); destroyStreamFillOperatorInfo(param); -} \ No newline at end of file +} + +static int32_t doInitStreamColumnMapInfo(SExprSupp* pExprSup, SSHashObj* pColMap) { + int32_t code = TSDB_CODE_SUCCESS; + int32_t lino = 0; + + for (int32_t i = 0; i < pExprSup->numOfExprs; ++i) { + SExprInfo* pOneExpr = &pExprSup->pExprInfo[i]; + int32_t destSlotId = pOneExpr->base.resSchema.slotId; + for (int32_t j = 0; j < pOneExpr->base.numOfParams; ++j) { + SFunctParam* pFuncParam = &pOneExpr->base.pParam[j]; + if (pFuncParam->type == FUNC_PARAM_TYPE_COLUMN) { + int32_t sourceSlotId = pFuncParam->pCol->slotId; + code = tSimpleHashPut(pColMap, &sourceSlotId, sizeof(int32_t), &destSlotId, sizeof(int32_t)); + QUERY_CHECK_CODE(code, lino, _end); + } + } + } + +_end: + if (code != TSDB_CODE_SUCCESS) { + qError("%s failed at line %d since %s.", __func__, lino, tstrerror(code)); + } + return code; +} + +int32_t initStreamFillOperatorColumnMapInfo(SExprSupp* pExprSup, SOperatorInfo* pOperator) { + if (pOperator != NULL && pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL) { + SStreamFillOperatorInfo* pInfo = (SStreamFillOperatorInfo*)pOperator->info; + if (pInfo->nbSup.recParam.pColIdMap == NULL) { + return TSDB_CODE_SUCCESS; + } + return doInitStreamColumnMapInfo(pExprSup, pInfo->nbSup.recParam.pColIdMap); + } + return TSDB_CODE_SUCCESS; +} diff --git a/source/libs/executor/src/streamfilloperator.c b/source/libs/executor/src/streamfilloperator.c index 847da75619..ea8a5176c8 100644 --- a/source/libs/executor/src/streamfilloperator.c +++ b/source/libs/executor/src/streamfilloperator.c @@ -22,6 +22,7 @@ #include "executorInt.h" #include "streamexecutorInt.h" +#include "streamsession.h" #include "streaminterval.h" #include "tcommon.h" #include "thash.h" @@ -1746,7 +1747,8 @@ static void setValueForFillInfo(SStreamFillSupporter* pFillSup, SStreamFillInfo* } } -int32_t getDownStreamInfo(SOperatorInfo* downstream, int8_t* triggerType, SInterval* pInterval, int16_t* pOperatorFlag) { +int32_t getDownStreamInfo(SOperatorInfo* downstream, int8_t* triggerType, SInterval* pInterval, + int16_t* pOperatorFlag) { int32_t code = TSDB_CODE_SUCCESS; int32_t lino = 0; if (IS_NORMAL_INTERVAL_OP(downstream)) { @@ -1754,17 +1756,16 @@ int32_t getDownStreamInfo(SOperatorInfo* downstream, int8_t* triggerType, SInter *triggerType = pInfo->twAggSup.calTrigger; *pInterval = pInfo->interval; *pOperatorFlag = pInfo->basic.operatorFlag; - } else if (IS_CONTINUE_INTERVAL_OP(downstream)) { + } else { SStreamIntervalSliceOperatorInfo* pInfo = downstream->info; *triggerType = pInfo->twAggSup.calTrigger; *pInterval = pInfo->interval; pInfo->hasFill = true; *pOperatorFlag = pInfo->basic.operatorFlag; - } else { - code = TSDB_CODE_STREAM_INTERNAL_ERROR; } + QUERY_CHECK_CODE(code, lino, _end); - + _end: if (code != TSDB_CODE_SUCCESS) { qError("%s failed at line %d since %s", __func__, lino, tstrerror(code)); @@ -1891,6 +1892,10 @@ int32_t createStreamFillOperatorInfo(SOperatorInfo* downstream, SStreamFillPhysi initNonBlockAggSupptor(&pInfo->nbSup, &pInfo->pFillSup->interval, downstream); code = initStreamBasicInfo(&pInfo->basic, pOperator); QUERY_CHECK_CODE(code, lino, _error); + + code = streamClientCheckCfg(&pInfo->nbSup.recParam); + QUERY_CHECK_CODE(code, lino, _error); + pInfo->basic.operatorFlag = opFlag; if (isFinalOperator(&pInfo->basic)) { pInfo->nbSup.numOfKeep++; diff --git a/source/libs/executor/src/streamintervalnonblockoperator.c b/source/libs/executor/src/streamintervalnonblockoperator.c index 26693eecac..580825eb19 100644 --- a/source/libs/executor/src/streamintervalnonblockoperator.c +++ b/source/libs/executor/src/streamintervalnonblockoperator.c @@ -223,8 +223,10 @@ int32_t doStreamIntervalNonblockAggImpl(SOperatorInfo* pOperator, SSDataBlock* p code = pInfo->streamAggSup.stateStore.streamStateGetAllPrev(pInfo->streamAggSup.pState, &curKey, pInfo->pUpdated, pInfo->nbSup.numOfKeep); QUERY_CHECK_CODE(code, lino, _end); - code = checkAndSaveWinStateToDisc(startIndex, pInfo->pUpdated, 0, pInfo->basic.pTsDataState, &pInfo->streamAggSup, &pInfo->interval); - QUERY_CHECK_CODE(code, lino, _end); + if (!isRecalculateOperator(&pInfo->basic)) { + code = checkAndSaveWinStateToDisc(startIndex, pInfo->pUpdated, 0, pInfo->basic.pTsDataState, &pInfo->streamAggSup, &pInfo->interval); + QUERY_CHECK_CODE(code, lino, _end); + } } } @@ -704,11 +706,14 @@ int32_t doStreamIntervalNonblockAggNext(SOperatorInfo* pOperator, SSDataBlock** if (pBlock == NULL) { qDebug("===stream===%s return data:%s. rev rows:%d", GET_TASKID(pTaskInfo), getStreamOpName(pOperator->operatorType), pInfo->basic.numOfRecv); - if (isFinalOperator(&pInfo->basic) && isRecalculateOperator(&pInfo->basic)) { - code = pAggSup->stateStore.streamStateFlushReaminInfoToDisk(pInfo->basic.pTsDataState); - QUERY_CHECK_CODE(code, lino, _end); - code = buildRetriveRequest(pTaskInfo, pAggSup, pInfo->basic.pTsDataState, &pInfo->nbSup); - QUERY_CHECK_CODE(code, lino, _end); + if (isFinalOperator(&pInfo->basic)) { + if (isRecalculateOperator(&pInfo->basic)) { + code = buildRetriveRequest(pTaskInfo, pAggSup, pInfo->basic.pTsDataState, &pInfo->nbSup); + QUERY_CHECK_CODE(code, lino, _end); + } else { + code = pAggSup->stateStore.streamStateFlushReaminInfoToDisk(pInfo->basic.pTsDataState); + QUERY_CHECK_CODE(code, lino, _end); + } } pOperator->status = OP_RES_TO_RETURN; break; @@ -820,7 +825,7 @@ int32_t doStreamIntervalNonblockAggNext(SOperatorInfo* pOperator, SSDataBlock** code = closeNonblockIntervalWindow(pAggSup->pResultRows, &pInfo->twAggSup, &pInfo->interval, pInfo->pUpdated, pTaskInfo); QUERY_CHECK_CODE(code, lino, _end); - if (!isHistoryOperator(&pInfo->basic)) { + if (!isHistoryOperator(&pInfo->basic) && !isRecalculateOperator(&pInfo->basic)) { code = checkAndSaveWinStateToDisc(0, pInfo->pUpdated, 0, pInfo->basic.pTsDataState, &pInfo->streamAggSup, &pInfo->interval); QUERY_CHECK_CODE(code, lino, _end); } @@ -1052,7 +1057,7 @@ static int32_t doStreamFinalntervalNonblockAggImpl(SOperatorInfo* pOperator, SSD QUERY_CHECK_CODE(code, lino, _end); } - if (!isHistoryOperator(&pInfo->basic)) { + if (!isHistoryOperator(&pInfo->basic) && !isRecalculateOperator(&pInfo->basic)) { code = checkAndSaveWinStateToDisc(0, pInfo->pUpdated, 0, pInfo->basic.pTsDataState, &pInfo->streamAggSup, &pInfo->interval); QUERY_CHECK_CODE(code, lino, _end); } diff --git a/source/libs/executor/src/streamintervalsliceoperator.c b/source/libs/executor/src/streamintervalsliceoperator.c index 60c7ea867d..acbd3a3c76 100644 --- a/source/libs/executor/src/streamintervalsliceoperator.c +++ b/source/libs/executor/src/streamintervalsliceoperator.c @@ -753,18 +753,20 @@ int32_t createStreamIntervalSliceOperatorInfo(SOperatorInfo* downstream, SPhysiN pInfo->hasInterpoFunc = windowinterpNeeded(pExpSup->pCtx, numOfExprs); initNonBlockAggSupptor(&pInfo->nbSup, &pInfo->interval, NULL); - setOperatorInfo(pOperator, "StreamIntervalSliceOperator", pPhyNode->type, true, OP_NOT_OPENED, pInfo, pTaskInfo); + setOperatorInfo(pOperator, "StreamIntervalSliceOperator", nodeType(pPhyNode), true, OP_NOT_OPENED, pInfo, pTaskInfo); code = initStreamBasicInfo(&pInfo->basic, pOperator); QUERY_CHECK_CODE(code, lino, _error); if (pIntervalPhyNode->window.triggerType == STREAM_TRIGGER_CONTINUOUS_WINDOW_CLOSE) { + qDebug("create continuous interval operator. op type:%d, task type:%d, task id:%s", nodeType(pPhyNode), + pHandle->fillHistory, GET_TASKID(pTaskInfo)); if (pHandle->fillHistory == STREAM_HISTORY_OPERATOR) { setFillHistoryOperatorFlag(&pInfo->basic); } else if (pHandle->fillHistory == STREAM_RECALCUL_OPERATOR) { setRecalculateOperatorFlag(&pInfo->basic); } pInfo->nbSup.pWindowAggFn = doStreamIntervalNonblockAggImpl; - if (pPhyNode->type == QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_INTERVAL) { + if (nodeType(pPhyNode) == QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_INTERVAL) { setSingleOperatorFlag(&pInfo->basic); } pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamIntervalNonblockAggNext, NULL, diff --git a/source/libs/executor/src/streamscanoperator.c b/source/libs/executor/src/streamscanoperator.c index 9566b35bab..80d77e7f3c 100644 --- a/source/libs/executor/src/streamscanoperator.c +++ b/source/libs/executor/src/streamscanoperator.c @@ -83,7 +83,8 @@ int32_t copyRecDataToBuff(TSKEY calStart, TSKEY calEnd, uint64_t uid, uint64_t v return pkLen + sizeof(SRecDataInfo); } -int32_t saveRecalculateData(SStateStore* pStateStore, STableTsDataState* pTsDataState, SSDataBlock* pSrcBlock, EStreamType mode) { +int32_t saveRecalculateData(SStateStore* pStateStore, STableTsDataState* pTsDataState, SSDataBlock* pSrcBlock, + EStreamType mode) { int32_t code = TSDB_CODE_SUCCESS; int32_t lino = 0; @@ -92,8 +93,10 @@ int32_t saveRecalculateData(SStateStore* pStateStore, STableTsDataState* pTsData } SColumnInfoData* pSrcStartTsCol = (SColumnInfoData*)taosArrayGet(pSrcBlock->pDataBlock, START_TS_COLUMN_INDEX); SColumnInfoData* pSrcEndTsCol = (SColumnInfoData*)taosArrayGet(pSrcBlock->pDataBlock, END_TS_COLUMN_INDEX); - SColumnInfoData* pSrcCalStartTsCol = (SColumnInfoData*)taosArrayGet(pSrcBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX); - SColumnInfoData* pSrcCalEndTsCol = (SColumnInfoData*)taosArrayGet(pSrcBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX); + SColumnInfoData* pSrcCalStartTsCol = + (SColumnInfoData*)taosArrayGet(pSrcBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX); + SColumnInfoData* pSrcCalEndTsCol = + (SColumnInfoData*)taosArrayGet(pSrcBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX); SColumnInfoData* pSrcUidCol = taosArrayGet(pSrcBlock->pDataBlock, UID_COLUMN_INDEX); SColumnInfoData* pSrcGpCol = taosArrayGet(pSrcBlock->pDataBlock, GROUPID_COLUMN_INDEX); TSKEY* srcStartTsCol = (TSKEY*)pSrcStartTsCol->pData; @@ -113,9 +116,10 @@ int32_t saveRecalculateData(SStateStore* pStateStore, STableTsDataState* pTsData calStart = srcStartTsCol[i]; calEnd = srcEndTsCol[i]; } - int32_t len = copyRecDataToBuff(calStart, calEnd, srcUidData[i], pSrcBlock->info.version, mode, NULL, 0, - pTsDataState->pRecValueBuff, pTsDataState->recValueLen); - code = pStateStore->streamStateSessionSaveToDisk(pTsDataState, &key, pTsDataState->pRecValueBuff, len); + int32_t len = copyRecDataToBuff(calStart, calEnd, srcUidData[i], pSrcBlock->info.version, mode, NULL, 0, + pTsDataState->pRecValueBuff, pTsDataState->recValueLen); + code = pStateStore->streamStateMergeAndSaveScanRange(pTsDataState, &key.win, key.groupId, + pTsDataState->pRecValueBuff, len); QUERY_CHECK_CODE(code, lino, _end); } @@ -332,11 +336,6 @@ static int32_t doStreamBlockScan(SOperatorInfo* pOperator, SSDataBlock** ppRes) case STREAM_CHECKPOINT: { qError("stream check point error. msg type: STREAM_INPUT__DATA_BLOCK"); } break; - case STREAM_RETRIEVE: { - code = saveRecalculateData(&pInfo->stateStore, pInfo->basic.pTsDataState, pBlock, STREAM_RETRIEVE); - QUERY_CHECK_CODE(code, lino, _end); - continue; - } break; case STREAM_RECALCULATE_START: { if (!isSemiOperator(&pInfo->basic)) { code = pInfo->stateStore.streamStateFlushReaminInfoToDisk(pInfo->basic.pTsDataState); @@ -388,7 +387,7 @@ static int32_t buildAndSaveRecalculateData(SSDataBlock* pSrcBlock, TSKEY* pTsCol len = copyRecDataToBuff(pTsCol[rowId], pTsCol[rowId], pSrcBlock->info.id.uid, pSrcBlock->info.version, STREAM_CLEAR, NULL, 0, pTsDataState->pRecValueBuff, pTsDataState->recValueLen); SSessionKey key = {.win.skey = pTsCol[rowId], .win.ekey = pTsCol[rowId], .groupId = 0}; - code = pStateStore->streamStateSessionSaveToDisk(pTsDataState, &key, pTsDataState->pRecValueBuff, len); + code = pStateStore->streamState1SessionSaveToDisk(pTsDataState, &key, pTsDataState->pRecValueBuff, len); QUERY_CHECK_CODE(code, lino, _end); uint64_t gpId = 0; code = appendPkToSpecialBlock(pDestBlock, pTsCol, pPkColDataInfo, rowId, &pSrcBlock->info.id.uid, &gpId, NULL); @@ -399,7 +398,7 @@ static int32_t buildAndSaveRecalculateData(SSDataBlock* pSrcBlock, TSKEY* pTsCol len = copyRecDataToBuff(pTsCol[rowId], pTsCol[rowId], pSrcBlock->info.id.uid, pSrcBlock->info.version, STREAM_DELETE_DATA, NULL, 0, pTsDataState->pRecValueBuff, pTsDataState->recValueLen); - code = pStateStore->streamStateSessionSaveToDisk(pTsDataState, &key, pTsDataState->pRecValueBuff, len); + code = pStateStore->streamState1SessionSaveToDisk(pTsDataState, &key, pTsDataState->pRecValueBuff, len); QUERY_CHECK_CODE(code, lino, _end); code = appendPkToSpecialBlock(pDestBlock, pTsCol, pPkColDataInfo, rowId, &pSrcBlock->info.id.uid, &gpId, NULL); @@ -1252,6 +1251,7 @@ static int32_t doDataRangeScan(SStreamScanInfo* pInfo, SExecTaskInfo* pTaskInfo, if (pInfo->pCreateTbRes->info.rows > 0) { (*ppRes) = pInfo->pCreateTbRes; pInfo->scanMode = STREAM_SCAN_FROM_RES; + break; } (*ppRes) = pTsdbBlock; break; @@ -1341,6 +1341,7 @@ static int32_t doStreamRecalculateDataScan(SOperatorInfo* pOperator, SSDataBlock (*ppRes) = pInfo->pRangeScanRes; pInfo->pRangeScanRes = NULL; pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE; + printDataBlock((*ppRes), "stream tsdb scan", GET_TASKID(pTaskInfo)); goto _end; } break; case STREAM_SCAN_FROM_CREATE_TABLERES: { @@ -1513,6 +1514,11 @@ _end: return code; } +static void destroyStreamRecalculateParam(SStreamRecParam* pParam) { + tSimpleHashCleanup(pParam->pColIdMap); + pParam->pColIdMap = NULL; +} + static void destroyStreamDataScanOperatorInfo(void* param) { if (param == NULL) { return; @@ -1562,6 +1568,8 @@ static void destroyStreamDataScanOperatorInfo(void* param) { taosArrayDestroy(pStreamScan->pRecRangeRes); pStreamScan->pRecRangeRes = NULL; + destroyStreamRecalculateParam(&pStreamScan->recParam); + taosMemoryFree(pStreamScan); } @@ -1633,6 +1641,9 @@ static void initStreamRecalculateParam(STableScanPhysiNode* pTableScanNode, SStr pParam->sqlCapcity = tListLen(pParam->pSql); (void)tsnprintf(pParam->pUrl, tListLen(pParam->pUrl), "http://%s:%d/rest/sql", tsAdapterFqdn, tsAdapterPort); (void)tsnprintf(pParam->pAuth, tListLen(pParam->pAuth), "Authorization: Basic %s", tsAdapterToken); + + _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY); + pParam->pColIdMap = tSimpleHashInit(32, hashFn); } int32_t createStreamDataScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode, SNode* pTagCond, diff --git a/source/libs/executor/src/streamtimesliceoperator.c b/source/libs/executor/src/streamtimesliceoperator.c index 1b454f63ba..e1e13f8aed 100644 --- a/source/libs/executor/src/streamtimesliceoperator.c +++ b/source/libs/executor/src/streamtimesliceoperator.c @@ -456,7 +456,10 @@ static int32_t fillPointResult(SStreamFillSupporter* pFillSup, SResultRowData* p qError("%s failed at line %d since fill errror", __func__, __LINE__); } } else { - int32_t srcSlot = pFillCol->pExpr->base.pParam[0].pCol->slotId; + int32_t srcSlot = pFillCol->pExpr->base.pParam[0].pCol->slotId; + if (pFillSup->normalFill) { + srcSlot = dstSlotId; + } SResultCellData* pCell = NULL; if (IS_FILL_CONST_VALUE(pFillSup->type) && (isGroupKeyFunc(pFillCol->pExpr) || isSelectGroupConstValueFunc(pFillCol->pExpr))) { @@ -532,7 +535,10 @@ static void fillLinearRange(SStreamFillSupporter* pFillSup, SStreamFillInfo* pFi qError("%s failed at line %d since fill errror", __func__, lino); } } else if (isInterpFunc(pFillCol->pExpr) || pFillSup->normalFill) { - int32_t srcSlot = pFillCol->pExpr->base.pParam[0].pCol->slotId; + int32_t srcSlot = pFillCol->pExpr->base.pParam[0].pCol->slotId; + if (pFillSup->normalFill) { + srcSlot = dstSlotId; + } SResultCellData* pCell = getSliceResultCell(pFillInfo->pResRow->pRowVal, srcSlot, pFillSup->pOffsetInfo); if (IS_VAR_DATA_TYPE(type) || type == TSDB_DATA_TYPE_BOOL || pCell->isNull) { colDataSetNULL(pDstCol, index); diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c index 75e01d1356..955299dd44 100644 --- a/source/libs/parser/src/parTranslater.c +++ b/source/libs/parser/src/parTranslater.c @@ -12669,6 +12669,13 @@ static int32_t checkStreamQuery(STranslateContext* pCxt, SCreateStreamStmt* pStm } } + if (pStmt->pOptions->triggerType == STREAM_TRIGGER_CONTINUOUS_WINDOW_CLOSE) { + if (pSelect->pWindow != NULL && QUERY_NODE_INTERVAL_WINDOW != nodeType(pSelect->pWindow)) { + return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_STREAM_QUERY, + "When trigger was force window close, Stream only support interval window"); + } + } + if (NULL != pSelect->pGroupByList) { return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_STREAM_QUERY, "Unsupported Group by"); } @@ -13505,7 +13512,7 @@ static int32_t buildCreateStreamReq(STranslateContext* pCxt, SCreateStreamStmt* if (TSDB_CODE_SUCCESS == code) { code = columnDefNodeToField(pStmt->pCols, &pReq->pCols, false, false); } - pReq->recalculateInterval = 0; + pReq->recalculateInterval = 3600000; if (NULL != pStmt->pOptions->pRecInterval) { SValueNode* pValueNode = ((SValueNode*)pStmt->pOptions->pRecInterval); pReq->recalculateInterval = diff --git a/source/libs/stream/src/streamSessionState.c b/source/libs/stream/src/streamSessionState.c index e4db1101c1..70b1715769 100644 --- a/source/libs/stream/src/streamSessionState.c +++ b/source/libs/stream/src/streamSessionState.c @@ -1283,9 +1283,7 @@ int32_t mergeAndSaveScanRange(STableTsDataState* pTsDataState, STimeWindow* pWin rangeKey.pUIds = tSimpleHashInit(8, hashFn); code = putRangeIdInfo(&rangeKey, gpId, uId); QUERY_CHECK_CODE(code, lino, _end); - if (index < 0) { - index = 0; - } + index++; taosArrayInsert(pRangeArray, index, &rangeKey); _end: @@ -1379,7 +1377,8 @@ int32_t popScanRange(STableTsDataState* pTsDataState, SScanRange* pRange) { SStreamStateCur* pCur = NULL; SArray* pRangeArray = pTsDataState->pScanRanges; if (taosArrayGetSize(pRangeArray) > 0) { - (*pRange) = *(SScanRange*) taosArrayPop(pRangeArray); + (*pRange) = *(SScanRange*) taosArrayGet(pRangeArray, 0); + taosArrayRemove(pRangeArray, 0); goto _end; } diff --git a/source/libs/stream/src/tstreamFileState.c b/source/libs/stream/src/tstreamFileState.c index 1305eb6bdd..8007d70a79 100644 --- a/source/libs/stream/src/tstreamFileState.c +++ b/source/libs/stream/src/tstreamFileState.c @@ -890,7 +890,7 @@ bool hasRowBuff(SStreamFileState* pFileState, const SWinKey* pKey, bool hasLimit if (pIsLast != NULL) { (*pIsLast) = false; } - + SRowBuffPos** pos = tSimpleHashGet(pFileState->rowStateBuff, pKey, sizeof(SWinKey)); if (pos) { res = true; @@ -901,17 +901,19 @@ bool hasRowBuff(SStreamFileState* pFileState, const SWinKey* pKey, bool hasLimit if (ppBuff != NULL) { SArray* pWinStates = (SArray*)(*ppBuff); if (pIsLast != NULL) { - SWinKey* pLastKey = (SWinKey*) taosArrayGetLast(pWinStates); + SWinKey* pLastKey = (SWinKey*)taosArrayGetLast(pWinStates); *pIsLast = (winKeyCmprImpl(pKey, pLastKey) == 0); } if (hasLimit && taosArrayGetSize(pWinStates) <= MIN_NUM_OF_SORT_CACHE_WIN) { res = true; } if (qDebugFlag & DEBUG_DEBUG) { - SWinKey* fistKey = (SWinKey*)taosArrayGet(pWinStates, 0); - qDebug("===stream===check window state. buff min ts:%" PRId64 ",groupId:%" PRIu64 ".key ts:%" PRId64 - ",groupId:%" PRIu64, - fistKey->ts, fistKey->groupId, pKey->ts, pKey->groupId); + if (taosArrayGetSize(pWinStates) > 0) { + SWinKey* fistKey = (SWinKey*)taosArrayGet(pWinStates, 0); + qDebug("===stream===check window state. buff min ts:%" PRId64 ",groupId:%" PRIu64 ".key ts:%" PRId64 + ",groupId:%" PRIu64, + fistKey->ts, fistKey->groupId, pKey->ts, pKey->groupId); + } } } else { res = true; diff --git a/tests/parallel_test/longtimeruning_cases.task b/tests/parallel_test/longtimeruning_cases.task index 64c2b1b45b..3afb835ee7 100644 --- a/tests/parallel_test/longtimeruning_cases.task +++ b/tests/parallel_test/longtimeruning_cases.task @@ -7,6 +7,8 @@ # #,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/stream_multi_agg.py #,,n,system-test,python3 ./test.py -f 8-stream/stream_basic.py +,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/continuous_window_close_interval_basic.py +,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/continuous_window_close_interval.py # army-test #,,y,army,./pytest.sh python3 ./test.py -f multi-level/mlevel_basic.py -N 3 -L 3 -D 2 diff --git a/tests/pytest/util/common.py b/tests/pytest/util/common.py index c12f324fd7..c2861155de 100644 --- a/tests/pytest/util/common.py +++ b/tests/pytest/util/common.py @@ -923,12 +923,12 @@ class TDCom: else: if watermark is None: - if trigger_mode == "max_delay": + if trigger_mode == "max_delay" or trigger_mode == "continuous_window_close" : stream_options = f'trigger {trigger_mode} {max_delay}' else: stream_options = f'trigger {trigger_mode}' else: - if trigger_mode == "max_delay": + if trigger_mode == "max_delay" or trigger_mode == "continuous_window_close" : stream_options = f'trigger {trigger_mode} {max_delay} watermark {watermark}' else: stream_options = f'trigger {trigger_mode} watermark {watermark}' diff --git a/tests/script/tsim/stream/nonblockIntervalBasic.sim b/tests/script/tsim/stream/nonblockIntervalBasic.sim index 855982e07a..5af09a1bca 100644 --- a/tests/script/tsim/stream/nonblockIntervalBasic.sim +++ b/tests/script/tsim/stream/nonblockIntervalBasic.sim @@ -13,6 +13,13 @@ sql use test; sql create stable st(ts timestamp,a int,b int,c int) tags(ta int,tb int,tc int); sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); + + +sql_error create stream streams_er1 trigger continuous_window_close ignore update 0 ignore expired 0 into streamt_et1 as select _wstart, count(*) c1, sum(b) c2 from st partition by tbname session(ts, 10s); +sql_error create stream streams_er2 trigger continuous_window_close ignore update 0 ignore expired 0 into streamt_et2 as select _wstart, count(*) c1, sum(b) c2 from st partition by tbname state_window(a) ; +sql_error create stream streams_er3 trigger continuous_window_close ignore update 0 ignore expired 0 into streamt_et3 as select _wstart, count(*) c1, sum(b) c2 from st partition by tbname count_window(10); +sql_error create stream streams_er4 trigger continuous_window_close ignore update 0 ignore expired 0 into streamt_et4 as select _wstart, count(*) c1, sum(b) c2 from st partition by tbname event_window start with a = 0 end with b = 9; + sql create stream streams1 trigger continuous_window_close ignore update 0 ignore expired 0 into streamt1 as select _wstart, count(*) c1, sum(b) c2 from st partition by tbname interval(10s) ; run tsim/stream/checkTaskStatus.sim @@ -449,7 +456,7 @@ if $rows != 6 then goto loop6 endi - +$loop_count = 0 loop7: sleep 500 print sql select * from information_schema.ins_tables where table_name like "streams6-tbn-%"; @@ -472,6 +479,7 @@ if $rows != 2 then goto loop7 endi +$loop_count = 0 loop8: sleep 500 print sql select * from information_schema.ins_tables where table_name like "streams7-tbn-%"; @@ -494,7 +502,7 @@ if $rows != 2 then goto loop8 endi - +$loop_count = 0 loop9: sleep 500 print sql select * from streamt6; @@ -517,6 +525,7 @@ if $rows != 2 then goto loop9 endi +$loop_count = 0 loop10: sleep 500 print sql select * from streamt7; @@ -560,6 +569,7 @@ sql insert into t2 values(1648791211000,2,2,3); sql insert into t1 values(1648791221000,1,3,3); +$loop_count = 0 loop11: sleep 500 print sql select * from streamt9; @@ -589,6 +599,7 @@ endi sql insert into t2 values(1648791211001,2,4,3); +$loop_count = 0 loop12: sleep 500 print sql select * from streamt8; @@ -611,6 +622,7 @@ if $rows != 1 then goto loop12 endi +$loop_count = 0 loop13: sleep 500 print sql select * from streamt9; diff --git a/tests/system-test/8-stream/continuous_window_close_interval.py b/tests/system-test/8-stream/continuous_window_close_interval.py new file mode 100644 index 0000000000..908b7049a8 --- /dev/null +++ b/tests/system-test/8-stream/continuous_window_close_interval.py @@ -0,0 +1,217 @@ +import sys +import threading +from util.log import * +from util.sql import * +from util.cases import * +from util.common import * + + +class TDTestCase: + updatecfgDict = {"debugFlag": 135, "asynclog": 0, "ratioOfVnodeStreamThreads": 4} + + def init(self, conn, logSql, replicaVar=1): + self.replicaVar = int(replicaVar) + tdLog.debug("start to execute %s" % __file__) + tdSql.init(conn.cursor(), logSql) + self.tdCom = tdCom + + def check_stream_all_task_status(self, stream_timeout=0): + """check stream status + + Args: + stream_name (str): stream_name + Returns: + str: status + """ + timeout = self.stream_timeout if stream_timeout is None else stream_timeout + + #check stream task rows + sql_task_status = f"select * from information_schema.ins_stream_tasks where status != \"ready\";" + sql_task_all = f"select * from information_schema.ins_stream_tasks;" + + #check stream task status + checktimes = 0 + while checktimes <= timeout: + tdLog.notice(f"checktimes:{checktimes}") + try: + tdSql.query(sql_task_status,row_tag=True) + result_task_status_rows = tdSql.getRows() + if result_task_status_rows == 0: + tdSql.query(sql_task_all,row_tag=True) + result_task_status_rows = tdSql.getRows() + if result_task_status_rows > 0: + break + time.sleep(1) + checktimes += 1 + except Exception as e: + tdLog.notice(f"Try to check stream status again, check times: {checktimes}") + checktimes += 1 + tdSql.print_error_frame_info(f"status is not ready") + else: + tdLog.notice(f"it has spend {checktimes} for checking stream task status but it failed") + if checktimes == timeout: + tdSql.print_error_frame_info(f"status is ready,") + + def docontinuous( + self, + interval, + watermark=None, + partition=None, + fill_value=None, + fill_history_value=None, + ignore_expired=0, + ignore_update=0, + use_exist_stb=None, + tag_value=None + ): + tdLog.info(f"*** testing stream continuous window close: interval: {interval}, partition: {partition}, fill_history: {fill_history_value}, use_exist_stb: {use_exist_stb}, fill: {fill_value}, tag_value: {tag_value} ***") + self.tdCom.case_name = sys._getframe().f_code.co_name + if watermark is not None: + self.tdCom.case_name = "watermark" + sys._getframe().f_code.co_name + self.tdCom.prepare_data(interval=interval, watermark=watermark, ext_stb=use_exist_stb) + tdLog.info( + f"testing stream continue_window_close finish prepare_data" + ) + + sqlstr = "alter local 'streamCoverage' '1'" + tdSql.query(sqlstr) + recalculatetime = 60 + recalculatetimeStr = f"recalculate {recalculatetime}s" + + self.stb_name = self.tdCom.stb_name.replace(f"{self.tdCom.dbname}.", "") + self.ctb_name = self.tdCom.ctb_name.replace(f"{self.tdCom.dbname}.", "") + self.tb_name = self.tdCom.tb_name.replace(f"{self.tdCom.dbname}.", "") + self.stb_stream_des_table = f'{self.stb_name}{self.tdCom.des_table_suffix}' + self.delete = True + + if partition == "tbname": + partition_elm_alias = self.tdCom.partition_tbname_alias + elif partition == "c1": + partition_elm_alias = self.tdCom.partition_col_alias + elif partition == "abs(c1)": + partition_elm_alias = self.tdCom.partition_expression_alias + else: + partition_elm_alias = self.tdCom.partition_tag_alias + if partition == "tbname": + stb_subtable_value = f'concat(concat("{self.stb_name}_{self.tdCom.subtable_prefix}", {partition_elm_alias}), "{self.tdCom.subtable_suffix}")' if self.tdCom.subtable else None + else: + stb_subtable_value = f'concat(concat("{self.stb_name}_{self.tdCom.subtable_prefix}", cast({partition_elm_alias} as varchar(20))), "{self.tdCom.subtable_suffix}")' if self.tdCom.subtable else None + + if watermark is not None: + watermark_value = f'{self.tdCom.dataDict["watermark"]}s' + else: + watermark_value = None + # create stb/ctb/tb stream + if fill_value: + if "value" in fill_value.lower(): + fill_value='VALUE,1,2,3,4,5,6,7,8,9,10,11,1,2,3,4,5,6,7,8,9,10,11' + tdLog.info( + f"*** testing stream continue_window_close + interval + fill. partition: {partition}, interval: {interval}, fill: {fill_value} ***" + ) + + # no subtable + # create stream super table and child table + tdLog.info("create stream super table and child table") + if use_exist_stb: + self.stb_stream_des_table = self.tdCom.ext_stb_stream_des_table + self.des_select_str = self.tdCom.ext_tb_source_select_str + else: + self.des_select_str = self.tdCom.stb_source_select_str + self.tdCom.create_stream(stream_name=f'{self.stb_name}{self.tdCom.stream_suffix}', des_table=self.stb_stream_des_table, source_sql=f'select _wstart AS wstart, {self.des_select_str} from {self.stb_name} partition by {partition} {partition_elm_alias} interval({self.tdCom.dataDict["interval"]}s)', trigger_mode="continuous_window_close", watermark=watermark_value, ignore_expired=ignore_expired, subtable_value=stb_subtable_value, fill_value=fill_value, use_exist_stb=use_exist_stb, tag_value=tag_value, max_delay=recalculatetimeStr) + + # wait and check stream_task status is ready + tdSql.query("show streams") + tdLog.info(f"tdSql.queryResult:{tdSql.queryResult},tdSql.queryRows:{tdSql.queryRows}") + self.check_stream_all_task_status( + stream_timeout=120 + ) + + # insert data + start_time = self.tdCom.date_time + print(f"range count:{self.tdCom.range_count}") + for i in range(self.tdCom.range_count): + if i == 0: + if watermark is not None: + window_close_ts = self.tdCom.cal_watermark_window_close_interval_endts(self.tdCom.date_time, self.tdCom.dataDict['interval'], self.tdCom.dataDict['watermark']) + else: + window_close_ts = self.tdCom.cal_watermark_window_close_interval_endts(self.tdCom.date_time, self.tdCom.dataDict['interval']) + else: + self.tdCom.date_time = window_close_ts + self.tdCom.offset + window_close_ts += self.tdCom.dataDict['interval']*self.tdCom.offset + + for num in range(int(window_close_ts/self.tdCom.offset-self.tdCom.date_time/self.tdCom.offset)): + ts_value=self.tdCom.date_time+num*self.tdCom.offset + self.tdCom.sinsert_rows(tbname=self.ctb_name, ts_value=ts_value) + if i%2 == 0: + self.tdCom.sinsert_rows(tbname=self.ctb_name, ts_value=ts_value) + if self.delete and i%2 != 0: + ts_cast_delete_value = self.tdCom.time_cast(ts_value) + self.tdCom.sdelete_rows(tbname=self.ctb_name, start_ts=ts_cast_delete_value) + + if not fill_value and partition != "c1": + for tbname in [self.stb_stream_des_table]: + if use_exist_stb and tbname == self.stb_stream_des_table: + tdSql.waitedQuery(f'select {self.tdCom.partitial_stb_filter_des_select_elm} from {self.stb_stream_des_table}', i, 60) + else: + tdSql.waitedQuery(f'select wstart, {self.tdCom.stb_output_select_str} from {tbname}', i, 60) + + self.tdCom.sinsert_rows(tbname=self.ctb_name, ts_value=window_close_ts-1) + + if not fill_value: + for tbname in [self.stb_stream_des_table]: + if use_exist_stb and tbname == self.stb_stream_des_table: + tdSql.waitedQuery(f'select {self.tdCom.partitial_stb_filter_des_select_elm} from {self.stb_stream_des_table}', i, 60) + else: + tdSql.waitedQuery(f'select wstart, {self.tdCom.stb_output_select_str} from {tbname}', i, 60) + + start_ts = start_time + future_ts = str(self.tdCom.date_time)+f'+{self.tdCom.dataDict["interval"]*(self.tdCom.range_count+2)}s' + end_ts = self.tdCom.time_cast(future_ts) + self.tdCom.sinsert_rows(tbname=self.ctb_name, ts_value=future_ts) + future_ts_bigint = self.tdCom.str_ts_trans_bigint(future_ts) + if watermark is not None: + window_close_ts = self.tdCom.cal_watermark_window_close_interval_endts(future_ts_bigint, self.tdCom.dataDict['interval'], self.tdCom.dataDict['watermark']) + else: + window_close_ts = self.tdCom.cal_watermark_window_close_interval_endts(future_ts_bigint, self.tdCom.dataDict['interval']) + self.tdCom.sinsert_rows(tbname=self.ctb_name, ts_value=window_close_ts) + + waitTime = recalculatetime * 2 + tdLog.info(f"sleep {waitTime} s") + time.sleep(waitTime) + + if fill_value: + for tbname in [self.stb_name]: + if "value" in fill_value.lower(): + fill_value='VALUE,1,2,3,6,7,8,9,10,11,1,2,3,4,5,6,7,8,9,10,11' + additional_options = f"where ts >= {start_ts} and ts <= {end_ts}" + self.tdCom.check_query_data(f'select wstart, {self.tdCom.fill_stb_output_select_str} from {tbname}{self.tdCom.des_table_suffix} order by wstart', f'select _wstart AS wstart, {self.tdCom.fill_stb_source_select_str} from {tbname} {additional_options} partition by {partition} interval({self.tdCom.dataDict["interval"]}s) fill ({fill_value}) order by wstart', fill_value=fill_value) + else: + for tbname in [self.stb_name]: + additional_options = f"where ts <= {end_ts}" + self.tdCom.check_query_data(f'select wstart, {self.tdCom.stb_output_select_str} from {tbname}{self.tdCom.des_table_suffix} order by wstart', f'select _wstart AS wstart, {self.tdCom.stb_source_select_str} from {tbname} {additional_options} partition by {partition} interval({self.tdCom.dataDict["interval"]}s) order by wstart', fill_value=fill_value) + + def run(self): + for fill_value in [None, "VALUE", "NULL", "PREV", "NEXT", "LINEAR"]: + self.docontinuous( + interval=random.randint(10, 15), + partition="tbname", + fill_value=fill_value + ) + for fill_value in ["VALUE", "NULL", "PREV", "NEXT", "LINEAR", None]: + self.docontinuous( + interval=random.randint(10, 12), + partition="t1 as t5,t2 as t11,t3 as t13, t4", + fill_value=fill_value + ) + + def stop(self): + tdLog.info("stop========================================") + tdSql.close() + tdLog.success(f"{__file__} successfully executed") + + +event = threading.Event() + + +tdCases.addLinux(__file__, TDTestCase()) +tdCases.addWindows(__file__, TDTestCase()) diff --git a/tests/system-test/8-stream/continuous_window_close_interval_basic.py b/tests/system-test/8-stream/continuous_window_close_interval_basic.py new file mode 100644 index 0000000000..4e080371cb --- /dev/null +++ b/tests/system-test/8-stream/continuous_window_close_interval_basic.py @@ -0,0 +1,242 @@ +import sys +import threading +from util.log import * +from util.sql import * +from util.cases import * +from util.common import * + + +class TDTestCase: + updatecfgDict = {"debugFlag": 135, "asynclog": 0, "ratioOfVnodeStreamThreads": 4} + + def init(self, conn, logSql, replicaVar=1): + self.replicaVar = int(replicaVar) + tdLog.debug("start to execute %s" % __file__) + tdSql.init(conn.cursor(), logSql) + self.tdCom = tdCom + + def check_stream_all_task_status(self, stream_timeout=0): + """check stream status + + Args: + stream_name (str): stream_name + Returns: + str: status + """ + timeout = self.stream_timeout if stream_timeout is None else stream_timeout + + #check stream task rows + sql_task_status = f"select * from information_schema.ins_stream_tasks where status != \"ready\";" + sql_task_all = f"select * from information_schema.ins_stream_tasks;" + + #check stream task status + checktimes = 0 + while checktimes <= timeout: + tdLog.notice(f"checktimes:{checktimes}") + try: + tdSql.query(sql_task_status,row_tag=True) + result_task_status_rows = tdSql.getRows() + if result_task_status_rows == 0: + tdSql.query(sql_task_all,row_tag=True) + result_task_status_rows = tdSql.getRows() + if result_task_status_rows > 0: + break + time.sleep(1) + checktimes += 1 + except Exception as e: + tdLog.notice(f"Try to check stream status again, check times: {checktimes}") + checktimes += 1 + tdSql.print_error_frame_info(f"status is not ready") + else: + tdLog.notice(f"it has spend {checktimes} for checking stream task status but it failed") + if checktimes == timeout: + tdSql.print_error_frame_info(f"status is ready,") + + def docontinuous( + self, + interval, + watermark=None, + partition=None, + fill_value=None, + fill_history_value=None, + ignore_expired=0, + ignore_update=0, + use_exist_stb=None, + tag_value=None, + ): + tdLog.info(f"*** testing stream continuous window close: interval: {interval}, partition: {partition}, fill_history: {fill_history_value}, use_exist_stb: {use_exist_stb}, fill: {fill_value}, tag_value: {tag_value} ***") + self.tdCom.case_name = sys._getframe().f_code.co_name + if watermark is not None: + self.tdCom.case_name = "watermark" + sys._getframe().f_code.co_name + self.tdCom.prepare_data(interval=interval, watermark=watermark, ext_stb=use_exist_stb) + tdLog.info( + f"testing stream continue_window_close finish prepare_data" + ) + + self.stb_name = self.tdCom.stb_name.replace(f"{self.tdCom.dbname}.", "") + self.ctb_name = self.tdCom.ctb_name.replace(f"{self.tdCom.dbname}.", "") + self.tb_name = self.tdCom.tb_name.replace(f"{self.tdCom.dbname}.", "") + self.stb_stream_des_table = f'{self.stb_name}{self.tdCom.des_table_suffix}' + self.ctb_stream_des_table = f'{self.ctb_name}{self.tdCom.des_table_suffix}' + defined_tag_count = len(tag_value.split()) if tag_value is not None else 0 + + if partition == "tbname": + partition_elm_alias = self.tdCom.partition_tbname_alias + elif partition == "c1": + partition_elm_alias = self.tdCom.partition_col_alias + elif partition == "abs(c1)": + partition_elm_alias = self.tdCom.partition_expression_alias + else: + partition_elm_alias = self.tdCom.partition_tag_alias + if partition == "tbname": + stb_subtable_value = f'concat(concat("{self.stb_name}_{self.tdCom.subtable_prefix}", {partition_elm_alias}), "{self.tdCom.subtable_suffix}")' if self.tdCom.subtable else None + ctb_subtable_value = f'concat(concat("{self.ctb_name}_{self.tdCom.subtable_prefix}", {partition_elm_alias}), "{self.tdCom.subtable_suffix}")' if self.tdCom.subtable else None + else: + stb_subtable_value = f'concat(concat("{self.stb_name}_{self.tdCom.subtable_prefix}", cast({partition_elm_alias} as varchar(20))), "{self.tdCom.subtable_suffix}")' if self.tdCom.subtable else None + ctb_subtable_value = f'concat(concat("{self.ctb_name}_{self.tdCom.subtable_prefix}", cast({partition_elm_alias} as varchar(20))), "{self.tdCom.subtable_suffix}")' if self.tdCom.subtable else None + + if watermark is not None: + watermark_value = f'{self.tdCom.dataDict["watermark"]}s' + else: + watermark_value = None + # create stb/ctb/tb stream + if fill_value: + if "value" in fill_value.lower(): + fill_value='VALUE,1,2,3,4,5,6,7,8,9,10,11,1,2,3,4,5,6,7,8,9,10,11' + tdLog.info( + f"*** testing stream continue_window_close + interval + fill. partition: {partition}, interval: {interval}, fill: {fill_value} ***" + ) + + # no subtable + # create stream super table and child table + tdLog.info("create stream super table and child table") + if use_exist_stb: + self.stb_stream_des_table = self.tdCom.ext_stb_stream_des_table + self.des_select_str = self.tdCom.ext_tb_source_select_str + else: + self.des_select_str = self.tdCom.stb_source_select_str + recalculatetime = 60000 + recalculatetimeStr = f"recalculate {recalculatetime}s" + self.tdCom.create_stream(stream_name=f'{self.stb_name}{self.tdCom.stream_suffix}', des_table=self.stb_stream_des_table, source_sql=f'select _wstart AS wstart, {self.des_select_str} from {self.stb_name} partition by {partition} {partition_elm_alias} interval({self.tdCom.dataDict["interval"]}s)', trigger_mode="continuous_window_close", watermark=watermark_value, ignore_expired=ignore_expired, subtable_value=stb_subtable_value, fill_value=fill_value, use_exist_stb=use_exist_stb, tag_value=tag_value, max_delay=recalculatetimeStr) + self.tdCom.create_stream(stream_name=f'{self.ctb_name}{self.tdCom.stream_suffix}', des_table=self.ctb_stream_des_table, source_sql=f'select _wstart AS wstart, {self.tdCom.stb_source_select_str} from {self.ctb_name} partition by {partition} {partition_elm_alias} interval({self.tdCom.dataDict["interval"]}s)', trigger_mode="continuous_window_close", watermark=watermark_value, ignore_expired=ignore_expired, subtable_value=ctb_subtable_value, fill_value=fill_value, use_exist_stb=use_exist_stb, max_delay=recalculatetimeStr) + + # wait and check stream_task status is ready + tdSql.query("show streams") + tdLog.info(f"tdSql.queryResult:{tdSql.queryResult},tdSql.queryRows:{tdSql.queryRows}") + localQueryResult = tdSql.queryResult + for stream_number in range(tdSql.queryRows): + stream_name = localQueryResult[stream_number][0] + self.check_stream_all_task_status( + stream_timeout=120 + ) + + # insert data + start_time = self.tdCom.date_time + print(f"range count:{self.tdCom.range_count}") + for i in range(self.tdCom.range_count): + if i == 0: + if watermark is not None: + window_close_ts = self.tdCom.cal_watermark_window_close_interval_endts(self.tdCom.date_time, self.tdCom.dataDict['interval'], self.tdCom.dataDict['watermark']) + else: + window_close_ts = self.tdCom.cal_watermark_window_close_interval_endts(self.tdCom.date_time, self.tdCom.dataDict['interval']) + else: + self.tdCom.date_time = window_close_ts + self.tdCom.offset + window_close_ts += self.tdCom.dataDict['interval']*self.tdCom.offset + + for num in range(int(window_close_ts/self.tdCom.offset-self.tdCom.date_time/self.tdCom.offset)): + ts_value=self.tdCom.date_time+num*self.tdCom.offset + self.tdCom.sinsert_rows(tbname=self.ctb_name, ts_value=ts_value) + + if not fill_value and partition != "c1": + for tbname in [self.stb_stream_des_table, self.ctb_stream_des_table]: + if use_exist_stb and tbname == self.stb_stream_des_table: + tdSql.waitedQuery(f'select {self.tdCom.partitial_stb_filter_des_select_elm} from {self.stb_stream_des_table}', i, 60) + else: + tdSql.waitedQuery(f'select wstart, {self.tdCom.stb_output_select_str} from {tbname}', i, 60) + + self.tdCom.sinsert_rows(tbname=self.ctb_name, ts_value=window_close_ts-1) + + if not fill_value: + for tbname in [self.stb_stream_des_table, self.ctb_stream_des_table]: + if use_exist_stb and tbname == self.stb_stream_des_table: + tdSql.waitedQuery(f'select {self.tdCom.partitial_stb_filter_des_select_elm} from {self.stb_stream_des_table}', i, 60) + else: + tdSql.waitedQuery(f'select wstart, {self.tdCom.stb_output_select_str} from {tbname}', i, 60) + + if self.tdCom.subtable: + for tname in [self.stb_name, self.ctb_name]: + ptn_counter = 0 + if partition == "c1": + tdSql.query(f'select * from {self.tb_name}') + for c1_value in tdSql.queryResult: + tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`', count_expected_res=self.tdCom.range_count) + elif partition == "tbname" and ptn_counter == 0: + tdLog.info(f'====={tname}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}=line{sys._getframe().f_lineno}') + tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`', count_expected_res=self.tdCom.range_count) + ptn_counter += 1 + else: + tdLog.info(f'====={tname}_{self.tdCom.subtable_prefix}=line{sys._getframe().f_lineno}') + tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}') + tdSql.query(f'select count(*) from `{tbname}`', count_expected_res=self.tdCom.range_count) + ptn_counter += 1 + + if fill_value: + start_ts = start_time + future_ts = str(self.tdCom.date_time)+f'+{self.tdCom.dataDict["interval"]*(self.tdCom.range_count+2)}s' + end_ts = self.tdCom.time_cast(future_ts) + self.tdCom.sinsert_rows(tbname=self.ctb_name, ts_value=future_ts) + future_ts_bigint = self.tdCom.str_ts_trans_bigint(future_ts) + if watermark is not None: + window_close_ts = self.tdCom.cal_watermark_window_close_interval_endts(future_ts_bigint, self.tdCom.dataDict['interval'], self.tdCom.dataDict['watermark']) + else: + window_close_ts = self.tdCom.cal_watermark_window_close_interval_endts(future_ts_bigint, self.tdCom.dataDict['interval']) + self.tdCom.sinsert_rows(tbname=self.ctb_name, ts_value=window_close_ts) + + self.tdCom.date_time = start_time + for tbname in [self.stb_name, self.ctb_name]: + if "value" in fill_value.lower(): + fill_value='VALUE,1,2,3,6,7,8,9,10,11,1,2,3,4,5,6,7,8,9,10,11' + additional_options = f"where ts >= {start_ts} and ts <= {end_ts}" + self.tdCom.check_query_data(f'select wstart, {self.tdCom.fill_stb_output_select_str} from {tbname}{self.tdCom.des_table_suffix} order by wstart', f'select _wstart AS wstart, {self.tdCom.fill_stb_source_select_str} from {tbname} {additional_options} partition by {partition} interval({self.tdCom.dataDict["interval"]}s) fill ({fill_value}) order by wstart', fill_value=fill_value) + + def run(self): + for fill_value in ["VALUE", "NULL", "PREV", "NEXT", "LINEAR", None]: + self.docontinuous( + interval=random.randint(10, 15), + partition="tbname", + fill_value=fill_value, + ) + for fill_value in ["NULL", "PREV", "NEXT", "LINEAR", None]: + self.docontinuous( + interval=random.randint(10, 12), + partition="t1 as t5,t2 as t11,t3 as t13, t4", + fill_value=fill_value, + ) + + self.docontinuous( + interval=random.randint(10, 12), + partition="c1", + fill_value=None + ) + + self.docontinuous( + interval=random.randint(10, 12), + partition="t1 as t5,t2 as t11,t3 as t13, t4", + fill_value=None, + use_exist_stb=True, + tag_value="t5,t11,t13" + ) + + def stop(self): + tdLog.info("stop========================================") + tdSql.close() + tdLog.success(f"{__file__} successfully executed") + + +event = threading.Event() + + +tdCases.addLinux(__file__, TDTestCase()) +tdCases.addWindows(__file__, TDTestCase())