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

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
This commit is contained in:
liuyao 2025-03-20 10:34:31 +08:00 committed by GitHub
parent 0a96294056
commit fac36002c1
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
19 changed files with 684 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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 {

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

@ -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
}

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

@ -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
}

4
source/libs/executor/src/projectoperator.c
View File

@ -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,

67
source/libs/executor/src/streamclient.c
View File

@ -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

53
source/libs/executor/src/streamfillnonblockoperator.c
View File

@ -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);
}
}
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;
}

15
source/libs/executor/src/streamfilloperator.c
View File

@ -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++;

23
source/libs/executor/src/streamintervalnonblockoperator.c
View File

@ -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);
}

6
source/libs/executor/src/streamintervalsliceoperator.c
View File

@ -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,

37
source/libs/executor/src/streamscanoperator.c
View File

@ -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,

10
source/libs/executor/src/streamtimesliceoperator.c
View File

@ -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);

9
source/libs/parser/src/parTranslater.c
View File

@ -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 =

7
source/libs/stream/src/streamSessionState.c
View File

@ -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;
}

14
source/libs/stream/src/tstreamFileState.c
View File

@ -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;

2
tests/parallel_test/longtimeruning_cases.task
View File

@ -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

4
tests/pytest/util/common.py
View File

@ -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}'

16
tests/script/tsim/stream/nonblockIntervalBasic.sim
View File

@ -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;

217
tests/system-test/8-stream/continuous_window_close_interval.py Normal file
View File

@ -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())

242
tests/system-test/8-stream/continuous_window_close_interval_basic.py Normal file
View File

@ -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())