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

Merge branch '3.0' of https://github.com/taosdata/TDengine into feat/stream_compression

This commit is contained in:
Hongze Cheng 2022-09-28 09:42:00 +08:00
parent 18f3bd53c5 5aad79527e
commit 95df98315d
43 changed files with 4732 additions and 205 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cmake/cmake.version
View File

@ -2,7 +2,7 @@
IF (DEFINED VERNUMBER) IF (DEFINED VERNUMBER)
SET(TD_VER_NUMBER ${VERNUMBER}) SET(TD_VER_NUMBER ${VERNUMBER})
ELSE () ELSE ()
SET(TD_VER_NUMBER "3.0.1.2") SET(TD_VER_NUMBER "3.0.1.3")
ENDIF () ENDIF ()
IF (DEFINED VERCOMPATIBLE) IF (DEFINED VERCOMPATIBLE)

2
cmake/taostools_CMakeLists.txt.in
View File

@ -2,7 +2,7 @@
# taos-tools # taos-tools
ExternalProject_Add(taos-tools ExternalProject_Add(taos-tools
GIT_REPOSITORY https://github.com/taosdata/taos-tools.git GIT_REPOSITORY https://github.com/taosdata/taos-tools.git
GIT_TAG f03c09a GIT_TAG 70f5a1c
SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools" SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools"
BINARY_DIR "" BINARY_DIR ""
#BUILD_IN_SOURCE TRUE #BUILD_IN_SOURCE TRUE

4
docs/en/28-releases/01-tdengine.md
View File

@ -6,6 +6,10 @@ description: TDengine release history, Release Notes and download links.
import Release from "/components/ReleaseV3"; import Release from "/components/ReleaseV3";
## 3.0.1.3
<Release type="tdengine" version="3.0.1.3" />
## 3.0.1.2 ## 3.0.1.2
<Release type="tdengine" version="3.0.1.2" /> <Release type="tdengine" version="3.0.1.2" />

4
docs/en/28-releases/02-tools.md
View File

@ -6,6 +6,10 @@ description: taosTools release history, Release Notes, download links.
import Release from "/components/ReleaseV3"; import Release from "/components/ReleaseV3";
## 2.2.3
<Release type="tools" version="2.2.3" />
## 2.2.2 ## 2.2.2
<Release type="tools" version="2.2.2" /> <Release type="tools" version="2.2.2" />

4
docs/zh/28-releases/01-tdengine.md
View File

@ -6,6 +6,10 @@ description: TDengine 发布历史、Release Notes 及下载链接
import Release from "/components/ReleaseV3"; import Release from "/components/ReleaseV3";
## 3.0.1.3
<Release type="tdengine" version="3.0.1.3" />
## 3.0.1.2 ## 3.0.1.2
<Release type="tdengine" version="3.0.1.2" /> <Release type="tdengine" version="3.0.1.2" />

4
docs/zh/28-releases/02-tools.md
View File

@ -6,6 +6,10 @@ description: taosTools 的发布历史、Release Notes 和下载链接
import Release from "/components/ReleaseV3"; import Release from "/components/ReleaseV3";
## 2.2.3
<Release type="tools" version="2.2.3" />
## 2.2.2 ## 2.2.2
<Release type="tools" version="2.2.2" /> <Release type="tools" version="2.2.2" />

9
include/common/tmsg.h
View File

@ -55,11 +55,10 @@ extern int32_t tMsgDict[];
#define TMSG_SEG_CODE(TYPE) (((TYPE)&0xff00) >> 8) #define TMSG_SEG_CODE(TYPE) (((TYPE)&0xff00) >> 8)
#define TMSG_SEG_SEQ(TYPE) ((TYPE)&0xff) #define TMSG_SEG_SEQ(TYPE) ((TYPE)&0xff)
#define TMSG_INFO(TYPE) \ #define TMSG_INFO(TYPE) \
((TYPE) >= 0 && ((TYPE) < TDMT_DND_MAX_MSG || (TYPE) < TDMT_MND_MAX_MSG || (TYPE) < TDMT_VND_MAX_MSG || \ ((TYPE) < TDMT_DND_MAX_MSG || (TYPE) < TDMT_MND_MAX_MSG || (TYPE) < TDMT_VND_MAX_MSG || (TYPE) < TDMT_SCH_MAX_MSG || \
(TYPE) < TDMT_SCH_MAX_MSG || (TYPE) < TDMT_STREAM_MAX_MSG || (TYPE) < TDMT_MON_MAX_MSG || \ (TYPE) < TDMT_STREAM_MAX_MSG || (TYPE) < TDMT_MON_MAX_MSG || (TYPE) < TDMT_SYNC_MAX_MSG) \
(TYPE) < TDMT_SYNC_MAX_MSG)) \ ? tMsgInfo[tMsgDict[TMSG_SEG_CODE(TYPE)] + TMSG_SEG_SEQ(TYPE)] \
? tMsgInfo[tMsgDict[TMSG_SEG_CODE(TYPE)] + TMSG_SEG_SEQ(TYPE)] \
: 0 : 0
#define TMSG_INDEX(TYPE) (tMsgDict[TMSG_SEG_CODE(TYPE)] + TMSG_SEG_SEQ(TYPE)) #define TMSG_INDEX(TYPE) (tMsgDict[TMSG_SEG_CODE(TYPE)] + TMSG_SEG_SEQ(TYPE))

1
include/libs/nodes/nodes.h
View File

@ -239,6 +239,7 @@ typedef enum ENodeType {
QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL, QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL,
QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL, QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL,
QUERY_NODE_PHYSICAL_PLAN_FILL, QUERY_NODE_PHYSICAL_PLAN_FILL,
QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL,
QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION, QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION,
QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION, QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION,
QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION, QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION,

2
include/libs/nodes/plannodes.h
View File

@ -464,6 +464,8 @@ typedef struct SFillPhysiNode {
EOrder inputTsOrder; EOrder inputTsOrder;
} SFillPhysiNode; } SFillPhysiNode;
typedef SFillPhysiNode SStreamFillPhysiNode;
typedef struct SMultiTableIntervalPhysiNode { typedef struct SMultiTableIntervalPhysiNode {
SIntervalPhysiNode interval; SIntervalPhysiNode interval;
SNodeList* pPartitionKeys; SNodeList* pPartitionKeys;

12
include/libs/stream/streamState.h
View File

@ -31,6 +31,7 @@ typedef struct {
TDB* db; TDB* db;
TTB* pStateDb; TTB* pStateDb;
TTB* pFuncStateDb; TTB* pFuncStateDb;
TTB* pFillStateDb; // todo refactor
TXN txn; TXN txn;
} SStreamState; } SStreamState;
@ -51,15 +52,22 @@ int32_t streamStateFuncDel(SStreamState* pState, const STupleKey* key);
int32_t streamStatePut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen); int32_t streamStatePut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen);
int32_t streamStateGet(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen); int32_t streamStateGet(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen);
int32_t streamStateDel(SStreamState* pState, const SWinKey* key); int32_t streamStateDel(SStreamState* pState, const SWinKey* key);
int32_t streamStateFillPut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen);
int32_t streamStateFillGet(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen);
int32_t streamStateFillDel(SStreamState* pState, const SWinKey* key);
int32_t streamStateAddIfNotExist(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen); int32_t streamStateAddIfNotExist(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen);
int32_t streamStateReleaseBuf(SStreamState* pState, const SWinKey* key, void* pVal); int32_t streamStateReleaseBuf(SStreamState* pState, const SWinKey* key, void* pVal);
void streamFreeVal(void* val); void streamFreeVal(void* val);
SStreamStateCur* streamStateGetCur(SStreamState* pState, const SWinKey* key); SStreamStateCur* streamStateGetCur(SStreamState* pState, const SWinKey* key);
SStreamStateCur* streamStateSeekKeyNext(SStreamState* pState, const SWinKey* key); SStreamStateCur* streamStateGetAndCheckCur(SStreamState* pState, SWinKey* key);
SStreamStateCur* streamStateSeekKeyPrev(SStreamState* pState, const SWinKey* key); SStreamStateCur* streamStateFillSeekKeyNext(SStreamState* pState, const SWinKey* key);
SStreamStateCur* streamStateFillSeekKeyPrev(SStreamState* pState, const SWinKey* key);
void streamStateFreeCur(SStreamStateCur* pCur); void streamStateFreeCur(SStreamStateCur* pCur);
int32_t streamStateGetGroupKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen);
int32_t streamStateGetKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen); int32_t streamStateGetKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen);
int32_t streamStateSeekFirst(SStreamState* pState, SStreamStateCur* pCur); int32_t streamStateSeekFirst(SStreamState* pState, SStreamStateCur* pCur);

19
source/common/src/tglobal.c
View File

@ -206,7 +206,9 @@ static int32_t taosLoadCfg(SConfig *pCfg, const char **envCmd, const char *input
tstrncpy(cfgFile, cfgDir, sizeof(cfgDir)); tstrncpy(cfgFile, cfgDir, sizeof(cfgDir));
} }
if (apolloUrl == NULL || apolloUrl[0] == '\0') cfgGetApollUrl(envCmd, envFile, apolloUrl); if (apolloUrl != NULL && apolloUrl[0] == '\0') {
cfgGetApollUrl(envCmd, envFile, apolloUrl);
}
if (cfgLoad(pCfg, CFG_STYPE_APOLLO_URL, apolloUrl) != 0) { if (cfgLoad(pCfg, CFG_STYPE_APOLLO_URL, apolloUrl) != 0) {
uError("failed to load from apollo url:%s since %s", apolloUrl, terrstr()); uError("failed to load from apollo url:%s since %s", apolloUrl, terrstr());
@ -1132,11 +1134,20 @@ int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDi
if (tsc) { if (tsc) {
tsLogEmbedded = 0; tsLogEmbedded = 0;
if (taosAddClientLogCfg(pCfg) != 0) return -1; if (taosAddClientLogCfg(pCfg) != 0) {
cfgCleanup(pCfg);
return -1;
}
} else { } else {
tsLogEmbedded = 1; tsLogEmbedded = 1;
if (taosAddClientLogCfg(pCfg) != 0) return -1; if (taosAddClientLogCfg(pCfg) != 0) {
if (taosAddServerLogCfg(pCfg) != 0) return -1; cfgCleanup(pCfg);
return -1;
}
if (taosAddServerLogCfg(pCfg) != 0) {
cfgCleanup(pCfg);
return -1;
}
} }
if (taosLoadCfg(pCfg, envCmd, cfgDir, envFile, apolloUrl) != 0) { if (taosLoadCfg(pCfg, envCmd, cfgDir, envFile, apolloUrl) != 0) {

1
source/dnode/mgmt/mgmt_qnode/src/qmWorker.c
View File

@ -87,6 +87,7 @@ int32_t qmPutRpcMsgToQueue(SQnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) {
return 0; return 0;
default: default:
terrno = TSDB_CODE_INVALID_PARA; terrno = TSDB_CODE_INVALID_PARA;
taosFreeQitem(pMsg);
return -1; return -1;
} }
} }

11
source/dnode/mgmt/mgmt_vnode/src/vmFile.c
View File

@ -141,6 +141,7 @@ _OVER:
} }
int32_t vmWriteVnodeListToFile(SVnodeMgmt *pMgmt) { int32_t vmWriteVnodeListToFile(SVnodeMgmt *pMgmt) {
int32_t ret = 0;
char file[PATH_MAX] = {0}; char file[PATH_MAX] = {0};
char realfile[PATH_MAX] = {0}; char realfile[PATH_MAX] = {0};
snprintf(file, sizeof(file), "%s%svnodes.json.bak", pMgmt->path, TD_DIRSEP); snprintf(file, sizeof(file), "%s%svnodes.json.bak", pMgmt->path, TD_DIRSEP);
@ -161,13 +162,16 @@ int32_t vmWriteVnodeListToFile(SVnodeMgmt *pMgmt) {
char *content = taosMemoryCalloc(1, maxLen + 1); char *content = taosMemoryCalloc(1, maxLen + 1);
if (content == NULL) { if (content == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1; ret = -1;
goto _OVER;
} }
len += snprintf(content + len, maxLen - len, "{\n"); len += snprintf(content + len, maxLen - len, "{\n");
len += snprintf(content + len, maxLen - len, " \"vnodes\": [\n"); len += snprintf(content + len, maxLen - len, " \"vnodes\": [\n");
for (int32_t i = 0; i < numOfVnodes; ++i) { for (int32_t i = 0; i < numOfVnodes; ++i) {
SVnodeObj *pVnode = pVnodes[i]; SVnodeObj *pVnode = pVnodes[i];
if (pVnode == NULL) continue;
len += snprintf(content + len, maxLen - len, " {\n"); len += snprintf(content + len, maxLen - len, " {\n");
len += snprintf(content + len, maxLen - len, " \"vgId\": %d,\n", pVnode->vgId); len += snprintf(content + len, maxLen - len, " \"vgId\": %d,\n", pVnode->vgId);
len += snprintf(content + len, maxLen - len, " \"dropped\": %d,\n", pVnode->dropped); len += snprintf(content + len, maxLen - len, " \"dropped\": %d,\n", pVnode->dropped);
@ -180,12 +184,13 @@ int32_t vmWriteVnodeListToFile(SVnodeMgmt *pMgmt) {
} }
len += snprintf(content + len, maxLen - len, " ]\n"); len += snprintf(content + len, maxLen - len, " ]\n");
len += snprintf(content + len, maxLen - len, "}\n"); len += snprintf(content + len, maxLen - len, "}\n");
terrno = 0;
_OVER:
taosWriteFile(pFile, content, len); taosWriteFile(pFile, content, len);
taosFsyncFile(pFile); taosFsyncFile(pFile);
taosCloseFile(&pFile); taosCloseFile(&pFile);
taosMemoryFree(content); taosMemoryFree(content);
terrno = 0;
for (int32_t i = 0; i < numOfVnodes; ++i) { for (int32_t i = 0; i < numOfVnodes; ++i) {
SVnodeObj *pVnode = pVnodes[i]; SVnodeObj *pVnode = pVnodes[i];
@ -196,6 +201,8 @@ int32_t vmWriteVnodeListToFile(SVnodeMgmt *pMgmt) {
taosMemoryFree(pVnodes); taosMemoryFree(pVnodes);
} }
if (ret != 0) return -1;
dDebug("successed to write %s, numOfVnodes:%d", realfile, numOfVnodes); dDebug("successed to write %s, numOfVnodes:%d", realfile, numOfVnodes);
return taosRenameFile(file, realfile); return taosRenameFile(file, realfile);
} }

20
source/libs/executor/inc/executorimpl.h
View File

@ -34,6 +34,7 @@ extern "C" {
#include "scalar.h" #include "scalar.h"
#include "taosdef.h" #include "taosdef.h"
#include "tarray.h" #include "tarray.h"
#include "tfill.h"
#include "thash.h" #include "thash.h"
#include "tlockfree.h" #include "tlockfree.h"
#include "tmsg.h" #include "tmsg.h"
@ -798,6 +799,22 @@ typedef struct SStreamPartitionOperatorInfo {
SSDataBlock* pDelRes; SSDataBlock* pDelRes;
} SStreamPartitionOperatorInfo; } SStreamPartitionOperatorInfo;
typedef struct SStreamFillOperatorInfo {
SStreamFillSupporter* pFillSup;
SSDataBlock* pRes;
SSDataBlock* pSrcBlock;
int32_t srcRowIndex;
SSDataBlock* pPrevSrcBlock;
SSDataBlock* pSrcDelBlock;
int32_t srcDelRowIndex;
SSDataBlock* pDelRes;
SNode* pCondition;
SArray* pColMatchColInfo;
int32_t primaryTsCol;
int32_t primarySrcSlotId;
SStreamFillInfo* pFillInfo;
} SStreamFillOperatorInfo;
typedef struct STimeSliceOperatorInfo { typedef struct STimeSliceOperatorInfo {
SSDataBlock* pRes; SSDataBlock* pRes;
STimeWindow win; STimeWindow win;
@ -1006,6 +1023,8 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream,
SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
SExecTaskInfo* pTaskInfo); SExecTaskInfo* pTaskInfo);
SOperatorInfo* createStreamFillOperatorInfo(SOperatorInfo* downstream, SStreamFillPhysiNode* pPhyFillNode,
SExecTaskInfo* pTaskInfo);
int32_t projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSDataBlock* pSrcBlock, SqlFunctionCtx* pCtx, int32_t projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSDataBlock* pSrcBlock, SqlFunctionCtx* pCtx,
int32_t numOfOutput, SArray* pPseudoList); int32_t numOfOutput, SArray* pPseudoList);
@ -1094,6 +1113,7 @@ int32_t setOutputBuf(STimeWindow* win, SResultRow** pResult, int64_t tableGroupI
SExecTaskInfo* pTaskInfo); SExecTaskInfo* pTaskInfo);
int32_t releaseOutputBuf(SExecTaskInfo* pTaskInfo, SWinKey* pKey, SResultRow* pResult); int32_t releaseOutputBuf(SExecTaskInfo* pTaskInfo, SWinKey* pKey, SResultRow* pResult);
int32_t saveOutputBuf(SExecTaskInfo* pTaskInfo, SWinKey* pKey, SResultRow* pResult, int32_t resSize); int32_t saveOutputBuf(SExecTaskInfo* pTaskInfo, SWinKey* pKey, SResultRow* pResult, int32_t resSize);
void getNextIntervalWindow(SInterval* pInterval, STimeWindow* tw, int32_t order);
#ifdef __cplusplus #ifdef __cplusplus
} }

113
source/libs/executor/inc/tfill.h
View File

@ -23,12 +23,13 @@ extern "C" {
#include "os.h" #include "os.h"
#include "taosdef.h" #include "taosdef.h"
#include "tcommon.h" #include "tcommon.h"
#include "tsimplehash.h"
struct SSDataBlock; struct SSDataBlock;
typedef struct SFillColInfo { typedef struct SFillColInfo {
SExprInfo *pExpr; SExprInfo* pExpr;
bool notFillCol; // denote if this column needs fill operation bool notFillCol; // denote if this column needs fill operation
SVariant fillVal; SVariant fillVal;
} SFillColInfo; } SFillColInfo;
@ -51,46 +52,96 @@ typedef struct {
} SRowVal; } SRowVal;
typedef struct SFillInfo { typedef struct SFillInfo {
TSKEY start; // start timestamp TSKEY start; // start timestamp
TSKEY end; // endKey for fill TSKEY end; // endKey for fill
TSKEY currentKey; // current active timestamp, the value may be changed during the fill procedure. TSKEY currentKey; // current active timestamp, the value may be changed during the fill procedure.
int32_t tsSlotId; // primary time stamp slot id int32_t tsSlotId; // primary time stamp slot id
int32_t srcTsSlotId; // timestamp column id in the source data block. int32_t srcTsSlotId; // timestamp column id in the source data block.
int32_t order; // order [TSDB_ORDER_ASC|TSDB_ORDER_DESC] int32_t order; // order [TSDB_ORDER_ASC|TSDB_ORDER_DESC]
int32_t type; // fill type int32_t type; // fill type
int32_t numOfRows; // number of rows in the input data block int32_t numOfRows; // number of rows in the input data block
int32_t index; // active row index int32_t index; // active row index
int32_t numOfTotal; // number of filled rows in one round int32_t numOfTotal; // number of filled rows in one round
int32_t numOfCurrent; // number of filled rows in current results int32_t numOfCurrent; // number of filled rows in current results
int32_t numOfCols; // number of columns, including the tags columns int32_t numOfCols; // number of columns, including the tags columns
SInterval interval; SInterval interval;
SRowVal prev; SRowVal prev;
SRowVal next; SRowVal next;
SSDataBlock *pSrcBlock; SSDataBlock* pSrcBlock;
int32_t alloc; // data buffer size in rows int32_t alloc; // data buffer size in rows
SFillColInfo* pFillCol; // column info for fill operations SFillColInfo* pFillCol; // column info for fill operations
SFillTagColInfo* pTags; // tags value for filling gap SFillTagColInfo* pTags; // tags value for filling gap
const char* id; const char* id;
} SFillInfo; } SFillInfo;
typedef struct SResultCellData {
bool isNull;
int8_t type;
int32_t bytes;
char pData[];
} SResultCellData;
typedef struct SResultRowData {
TSKEY key;
SResultCellData* pRowVal;
} SResultRowData;
typedef struct SStreamFillLinearInfo {
TSKEY nextEnd;
SArray* pDeltaVal; // double. value for Fill(linear).
SArray* pNextDeltaVal; // double. value for Fill(linear).
int64_t winIndex;
bool hasNext;
} SStreamFillLinearInfo;
typedef struct SStreamFillInfo {
TSKEY start; // startKey for fill
TSKEY end; // endKey for fill
TSKEY current; // current Key for fill
TSKEY preRowKey;
TSKEY nextRowKey;
SResultRowData* pResRow;
SStreamFillLinearInfo* pLinearInfo;
bool needFill;
int32_t type; // fill type
int32_t pos;
SArray* delRanges;
int32_t delIndex;
} SStreamFillInfo;
typedef struct SStreamFillSupporter {
int32_t type; // fill type
SInterval interval;
SResultRowData prev;
SResultRowData cur;
SResultRowData next;
SResultRowData nextNext;
SFillColInfo* pAllColInfo; // fill exprs and not fill exprs
int32_t numOfAllCols; // number of all exprs, including the tags columns
int32_t numOfFillCols;
int32_t numOfNotFillCols;
int32_t rowSize;
SSHashObj* pResMap;
bool hasDelete;
} SStreamFillSupporter;
int64_t getNumOfResultsAfterFillGap(SFillInfo* pFillInfo, int64_t ekey, int32_t maxNumOfRows); int64_t getNumOfResultsAfterFillGap(SFillInfo* pFillInfo, int64_t ekey, int32_t maxNumOfRows);
void taosFillSetStartInfo(struct SFillInfo* pFillInfo, int32_t numOfRows, TSKEY endKey);
void taosFillSetStartInfo(struct SFillInfo* pFillInfo, int32_t numOfRows, TSKEY endKey); void taosResetFillInfo(struct SFillInfo* pFillInfo, TSKEY startTimestamp);
void taosResetFillInfo(struct SFillInfo* pFillInfo, TSKEY startTimestamp); void taosFillSetInputDataBlock(struct SFillInfo* pFillInfo, const struct SSDataBlock* pInput);
void taosFillSetInputDataBlock(struct SFillInfo* pFillInfo, const struct SSDataBlock* pInput); struct SFillColInfo* createFillColInfo(SExprInfo* pExpr, int32_t numOfFillExpr, SExprInfo* pNotFillExpr,
struct SFillColInfo* createFillColInfo(SExprInfo* pExpr, int32_t numOfFillExpr, SExprInfo* pNotFillExpr, int32_t numOfNotFillCols, const struct SNodeListNode* val); int32_t numOfNotFillCols, const struct SNodeListNode* val);
bool taosFillHasMoreResults(struct SFillInfo* pFillInfo); bool taosFillHasMoreResults(struct SFillInfo* pFillInfo);
SFillInfo* taosCreateFillInfo(TSKEY skey, int32_t numOfFillCols, int32_t numOfNotFillCols, int32_t capacity, SFillInfo* taosCreateFillInfo(TSKEY skey, int32_t numOfFillCols, int32_t numOfNotFillCols, int32_t capacity,
SInterval* pInterval, int32_t fillType, struct SFillColInfo* pCol, int32_t slotId, SInterval* pInterval, int32_t fillType, struct SFillColInfo* pCol, int32_t slotId,
int32_t order, const char* id); int32_t order, const char* id);
void* taosDestroyFillInfo(struct SFillInfo *pFillInfo); void* taosDestroyFillInfo(struct SFillInfo* pFillInfo);
int64_t taosFillResultDataBlock(struct SFillInfo* pFillInfo, SSDataBlock* p, int32_t capacity); int64_t taosFillResultDataBlock(struct SFillInfo* pFillInfo, SSDataBlock* p, int32_t capacity);
int64_t getFillInfoStart(struct SFillInfo *pFillInfo); int64_t getFillInfoStart(struct SFillInfo* pFillInfo);
#ifdef __cplusplus #ifdef __cplusplus
} }

2
source/libs/executor/src/executorimpl.c
View File

@ -3742,6 +3742,8 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
pOptr = createMergeJoinOperatorInfo(ops, size, (SSortMergeJoinPhysiNode*)pPhyNode, pTaskInfo); pOptr = createMergeJoinOperatorInfo(ops, size, (SSortMergeJoinPhysiNode*)pPhyNode, pTaskInfo);
} else if (QUERY_NODE_PHYSICAL_PLAN_FILL == type) { } else if (QUERY_NODE_PHYSICAL_PLAN_FILL == type) {
pOptr = createFillOperatorInfo(ops[0], (SFillPhysiNode*)pPhyNode, pTaskInfo); pOptr = createFillOperatorInfo(ops[0], (SFillPhysiNode*)pPhyNode, pTaskInfo);
} else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL == type) {
pOptr = createStreamFillOperatorInfo(ops[0], (SStreamFillPhysiNode*)pPhyNode, pTaskInfo);
} else if (QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC == type) { } else if (QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC == type) {
pOptr = createIndefinitOutputOperatorInfo(ops[0], pPhyNode, pTaskInfo); pOptr = createIndefinitOutputOperatorInfo(ops[0], pPhyNode, pTaskInfo);
} else if (QUERY_NODE_PHYSICAL_PLAN_INTERP_FUNC == type) { } else if (QUERY_NODE_PHYSICAL_PLAN_INTERP_FUNC == type) {

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

@ -1022,11 +1022,7 @@ static uint64_t getGroupIdByCol(SStreamScanInfo* pInfo, uint64_t uid, TSKEY ts,
return calGroupIdByData(&pInfo->partitionSup, pInfo->pPartScalarSup, pPreRes, 0); return calGroupIdByData(&pInfo->partitionSup, pInfo->pPartScalarSup, pPreRes, 0);
} }
static uint64_t getGroupIdByData(SStreamScanInfo* pInfo, uint64_t uid, TSKEY ts, int64_t maxVersion) { static uint64_t getGroupIdByUid(SStreamScanInfo* pInfo, uint64_t uid) {
if (pInfo->partitionSup.needCalc) {
return getGroupIdByCol(pInfo, uid, ts, maxVersion);
}
SHashObj* map = pInfo->pTableScanOp->pTaskInfo->tableqinfoList.map; SHashObj* map = pInfo->pTableScanOp->pTaskInfo->tableqinfoList.map;
uint64_t* groupId = taosHashGet(map, &uid, sizeof(int64_t)); uint64_t* groupId = taosHashGet(map, &uid, sizeof(int64_t));
if (groupId) { if (groupId) {
@ -1035,6 +1031,14 @@ static uint64_t getGroupIdByData(SStreamScanInfo* pInfo, uint64_t uid, TSKEY ts,
return 0; return 0;
} }
static uint64_t getGroupIdByData(SStreamScanInfo* pInfo, uint64_t uid, TSKEY ts, int64_t maxVersion) {
if (pInfo->partitionSup.needCalc) {
return getGroupIdByCol(pInfo, uid, ts, maxVersion);
}
return getGroupIdByUid(pInfo, uid);
}
static bool prepareRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pBlock, int32_t* pRowIndex) { static bool prepareRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pBlock, int32_t* pRowIndex) {
if ((*pRowIndex) == pBlock->info.rows) { if ((*pRowIndex) == pBlock->info.rows) {
return false; return false;
@ -1081,26 +1085,32 @@ static bool prepareRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pBlock, int32_
return true; return true;
} }
static STimeWindow getSlidingWindow(TSKEY* startTsCol, TSKEY* endTsCol, SInterval* pInterval, static STimeWindow getSlidingWindow(TSKEY* startTsCol, TSKEY* endTsCol, uint64_t* gpIdCol, SInterval* pInterval,
SDataBlockInfo* pDataBlockInfo, int32_t* pRowIndex, bool hasGroup) { SDataBlockInfo* pDataBlockInfo, int32_t* pRowIndex, bool hasGroup) {
SResultRowInfo dumyInfo; SResultRowInfo dumyInfo;
dumyInfo.cur.pageId = -1; dumyInfo.cur.pageId = -1;
STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, startTsCol[*pRowIndex], pInterval, TSDB_ORDER_ASC); STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, startTsCol[*pRowIndex], pInterval, TSDB_ORDER_ASC);
STimeWindow endWin = win; STimeWindow endWin = win;
STimeWindow preWin = win; STimeWindow preWin = win;
uint64_t groupId = gpIdCol[*pRowIndex];
while (1) { while (1) {
if (hasGroup) { if (hasGroup) {
(*pRowIndex) += 1; (*pRowIndex) += 1;
} else { } else {
(*pRowIndex) += getNumOfRowsInTimeWindow(pDataBlockInfo, startTsCol, *pRowIndex, endWin.ekey, binarySearchForKey, while ((groupId == gpIdCol[(*pRowIndex)] && startTsCol[*pRowIndex] < endWin.ekey)) {
NULL, TSDB_ORDER_ASC); (*pRowIndex) += 1;
if ((*pRowIndex) == pDataBlockInfo->rows) {
break;
}
}
} }
do { do {
preWin = endWin; preWin = endWin;
getNextTimeWindow(pInterval, &endWin, TSDB_ORDER_ASC); getNextTimeWindow(pInterval, &endWin, TSDB_ORDER_ASC);
} while (endTsCol[(*pRowIndex) - 1] >= endWin.skey); } while (endTsCol[(*pRowIndex) - 1] >= endWin.skey);
endWin = preWin; endWin = preWin;
if (win.ekey == endWin.ekey || (*pRowIndex) == pDataBlockInfo->rows) { if (win.ekey == endWin.ekey || (*pRowIndex) == pDataBlockInfo->rows || groupId != gpIdCol[*pRowIndex]) {
win.ekey = endWin.ekey; win.ekey = endWin.ekey;
return win; return win;
} }
@ -1235,11 +1245,13 @@ static int32_t generateIntervalScanRange(SStreamScanInfo* pInfo, SSDataBlock* pS
int64_t version = pSrcBlock->info.version - 1; int64_t version = pSrcBlock->info.version - 1;
for (int32_t i = 0; i < rows;) { for (int32_t i = 0; i < rows;) {
uint64_t srcUid = srcUidData[i]; uint64_t srcUid = srcUidData[i];
uint64_t groupId = getGroupIdByData(pInfo, srcUid, srcStartTsCol[i], version); uint64_t groupId = srcGp[i];
uint64_t srcGpId = srcGp[i]; if (groupId == 0) {
TSKEY calStartTs = srcStartTsCol[i]; groupId = getGroupIdByData(pInfo, srcUid, srcStartTsCol[i], version);
}
TSKEY calStartTs = srcStartTsCol[i];
colDataAppend(pCalStartTsCol, pDestBlock->info.rows, (const char*)(&calStartTs), false); colDataAppend(pCalStartTsCol, pDestBlock->info.rows, (const char*)(&calStartTs), false);
STimeWindow win = getSlidingWindow(srcStartTsCol, srcEndTsCol, &pInfo->interval, &pSrcBlock->info, &i, STimeWindow win = getSlidingWindow(srcStartTsCol, srcEndTsCol, srcGp, &pInfo->interval, &pSrcBlock->info, &i,
pInfo->partitionSup.needCalc); pInfo->partitionSup.needCalc);
TSKEY calEndTs = srcStartTsCol[i - 1]; TSKEY calEndTs = srcStartTsCol[i - 1];
colDataAppend(pCalEndTsCol, pDestBlock->info.rows, (const char*)(&calEndTs), false); colDataAppend(pCalEndTsCol, pDestBlock->info.rows, (const char*)(&calEndTs), false);
@ -1248,15 +1260,6 @@ static int32_t generateIntervalScanRange(SStreamScanInfo* pInfo, SSDataBlock* pS
colDataAppend(pEndTsCol, pDestBlock->info.rows, (const char*)(&win.ekey), false); colDataAppend(pEndTsCol, pDestBlock->info.rows, (const char*)(&win.ekey), false);
colDataAppend(pGpCol, pDestBlock->info.rows, (const char*)(&groupId), false); colDataAppend(pGpCol, pDestBlock->info.rows, (const char*)(&groupId), false);
pDestBlock->info.rows++; pDestBlock->info.rows++;
if (pInfo->partitionSup.needCalc && srcGpId != 0 && groupId != srcGpId) {
colDataAppend(pCalStartTsCol, pDestBlock->info.rows, (const char*)(&calStartTs), false);
colDataAppend(pCalEndTsCol, pDestBlock->info.rows, (const char*)(&calEndTs), false);
colDataAppend(pDeUidCol, pDestBlock->info.rows, (const char*)(&srcUid), false);
colDataAppend(pStartTsCol, pDestBlock->info.rows, (const char*)(&win.skey), false);
colDataAppend(pEndTsCol, pDestBlock->info.rows, (const char*)(&win.ekey), false);
colDataAppend(pGpCol, pDestBlock->info.rows, (const char*)(&srcGpId), false);
pDestBlock->info.rows++;
}
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -1331,7 +1334,7 @@ void appendOneRow(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, uint64_t*
static void checkUpdateData(SStreamScanInfo* pInfo, bool invertible, SSDataBlock* pBlock, bool out) { static void checkUpdateData(SStreamScanInfo* pInfo, bool invertible, SSDataBlock* pBlock, bool out) {
if (out) { if (out) {
blockDataCleanup(pInfo->pUpdateDataRes); blockDataCleanup(pInfo->pUpdateDataRes);
blockDataEnsureCapacity(pInfo->pUpdateDataRes, pBlock->info.rows); blockDataEnsureCapacity(pInfo->pUpdateDataRes, pBlock->info.rows * 2);
} }
SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, pInfo->primaryTsIndex); SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, pInfo->primaryTsIndex);
ASSERT(pColDataInfo->info.type == TSDB_DATA_TYPE_TIMESTAMP); ASSERT(pColDataInfo->info.type == TSDB_DATA_TYPE_TIMESTAMP);
@ -1352,10 +1355,12 @@ static void checkUpdateData(SStreamScanInfo* pInfo, bool invertible, SSDataBlock
isDeletedStreamWindow(&win, pBlock->info.groupId, pInfo->pTableScanOp, &pInfo->twAggSup); isDeletedStreamWindow(&win, pBlock->info.groupId, pInfo->pTableScanOp, &pInfo->twAggSup);
if ((update || closedWin) && out) { if ((update || closedWin) && out) {
qDebug("stream update check not pass, update %d, closedWin %d", update, closedWin); qDebug("stream update check not pass, update %d, closedWin %d", update, closedWin);
uint64_t gpId = closedWin && pInfo->partitionSup.needCalc uint64_t gpId = 0;
? calGroupIdByData(&pInfo->partitionSup, pInfo->pPartScalarSup, pBlock, rowId)
: 0;
appendOneRow(pInfo->pUpdateDataRes, tsCol + rowId, tsCol + rowId, &pBlock->info.uid, &gpId); appendOneRow(pInfo->pUpdateDataRes, tsCol + rowId, tsCol + rowId, &pBlock->info.uid, &gpId);
if (closedWin && pInfo->partitionSup.needCalc) {
gpId = calGroupIdByData(&pInfo->partitionSup, pInfo->pPartScalarSup, pBlock, rowId);
appendOneRow(pInfo->pUpdateDataRes, tsCol + rowId, tsCol + rowId, &pBlock->info.uid, &gpId);
}
} }
} }
if (out && pInfo->pUpdateDataRes->info.rows > 0) { if (out && pInfo->pUpdateDataRes->info.rows > 0) {
@ -1532,6 +1537,30 @@ static int32_t filterDelBlockByUid(SSDataBlock* pDst, const SSDataBlock* pSrc, S
return 0; return 0;
} }
// for partition by tag
static void setBlockGroupIdByUid(SStreamScanInfo* pInfo, SSDataBlock* pBlock) {
SColumnInfoData* pStartTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
TSKEY* startTsCol = (TSKEY*)pStartTsCol->pData;
SColumnInfoData* pGpCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
uint64_t* gpCol = (uint64_t*)pGpCol->pData;
SColumnInfoData* pUidCol = taosArrayGet(pBlock->pDataBlock, UID_COLUMN_INDEX);
uint64_t* uidCol = (uint64_t*)pUidCol->pData;
int32_t rows = pBlock->info.rows;
if (!pInfo->partitionSup.needCalc) {
for (int32_t i = 0; i < rows; i++) {
uint64_t groupId = getGroupIdByUid(pInfo, uidCol[i]);
colDataAppend(pGpCol, i, (const char*)&groupId, false);
}
} else {
// SSDataBlock* pPreRes = readPreVersionData(pInfo->pTableScanOp, uidCol[i], startTsCol, ts, maxVersion);
// if (!pPreRes || pPreRes->info.rows == 0) {
// return 0;
// }
// ASSERT(pPreRes->info.rows == 1);
// return calGroupIdByData(&pInfo->partitionSup, pInfo->pPartScalarSup, pPreRes, 0);
}
}
static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
// NOTE: this operator does never check if current status is done or not // NOTE: this operator does never check if current status is done or not
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
@ -1628,7 +1657,8 @@ FETCH_NEXT_BLOCK:
} else { } else {
pDelBlock = pBlock; pDelBlock = pBlock;
} }
printDataBlock(pBlock, "stream scan delete recv filtered"); setBlockGroupIdByUid(pInfo, pDelBlock);
printDataBlock(pDelBlock, "stream scan delete recv filtered");
if (!isIntervalWindow(pInfo) && !isSessionWindow(pInfo) && !isStateWindow(pInfo)) { if (!isIntervalWindow(pInfo) && !isSessionWindow(pInfo) && !isStateWindow(pInfo)) {
generateDeleteResultBlock(pInfo, pDelBlock, pInfo->pDeleteDataRes); generateDeleteResultBlock(pInfo, pDelBlock, pInfo->pDeleteDataRes);
pInfo->pDeleteDataRes->info.type = STREAM_DELETE_RESULT; pInfo->pDeleteDataRes->info.type = STREAM_DELETE_RESULT;

1113
source/libs/executor/src/tfill.c
View File

File diff suppressed because it is too large Load Diff

77
source/libs/executor/src/timewindowoperator.c
View File

@ -271,6 +271,10 @@ static void getNextTimeWindow(SInterval* pInterval, int32_t precision, int32_t o
tw->ekey -= 1; tw->ekey -= 1;
} }
void getNextIntervalWindow(SInterval* pInterval, STimeWindow* tw, int32_t order) {
getNextTimeWindow(pInterval, pInterval->precision, order, tw);
}
void doTimeWindowInterpolation(SArray* pPrevValues, SArray* pDataBlock, TSKEY prevTs, int32_t prevRowIndex, TSKEY curTs, void doTimeWindowInterpolation(SArray* pPrevValues, SArray* pDataBlock, TSKEY prevTs, int32_t prevRowIndex, TSKEY curTs,
int32_t curRowIndex, TSKEY windowKey, int32_t type, SExprSupp* pSup) { int32_t curRowIndex, TSKEY windowKey, int32_t type, SExprSupp* pSup) {
SqlFunctionCtx* pCtx = pSup->pCtx; SqlFunctionCtx* pCtx = pSup->pCtx;
@ -3146,20 +3150,21 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
} }
doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf); doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
if (pInfo->binfo.pRes->info.rows == 0) { if (pInfo->binfo.pRes->info.rows != 0) {
pOperator->status = OP_EXEC_DONE; printDataBlock(pInfo->binfo.pRes, IS_FINAL_OP(pInfo) ? "interval final" : "interval semi");
if (!IS_FINAL_OP(pInfo)) { return pInfo->binfo.pRes;
clearFunctionContext(&pOperator->exprSupp);
// semi interval operator clear disk buffer
clearStreamIntervalOperator(pInfo);
qDebug("===stream===clear semi operator");
} else {
freeAllPages(pInfo->pRecycledPages, pInfo->aggSup.pResultBuf);
}
return NULL;
} }
printDataBlock(pInfo->binfo.pRes, IS_FINAL_OP(pInfo) ? "interval final" : "interval semi");
return pInfo->binfo.pRes; doSetOperatorCompleted(pOperator);
if (!IS_FINAL_OP(pInfo)) {
clearFunctionContext(&pOperator->exprSupp);
// semi interval operator clear disk buffer
clearStreamIntervalOperator(pInfo);
qDebug("===stream===clear semi operator");
} else {
freeAllPages(pInfo->pRecycledPages, pInfo->aggSup.pResultBuf);
}
return NULL;
} else { } else {
if (!IS_FINAL_OP(pInfo)) { if (!IS_FINAL_OP(pInfo)) {
doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf); doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
@ -3316,7 +3321,13 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
return pInfo->pPullDataRes; return pInfo->pPullDataRes;
} }
// we should send result first. doBuildDeleteResult(pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes);
if (pInfo->pDelRes->info.rows != 0) {
// process the rest of the data
printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "interval final" : "interval semi");
return pInfo->pDelRes;
}
doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf); doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
if (pInfo->binfo.pRes->info.rows != 0) { if (pInfo->binfo.pRes->info.rows != 0) {
printDataBlock(pInfo->binfo.pRes, IS_FINAL_OP(pInfo) ? "interval final" : "interval semi"); printDataBlock(pInfo->binfo.pRes, IS_FINAL_OP(pInfo) ? "interval final" : "interval semi");
@ -3330,13 +3341,6 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
// process the rest of the data // process the rest of the data
return pInfo->pUpdateRes; return pInfo->pUpdateRes;
} }
doBuildDeleteResult(pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes);
if (pInfo->pDelRes->info.rows != 0) {
// process the rest of the data
printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "interval final" : "interval semi");
return pInfo->pDelRes;
}
return NULL; return NULL;
} }
@ -5744,19 +5748,18 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
if (pOperator->status == OP_RES_TO_RETURN) { if (pOperator->status == OP_RES_TO_RETURN) {
doBuildDeleteResult(pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes); doBuildDeleteResult(pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes);
if (pInfo->pDelRes->info.rows > 0) { if (pInfo->pDelRes->info.rows > 0) {
printDataBlock(pInfo->pDelRes, "single interval"); printDataBlock(pInfo->pDelRes, "single interval delete");
return pInfo->pDelRes; return pInfo->pDelRes;
} }
doBuildResult(pOperator, pInfo->binfo.pRes, &pInfo->groupResInfo); doBuildResult(pOperator, pInfo->binfo.pRes, &pInfo->groupResInfo);
// doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf); if (pInfo->binfo.pRes->info.rows > 0) {
if (pInfo->binfo.pRes->info.rows == 0 || !hasRemainResults(&pInfo->groupResInfo)) { printDataBlock(pInfo->binfo.pRes, "single interval");
pOperator->status = OP_EXEC_DONE; return pInfo->binfo.pRes;
qDebug("===stream===single interval is done");
freeAllPages(pInfo->pRecycledPages, pInfo->aggSup.pResultBuf);
} }
printDataBlock(pInfo->binfo.pRes, "single interval");
return pInfo->binfo.pRes->info.rows == 0 ? NULL : pInfo->binfo.pRes; doSetOperatorCompleted(pOperator);
return NULL;
} }
SOperatorInfo* downstream = pOperator->pDownstream[0]; SOperatorInfo* downstream = pOperator->pDownstream[0];
@ -5823,24 +5826,24 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
} }
taosArraySort(pUpdated, resultrowComparAsc); taosArraySort(pUpdated, resultrowComparAsc);
// new disc buf
// finalizeUpdatedResult(pOperator->exprSupp.numOfExprs, pInfo->aggSup.pResultBuf, pUpdated,
// pSup->rowEntryInfoOffset);
initMultiResInfoFromArrayList(&pInfo->groupResInfo, pUpdated); initMultiResInfoFromArrayList(&pInfo->groupResInfo, pUpdated);
blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity); blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
removeDeleteResults(pUpdatedMap, pInfo->pDelWins); removeDeleteResults(pUpdatedMap, pInfo->pDelWins);
taosHashCleanup(pUpdatedMap); taosHashCleanup(pUpdatedMap);
doBuildDeleteResult(pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes); doBuildDeleteResult(pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes);
if (pInfo->pDelRes->info.rows > 0) { if (pInfo->pDelRes->info.rows > 0) {
printDataBlock(pInfo->pDelRes, "single interval"); printDataBlock(pInfo->pDelRes, "single interval delete");
return pInfo->pDelRes; return pInfo->pDelRes;
} }
// doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
// new disc buf
doBuildResult(pOperator, pInfo->binfo.pRes, &pInfo->groupResInfo); doBuildResult(pOperator, pInfo->binfo.pRes, &pInfo->groupResInfo);
printDataBlock(pInfo->binfo.pRes, "single interval"); if (pInfo->binfo.pRes->info.rows > 0) {
return pInfo->binfo.pRes->info.rows == 0 ? NULL : pInfo->binfo.pRes; printDataBlock(pInfo->binfo.pRes, "single interval");
return pInfo->binfo.pRes;
}
return NULL;
} }
void destroyStreamIntervalOperatorInfo(void* param) { void destroyStreamIntervalOperatorInfo(void* param) {

1
source/libs/index/src/indexCache.c
View File

@ -302,6 +302,7 @@ static int32_t cacheSearchCompareFunc_JSON(void* cache, SIndexTerm* term, SIdxTR
char* p = taosMemoryCalloc(1, strlen(c->colVal) + 1); char* p = taosMemoryCalloc(1, strlen(c->colVal) + 1);
memcpy(p, c->colVal, strlen(c->colVal)); memcpy(p, c->colVal, strlen(c->colVal));
cond = cmpFn(p + skip, term->colVal, dType); cond = cmpFn(p + skip, term->colVal, dType);
taosMemoryFree(p);
} }
} }
if (cond == MATCH) { if (cond == MATCH) {

42
source/libs/index/src/indexFstFile.c
View File

@ -69,6 +69,8 @@ static int idxFileCtxDoReadFrom(IFileCtx* ctx, uint8_t* buf, int len, int32_t of
int32_t blkOffset = offset % kBlockSize; int32_t blkOffset = offset % kBlockSize;
int32_t blkLeft = kBlockSize - blkOffset; int32_t blkLeft = kBlockSize - blkOffset;
if (offset >= ctx->file.size) return 0;
do { do {
char key[128] = {0}; char key[128] = {0};
idxGenLRUKey(key, ctx->file.buf, blkId); idxGenLRUKey(key, ctx->file.buf, blkId);
@ -80,24 +82,34 @@ static int idxFileCtxDoReadFrom(IFileCtx* ctx, uint8_t* buf, int len, int32_t of
memcpy(buf + total, blk->buf + blkOffset, nread); memcpy(buf + total, blk->buf + blkOffset, nread);
taosLRUCacheRelease(ctx->lru, h, false); taosLRUCacheRelease(ctx->lru, h, false);
} else { } else {
int32_t cacheMemSize = sizeof(SDataBlock) + kBlockSize; int32_t left = ctx->file.size - offset;
if (left < kBlockSize) {
nread = TMIN(left, len);
int32_t bytes = taosPReadFile(ctx->file.pFile, buf + total, nread, offset);
assert(bytes == nread);
SDataBlock* blk = taosMemoryCalloc(1, cacheMemSize); total += bytes;
blk->blockId = blkId; return total;
blk->nread = taosPReadFile(ctx->file.pFile, blk->buf, kBlockSize, blkId * kBlockSize); } else {
assert(blk->nread <= kBlockSize); int32_t cacheMemSize = sizeof(SDataBlock) + kBlockSize;
if (blk->nread < kBlockSize && blk->nread < len) { SDataBlock* blk = taosMemoryCalloc(1, cacheMemSize);
break; blk->blockId = blkId;
} blk->nread = taosPReadFile(ctx->file.pFile, blk->buf, kBlockSize, blkId * kBlockSize);
assert(blk->nread <= kBlockSize);
nread = TMIN(blkLeft, len); if (blk->nread < kBlockSize && blk->nread < len) {
memcpy(buf + total, blk->buf + blkOffset, nread); break;
}
LRUStatus s = taosLRUCacheInsert(ctx->lru, key, strlen(key), blk, cacheMemSize, deleteDataBlockFromLRU, NULL, nread = TMIN(blkLeft, len);
TAOS_LRU_PRIORITY_LOW); memcpy(buf + total, blk->buf + blkOffset, nread);
if (s != TAOS_LRU_STATUS_OK) {
return -1; LRUStatus s = taosLRUCacheInsert(ctx->lru, key, strlen(key), blk, cacheMemSize, deleteDataBlockFromLRU, NULL,
TAOS_LRU_PRIORITY_LOW);
if (s != TAOS_LRU_STATUS_OK) {
return -1;
}
} }
} }
total += nread; total += nread;
@ -146,9 +158,7 @@ IFileCtx* idxFileCtxCreate(WriterType type, const char* path, bool readOnly, int
} else { } else {
ctx->file.pFile = taosOpenFile(path, TD_FILE_READ); ctx->file.pFile = taosOpenFile(path, TD_FILE_READ);
int64_t size = 0;
taosFStatFile(ctx->file.pFile, &ctx->file.size, NULL); taosFStatFile(ctx->file.pFile, &ctx->file.size, NULL);
ctx->file.size = (int)size;
#ifdef USE_MMAP #ifdef USE_MMAP
ctx->file.ptr = (char*)tfMmapReadOnly(ctx->file.pFile, ctx->file.size); ctx->file.ptr = (char*)tfMmapReadOnly(ctx->file.pFile, ctx->file.size);
#endif #endif

4
source/libs/index/test/jsonUT.cc
View File

@ -172,9 +172,9 @@ TEST_F(JsonEnv, testWriteMillonData) {
{ {
std::string colName("voltagefdadfa"); std::string colName("voltagefdadfa");
std::string colVal("abxxxxxxxxxxxx"); std::string colVal("abxxxxxxxxxxxx");
for (int i = 0; i < 10; i++) { for (int i = 0; i < 10000; i++) {
colVal[i % colVal.size()] = '0' + i % 128; colVal[i % colVal.size()] = '0' + i % 128;
for (size_t i = 0; i < 100; i++) { for (size_t i = 0; i < 10; i++) {
SIndexTerm* term = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), SIndexTerm* term = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
colVal.c_str(), colVal.size()); colVal.c_str(), colVal.size());

3
source/libs/nodes/src/nodesCodeFuncs.c
View File

@ -254,6 +254,7 @@ const char* nodesNodeName(ENodeType type) {
case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL: case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL:
return "PhysiStreamSemiInterval"; return "PhysiStreamSemiInterval";
case QUERY_NODE_PHYSICAL_PLAN_FILL: case QUERY_NODE_PHYSICAL_PLAN_FILL:
case QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL:
return "PhysiFill"; return "PhysiFill";
case QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION: case QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION:
return "PhysiSessionWindow"; return "PhysiSessionWindow";
@ -4635,6 +4636,7 @@ static int32_t specificNodeToJson(const void* pObj, SJson* pJson) {
case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL: case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL:
return physiIntervalNodeToJson(pObj, pJson); return physiIntervalNodeToJson(pObj, pJson);
case QUERY_NODE_PHYSICAL_PLAN_FILL: case QUERY_NODE_PHYSICAL_PLAN_FILL:
case QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL:
return physiFillNodeToJson(pObj, pJson); return physiFillNodeToJson(pObj, pJson);
case QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION: case QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION:
case QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION: case QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION:
@ -4788,6 +4790,7 @@ static int32_t jsonToSpecificNode(const SJson* pJson, void* pObj) {
case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL: case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL:
return jsonToPhysiIntervalNode(pJson, pObj); return jsonToPhysiIntervalNode(pJson, pObj);
case QUERY_NODE_PHYSICAL_PLAN_FILL: case QUERY_NODE_PHYSICAL_PLAN_FILL:
case QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL:
return jsonToPhysiFillNode(pJson, pObj); return jsonToPhysiFillNode(pJson, pObj);
case QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION: case QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION:
case QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION: case QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION:

2
source/libs/nodes/src/nodesMsgFuncs.c
View File

@ -3633,6 +3633,7 @@ static int32_t specificNodeToMsg(const void* pObj, STlvEncoder* pEncoder) {
code = physiIntervalNodeToMsg(pObj, pEncoder); code = physiIntervalNodeToMsg(pObj, pEncoder);
break; break;
case QUERY_NODE_PHYSICAL_PLAN_FILL: case QUERY_NODE_PHYSICAL_PLAN_FILL:
case QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL:
code = physiFillNodeToMsg(pObj, pEncoder); code = physiFillNodeToMsg(pObj, pEncoder);
break; break;
case QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION: case QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION:
@ -3770,6 +3771,7 @@ static int32_t msgToSpecificNode(STlvDecoder* pDecoder, void* pObj) {
code = msgToPhysiIntervalNode(pDecoder, pObj); code = msgToPhysiIntervalNode(pDecoder, pObj);
break; break;
case QUERY_NODE_PHYSICAL_PLAN_FILL: case QUERY_NODE_PHYSICAL_PLAN_FILL:
case QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL:
code = msgToPhysiFillNode(pDecoder, pObj); code = msgToPhysiFillNode(pDecoder, pObj);
break; break;
case QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION: case QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION:

4
source/libs/nodes/src/nodesUtilFuncs.c
View File

@ -511,6 +511,7 @@ SNode* nodesMakeNode(ENodeType type) {
case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL: case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL:
return makeNode(type, sizeof(SStreamSemiIntervalPhysiNode)); return makeNode(type, sizeof(SStreamSemiIntervalPhysiNode));
case QUERY_NODE_PHYSICAL_PLAN_FILL: case QUERY_NODE_PHYSICAL_PLAN_FILL:
case QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL:
return makeNode(type, sizeof(SFillPhysiNode)); return makeNode(type, sizeof(SFillPhysiNode));
case QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION: case QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION:
return makeNode(type, sizeof(SSessionWinodwPhysiNode)); return makeNode(type, sizeof(SSessionWinodwPhysiNode));
@ -1156,7 +1157,8 @@ void nodesDestroyNode(SNode* pNode) {
case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL: case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL:
destroyWinodwPhysiNode((SWinodwPhysiNode*)pNode); destroyWinodwPhysiNode((SWinodwPhysiNode*)pNode);
break; break;
case QUERY_NODE_PHYSICAL_PLAN_FILL: { case QUERY_NODE_PHYSICAL_PLAN_FILL:
case QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL: {
SFillPhysiNode* pPhyNode = (SFillPhysiNode*)pNode; SFillPhysiNode* pPhyNode = (SFillPhysiNode*)pNode;
destroyPhysiNode((SPhysiNode*)pPhyNode); destroyPhysiNode((SPhysiNode*)pPhyNode);
nodesDestroyList(pPhyNode->pFillExprs); nodesDestroyList(pPhyNode->pFillExprs);

4
source/libs/planner/src/planPhysiCreater.c
View File

@ -1409,7 +1409,9 @@ static int32_t createPartitionPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChi
static int32_t createFillPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren, SFillLogicNode* pFillNode, static int32_t createFillPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren, SFillLogicNode* pFillNode,
SPhysiNode** pPhyNode) { SPhysiNode** pPhyNode) {
SFillPhysiNode* pFill = (SFillPhysiNode*)makePhysiNode(pCxt, (SLogicNode*)pFillNode, QUERY_NODE_PHYSICAL_PLAN_FILL); SFillPhysiNode* pFill = (SFillPhysiNode*)makePhysiNode(
pCxt, (SLogicNode*)pFillNode,
pCxt->pPlanCxt->streamQuery ? QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL : QUERY_NODE_PHYSICAL_PLAN_FILL);
if (NULL == pFill) { if (NULL == pFill) {
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;
} }

67
source/libs/stream/src/streamState.c
View File

@ -40,6 +40,11 @@ SStreamState* streamStateOpen(char* path, SStreamTask* pTask, bool specPath) {
goto _err; goto _err;
} }
// todo refactor
if (tdbTbOpen("func.state.db", sizeof(SWinKey), -1, SWinKeyCmpr, pState->db, &pState->pFillStateDb) < 0) {
goto _err;
}
if (tdbTbOpen("func.state.db", sizeof(STupleKey), -1, STupleKeyCmpr, pState->db, &pState->pFuncStateDb) < 0) { if (tdbTbOpen("func.state.db", sizeof(STupleKey), -1, STupleKeyCmpr, pState->db, &pState->pFuncStateDb) < 0) {
goto _err; goto _err;
} }
@ -55,6 +60,7 @@ SStreamState* streamStateOpen(char* path, SStreamTask* pTask, bool specPath) {
_err: _err:
tdbTbClose(pState->pStateDb); tdbTbClose(pState->pStateDb);
tdbTbClose(pState->pFuncStateDb); tdbTbClose(pState->pFuncStateDb);
tdbTbClose(pState->pFillStateDb);
tdbClose(pState->db); tdbClose(pState->db);
taosMemoryFree(pState); taosMemoryFree(pState);
return NULL; return NULL;
@ -64,6 +70,7 @@ void streamStateClose(SStreamState* pState) {
tdbCommit(pState->db, &pState->txn); tdbCommit(pState->db, &pState->txn);
tdbTbClose(pState->pStateDb); tdbTbClose(pState->pStateDb);
tdbTbClose(pState->pFuncStateDb); tdbTbClose(pState->pFuncStateDb);
tdbTbClose(pState->pFillStateDb);
tdbClose(pState->db); tdbClose(pState->db);
taosMemoryFree(pState); taosMemoryFree(pState);
@ -126,14 +133,30 @@ int32_t streamStateFuncDel(SStreamState* pState, const STupleKey* key) {
int32_t streamStatePut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen) { int32_t streamStatePut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen) {
return tdbTbUpsert(pState->pStateDb, key, sizeof(SWinKey), value, vLen, &pState->txn); return tdbTbUpsert(pState->pStateDb, key, sizeof(SWinKey), value, vLen, &pState->txn);
} }
// todo refactor
int32_t streamStateFillPut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen) {
return tdbTbUpsert(pState->pFillStateDb, key, sizeof(SWinKey), value, vLen, &pState->txn);
}
int32_t streamStateGet(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen) { int32_t streamStateGet(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen) {
return tdbTbGet(pState->pStateDb, key, sizeof(SWinKey), pVal, pVLen); return tdbTbGet(pState->pStateDb, key, sizeof(SWinKey), pVal, pVLen);
} }
// todo refactor
int32_t streamStateFillGet(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen) {
return tdbTbGet(pState->pFillStateDb, key, sizeof(SWinKey), pVal, pVLen);
}
int32_t streamStateDel(SStreamState* pState, const SWinKey* key) { int32_t streamStateDel(SStreamState* pState, const SWinKey* key) {
return tdbTbDelete(pState->pStateDb, key, sizeof(SWinKey), &pState->txn); return tdbTbDelete(pState->pStateDb, key, sizeof(SWinKey), &pState->txn);
} }
// todo refactor
int32_t streamStateFillDel(SStreamState* pState, const SWinKey* key) {
return tdbTbDelete(pState->pFillStateDb, key, sizeof(SWinKey), &pState->txn);
}
int32_t streamStateAddIfNotExist(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen) { int32_t streamStateAddIfNotExist(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen) {
// todo refactor // todo refactor
int32_t size = *pVLen; int32_t size = *pVLen;
@ -165,6 +188,31 @@ SStreamStateCur* streamStateGetCur(SStreamState* pState, const SWinKey* key) {
return pCur; return pCur;
} }
SStreamStateCur* streamStateFillGetCur(SStreamState* pState, const SWinKey* key) {
SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
if (pCur == NULL) return NULL;
tdbTbcOpen(pState->pFillStateDb, &pCur->pCur, NULL);
int32_t c;
tdbTbcMoveTo(pCur->pCur, key, sizeof(SWinKey), &c);
if (c != 0) {
taosMemoryFree(pCur);
return NULL;
}
return pCur;
}
SStreamStateCur* streamStateGetAndCheckCur(SStreamState* pState, SWinKey* key) {
SStreamStateCur* pCur = streamStateFillGetCur(pState, key);
if (pCur) {
int32_t code = streamStateGetGroupKVByCur(pCur, key, NULL, 0);
if (code == 0) {
return pCur;
}
}
return NULL;
}
int32_t streamStateGetKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen) { int32_t streamStateGetKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen) {
const SWinKey* pKTmp = NULL; const SWinKey* pKTmp = NULL;
int32_t kLen; int32_t kLen;
@ -175,6 +223,17 @@ int32_t streamStateGetKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void**
return 0; return 0;
} }
int32_t streamStateGetGroupKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen) {
uint64_t groupId = pKey->groupId;
int32_t code = streamStateGetKVByCur(pCur, pKey, pVal, pVLen);
if (code == 0) {
if (pKey->groupId == groupId) {
return 0;
}
}
return -1;
}
int32_t streamStateSeekFirst(SStreamState* pState, SStreamStateCur* pCur) { int32_t streamStateSeekFirst(SStreamState* pState, SStreamStateCur* pCur) {
// //
return tdbTbcMoveToFirst(pCur->pCur); return tdbTbcMoveToFirst(pCur->pCur);
@ -185,12 +244,12 @@ int32_t streamStateSeekLast(SStreamState* pState, SStreamStateCur* pCur) {
return tdbTbcMoveToLast(pCur->pCur); return tdbTbcMoveToLast(pCur->pCur);
} }
SStreamStateCur* streamStateSeekKeyNext(SStreamState* pState, const SWinKey* key) { SStreamStateCur* streamStateFillSeekKeyNext(SStreamState* pState, const SWinKey* key) {
SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur)); SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
if (pCur == NULL) { if (pCur == NULL) {
return NULL; return NULL;
} }
if (tdbTbcOpen(pState->pStateDb, &pCur->pCur, NULL) < 0) { if (tdbTbcOpen(pState->pFillStateDb, &pCur->pCur, NULL) < 0) {
taosMemoryFree(pCur); taosMemoryFree(pCur);
return NULL; return NULL;
} }
@ -211,12 +270,12 @@ SStreamStateCur* streamStateSeekKeyNext(SStreamState* pState, const SWinKey* key
return pCur; return pCur;
} }
SStreamStateCur* streamStateSeekKeyPrev(SStreamState* pState, const SWinKey* key) { SStreamStateCur* streamStateFillSeekKeyPrev(SStreamState* pState, const SWinKey* key) {
SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur)); SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
if (pCur == NULL) { if (pCur == NULL) {
return NULL; return NULL;
} }
if (tdbTbcOpen(pState->pStateDb, &pCur->pCur, NULL) < 0) { if (tdbTbcOpen(pState->pFillStateDb, &pCur->pCur, NULL) < 0) {
taosMemoryFree(pCur); taosMemoryFree(pCur);
return NULL; return NULL;
} }

8
source/util/src/tqueue.c
View File

@ -151,15 +151,15 @@ int64_t taosQueueMemorySize(STaosQueue *queue) {
void *taosAllocateQitem(int32_t size, EQItype itype) { void *taosAllocateQitem(int32_t size, EQItype itype) {
STaosQnode *pNode = taosMemoryCalloc(1, sizeof(STaosQnode) + size); STaosQnode *pNode = taosMemoryCalloc(1, sizeof(STaosQnode) + size);
pNode->size = size;
pNode->itype = itype;
pNode->timestamp = taosGetTimestampUs();
if (pNode == NULL) { if (pNode == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL; return NULL;
} }
pNode->size = size;
pNode->itype = itype;
pNode->timestamp = taosGetTimestampUs();
if (itype == RPC_QITEM) { if (itype == RPC_QITEM) {
int64_t alloced = atomic_add_fetch_64(&tsRpcQueueMemoryUsed, size); int64_t alloced = atomic_add_fetch_64(&tsRpcQueueMemoryUsed, size);
if (alloced > tsRpcQueueMemoryAllowed) { if (alloced > tsRpcQueueMemoryAllowed) {

4
source/util/src/tuuid.c
View File

@ -20,8 +20,8 @@ static int32_t tUUIDSerialNo = 0;
int32_t tGenIdPI32(void) { int32_t tGenIdPI32(void) {
if (tUUIDHashId == 0) { if (tUUIDHashId == 0) {
char uid[64]; char uid[65] = {0};
int32_t code = taosGetSystemUUID(uid, tListLen(uid)); int32_t code = taosGetSystemUUID(uid, sizeof(uid));
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
} else { } else {

8
source/util/src/tworker.c
View File

@ -46,7 +46,7 @@ int32_t tQWorkerInit(SQWorkerPool *pool) {
void tQWorkerCleanup(SQWorkerPool *pool) { void tQWorkerCleanup(SQWorkerPool *pool) {
for (int32_t i = 0; i < pool->max; ++i) { for (int32_t i = 0; i < pool->max; ++i) {
SQWorker *worker = pool->workers + i; SQWorker *worker = pool->workers + i;
if (worker == NULL) continue; // if (worker == NULL) continue;
if (taosCheckPthreadValid(worker->thread)) { if (taosCheckPthreadValid(worker->thread)) {
taosQsetThreadResume(pool->qset); taosQsetThreadResume(pool->qset);
} }
@ -54,7 +54,7 @@ void tQWorkerCleanup(SQWorkerPool *pool) {
for (int32_t i = 0; i < pool->max; ++i) { for (int32_t i = 0; i < pool->max; ++i) {
SQWorker *worker = pool->workers + i; SQWorker *worker = pool->workers + i;
if (worker == NULL) continue; // if (worker == NULL) continue;
if (taosCheckPthreadValid(worker->thread)) { if (taosCheckPthreadValid(worker->thread)) {
taosThreadJoin(worker->thread, NULL); taosThreadJoin(worker->thread, NULL);
taosThreadClear(&worker->thread); taosThreadClear(&worker->thread);
@ -138,8 +138,8 @@ STaosQueue *tQWorkerAllocQueue(SQWorkerPool *pool, void *ahandle, FItem fp) {
} }
void tQWorkerFreeQueue(SQWorkerPool *pool, STaosQueue *queue) { void tQWorkerFreeQueue(SQWorkerPool *pool, STaosQueue *queue) {
taosCloseQueue(queue);
uDebug("worker:%s, queue:%p is freed", pool->name, queue); uDebug("worker:%s, queue:%p is freed", pool->name, queue);
taosCloseQueue(queue);
} }
int32_t tWWorkerInit(SWWorkerPool *pool) { int32_t tWWorkerInit(SWWorkerPool *pool) {
@ -283,8 +283,8 @@ STaosQueue *tWWorkerAllocQueue(SWWorkerPool *pool, void *ahandle, FItems fp) {
} }
void tWWorkerFreeQueue(SWWorkerPool *pool, STaosQueue *queue) { void tWWorkerFreeQueue(SWWorkerPool *pool, STaosQueue *queue) {
taosCloseQueue(queue);
uDebug("worker:%s, queue:%p is freed", pool->name, queue); uDebug("worker:%s, queue:%p is freed", pool->name, queue);
taosCloseQueue(queue);
} }
int32_t tSingleWorkerInit(SSingleWorker *pWorker, const SSingleWorkerCfg *pCfg) { int32_t tSingleWorkerInit(SSingleWorker *pWorker, const SSingleWorkerCfg *pCfg) {

11
tests/script/tsim/stream/deleteInterval.sim
View File

@ -413,13 +413,8 @@ if $data12 != 3 then
goto loop14 goto loop14
endi endi
return 1
sql drop stream if exists streams3;
sql drop database if exists test3;
sql drop database if exists test;
sql create database test3 vgroups 4; sql create database test3 vgroups 4;
sql create database test vgroups 1;
sql use test3; sql use test3;
sql create stable st(ts timestamp, a int, b int, c int, d double) tags(ta int,tb int,tc int); sql create stable st(ts timestamp, a int, b int, c int, d double) tags(ta int,tb int,tc int);
sql create table t1 using st tags(1,1,1); sql create table t1 using st tags(1,1,1);
@ -435,7 +430,7 @@ sql delete from t1;
loop15: loop15:
sleep 200 sleep 200
sql select * from test.streamt2 order by c1, c2, c3; sql select * from test.streamt3 order by c1, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
if $loop_count == 10 then if $loop_count == 10 then
@ -453,7 +448,7 @@ sql delete from t1 where ts > 100;
loop16: loop16:
sleep 200 sleep 200
sql select * from test.streamt2 order by c1, c2, c3; sql select * from test.streamt3 order by c1, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
if $loop_count == 10 then if $loop_count == 10 then
@ -471,7 +466,7 @@ sql delete from st;
loop17: loop17:
sleep 200 sleep 200
sql select * from test.streamt2 order by c1, c2, c3; sql select * from test.streamt3 order by c1, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
if $loop_count == 10 then if $loop_count == 10 then

375
tests/script/tsim/stream/fillIntervalDelete0.sim Normal file
View File

@ -0,0 +1,375 @@
$loop_all = 0
looptest:
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/exec.sh -n dnode1 -s start
sleep 200
sql connect
sql drop stream if exists streams1;
sql drop stream if exists streams2;
sql drop stream if exists streams3;
sql drop stream if exists streams4;
sql drop stream if exists streams5;
sql drop database if exists test1;
sql create database test1 vgroups 1;
sql use test1;
sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20));
sql create stream streams1 trigger at_once into streamt1 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(NULL);
sql create stream streams2 trigger at_once into streamt2 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(value,100,200,300);
sql create stream streams3 trigger at_once into streamt3 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(next);
sql create stream streams4 trigger at_once into streamt4 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(prev);
sql create stream streams5 trigger at_once into streamt5 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(linear);
sql insert into t1 values(1648791213000,1,1,1,1.0,'aaa');
sleep 200
$loop_count = 0
loop0:
sleep 200
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
sql select * from streamt1 order by ts;
if $rows != 1 then
print =====rows=$rows
goto loop0
endi
sql delete from t1;
$loop_count = 0
loop1:
sleep 200
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
sql select * from streamt1 order by ts;
if $rows != 0 then
print =====rows1=$rows
goto loop1
endi
sql select * from streamt2 order by ts;
if $rows != 0 then
print =====rows2=$rows
goto loop1
endi
sql select * from streamt3 order by ts;
if $rows != 0 then
print =====rows3=$rows
goto loop1
endi
sql select * from streamt4 order by ts;
if $rows != 0 then
print =====rows4=$rows
goto loop1
endi
sql select * from streamt5 order by ts;
if $rows != 0 then
print =====rows5=$rows
goto loop1
endi
sql insert into t1 values(1648791210000,4,4,4,4.0,'ddd');
sql insert into t1 values(1648791215000,2,2,2,2.0,'bbb');
sql insert into t1 values(1648791217000,3,3,3,3.0,'ccc');
sql insert into t1 values(1648791219000,5,5,5,5.0,'eee');
$loop_count = 0
loop2:
sleep 200
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
sql select * from streamt1 order by ts;
if $rows != 10 then
print =====rows=$rows
goto loop2
endi
#temp
system sh/stop_dnodes.sh
return 1
sql delete from t1 where ts >= 1648791214000;
$loop_count = 0
loop3:
sleep 200
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
sql select * from streamt1 order by ts;
if $rows != 1 then
print =====rows1=$rows
goto loop3
endi
sql select * from streamt2 order by ts;
if $rows != 1 then
print =====rows2=$rows
goto loop3
endi
sql select * from streamt3 order by ts;
if $rows != 1 then
print =====rows3=$rows
goto loop3
endi
sql select * from streamt4 order by ts;
if $rows != 1 then
print =====rows4=$rows
goto loop3
endi
sql select * from streamt5 order by ts;
if $rows != 1 then
print =====rows5=$rows
goto loop3
endi
if $data01 != 4 then
print =====data01=$data01
return -1
endi
sql insert into t1 values(1648791213000,5,5,5,5.0,'eee');
sql insert into t1 values(1648791215000,5,5,5,5.0,'eee');
sql insert into t1 values(1648791219000,6,6,6,6.0,'fff');
$loop_count = 0
loop4:
sleep 200
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
sql select * from streamt1 order by ts;
if $rows != 4 then
print =====rows=$rows
goto loop4
endi
sql delete from t1 where ts <= 1648791216000;
$loop_count = 0
loop5:
sleep 200
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
sql select * from streamt1 order by ts;
if $rows != 1 then
print =====rows1=$rows
goto loop5
endi
sql select * from streamt2 order by ts;
if $rows != 1 then
print =====rows2=$rows
goto loop5
endi
sql select * from streamt3 order by ts;
if $rows != 1 then
print =====rows3=$rows
goto loop5
endi
sql select * from streamt4 order by ts;
if $rows != 1 then
print =====rows4=$rows
goto loop5
endi
sql select * from streamt5 order by ts;
if $rows != 1 then
print =====rows5=$rows
goto loop5
endi
if $data01 != 6 then
print =====data01=$data01
return -1
endi
sql drop stream if exists streams6;
sql drop stream if exists streams7;
sql drop stream if exists streams8;
sql drop stream if exists streams9;
sql drop stream if exists streams10;
sql drop database if exists test6;
sql create database test6 vgroups 1;
sql use test6;
sql create stable st(ts timestamp, a int, b int , c int, d double, s varchar(20)) tags(ta int,tb int,tc int);
sql create table t1 using st tags(1,1,1);
sql create table t2 using st tags(1,1,1);
sql create stream streams6 trigger at_once into streamt6 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(NULL);
sql create stream streams7 trigger at_once into streamt7 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(value,100,200,300);
sql create stream streams8 trigger at_once into streamt8 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(next);
sql create stream streams9 trigger at_once into streamt9 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(prev);
sql create stream streams10 trigger at_once into streamt10 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(linear);
sql insert into t1 values(1648791210000,1,1,1,1.0,'aaa');
sql insert into t1 values(1648791217000,1,1,1,1.0,'aaa');
sql insert into t2 values(1648791215000,1,1,1,1.0,'aaa');
sleep 200
$loop_count = 0
loop7:
sleep 200
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
sql select * from streamt6 order by ts;
if $rows != 8 then
print =====rows=$rows
goto loop7
endi
sql delete from t1;
$loop_count = 0
loop8:
sleep 200
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
sql select * from streamt6 order by ts;
if $rows != 0 then
print =====rows6=$rows
goto loop8
endi
sql select * from streamt7 order by ts;
if $rows != 0 then
print =====rows7=$rows
goto loop8
endi
sql select * from streamt8 order by ts;
if $rows != 0 then
print =====rows8=$rows
goto loop8
endi
sql select * from streamt9 order by ts;
if $rows != 0 then
print =====rows9=$rows
goto loop8
endi
sql select * from streamt10 order by ts;
if $rows != 0 then
print =====rows10=$rows
goto loop8
endi
sql drop stream if exists streams0;
sql drop stream if exists streams1;
sql drop stream if exists streams2;
sql drop stream if exists streams3;
sql drop stream if exists streams4;
sql drop stream if exists streams5;
sql drop stream if exists streams6;
sql drop stream if exists streams7;
sql drop stream if exists streams8;
sql use test1;
sql select * from t1;
print $data00
$loop_all = $loop_all + 1
print ============loop_all=$loop_all
system sh/stop_dnodes.sh
#goto looptest

379
tests/script/tsim/stream/fillIntervalDelete1.sim Normal file
View File

@ -0,0 +1,379 @@
$loop_all = 0
looptest:
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/exec.sh -n dnode1 -s start
sleep 200
sql connect
sql drop stream if exists streams1;
sql drop stream if exists streams2;
sql drop stream if exists streams3;
sql drop stream if exists streams4;
sql drop stream if exists streams5;
sql drop database if exists test1;
sql create database test1 vgroups 1;
sql use test1;
sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20));
sql create stream streams1 trigger at_once into streamt1 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(NULL);
sql create stream streams2 trigger at_once into streamt2 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(value,100,200,300);
sql create stream streams3 trigger at_once into streamt3 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(next);
sql create stream streams4 trigger at_once into streamt4 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(prev);
sql create stream streams5 trigger at_once into streamt5 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(linear);
sql insert into t1 values(1648791210000,0,0,0,0.0,'aaa');
sql insert into t1 values(1648791213000,1,1,1,1.0,'bbb');
sql insert into t1 values(1648791215000,5,5,5,5.0,'ccc');
sql insert into t1 values(1648791217000,6,6,6,6.0,'ddd');
$loop_count = 0
loop0:
sleep 200
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
sql select * from streamt1 order by ts;
if $rows != 8 then
print =====rows=$rows
goto loop0
endi
sql delete from t1 where ts = 1648791213000;
$loop_count = 0
loop2:
sleep 200
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
sql select * from streamt1 order by ts;
if $rows != 8 then
print ====streamt1=rows1=$rows
goto loop2
endi
if $data31 != NULL then
print ====streamt1=data31=$data31
goto loop2
endi
sql select * from streamt2 order by ts;
if $rows != 8 then
print ====streamt2=rows2=$rows
goto loop2
endi
if $data31 != 100 then
print ====streamt2=data31=$data31
goto loop2
endi
sql select * from streamt3 order by ts;
if $rows != 8 then
print ====streamt3=rows3=$rows
goto loop2
endi
if $data31 != 5 then
print ====streamt3=data31=$data31
goto loop2
endi
sql select * from streamt4 order by ts;
if $rows != 8 then
print ====streamt4=rows4=$rows
goto loop2
endi
if $data31 != 0 then
print ====streamt4=data31=$data31
goto loop2
endi
sql select * from streamt5 order by ts;
if $rows != 8 then
print ====streamt5=rows5=$rows
goto loop2
endi
if $data31 != 3 then
print ====streamt5=data31=$data31
goto loop2
endi
sql insert into t1 values(1648791212000,5,5,5,5.0,'eee');
sql insert into t1 values(1648791213000,6,6,6,6.0,'fff');
$loop_count = 0
loop3:
sleep 200
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
sql select * from streamt1 order by ts;
if $data21 != 5 then
print ====133=rows=$rows
goto loop3
endi
if $data31 != 6 then
print ====137=rows=$rows
goto loop3
endi
sql delete from t1 where ts >= 1648791211000 and ts <= 1648791214000;
$loop_count = 0
loop4:
sleep 200
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
sql select * from streamt1 order by ts;
if $rows != 8 then
print ====streamt1=rows1=$rows
goto loop4
endi
if $data31 != NULL then
print ====streamt1=data31=$data31
goto loop4
endi
sql select * from streamt2 order by ts;
if $rows != 8 then
print ====streamt2=rows2=$rows
goto loop4
endi
if $data31 != 100 then
print ====streamt2=data31=$data31
goto loop4
endi
sql select * from streamt3 order by ts;
if $rows != 8 then
print ====streamt3=rows3=$rows
goto loop4
endi
if $data31 != 5 then
print ====streamt3=data31=$data31
goto loop4
endi
sql select * from streamt4 order by ts;
if $rows != 8 then
print ====streamt4=rows4=$rows
goto loop4
endi
if $data31 != 0 then
print ====streamt4=data31=$data31
goto loop4
endi
sql select * from streamt5 order by ts;
if $rows != 8 then
print ====streamt5=rows5=$rows
goto loop4
endi
if $data31 != 3 then
print ====streamt5=data31=$data31
goto loop4
endi
sql drop stream if exists streams6;
sql drop stream if exists streams7;
sql drop stream if exists streams8;
sql drop stream if exists streams9;
sql drop stream if exists streams10;
sql drop database if exists test6;
sql create database test6 vgroups 1;
sql use test6;
sql create stable st(ts timestamp, a int, b int , c int, d double, s varchar(20)) tags(ta int,tb int,tc int);
sql create table t1 using st tags(1,1,1);
sql create table t2 using st tags(1,1,1);
sql create stream streams6 trigger at_once into streamt6 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(NULL);
sql create stream streams7 trigger at_once into streamt7 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(value,100,200,300);
sql create stream streams8 trigger at_once into streamt8 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(next);
sql create stream streams9 trigger at_once into streamt9 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(prev);
sql create stream streams10 trigger at_once into streamt10 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(linear);
sql insert into t1 values(1648791210000,1,1,1,1.0,'aaa');
sql insert into t1 values(1648791215000,6,8,8,8.0,'bbb');
sql insert into t1 values(1648791220000,11,10,10,10.0,'ccc');
sql insert into t1 values(1648791221000,6,6,6,6.0,'fff');
sql insert into t2 values(1648791212000,4,4,4,4.0,'ddd');
sql insert into t2 values(1648791214000,5,5,5,5.0,'eee');
sql insert into t2 values(1648791216000,2,2,2,2.0,'bbb');
sql insert into t2 values(1648791222000,6,6,6,6.0,'fff');
$loop_count = 0
loop5:
sleep 200
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
sql select * from streamt6 order by ts;
if $rows != 13 then
print ====streamt6=rows1=$rows
goto loop5
endi
if $data21 != 4 then
print ====streamt6=data21=$data21
goto loop5
endi
sql delete from t2;
print delete from t2;
$loop_count = 0
loop6:
sleep 200
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
sql select * from streamt6 order by ts;
if $rows != 12 then
print ====streamt6=rows2=$rows
goto loop6
endi
if $data31 != NULL then
print ====streamt6=data31=$data31
goto loop6
endi
sql select * from streamt7 order by ts;
if $rows != 12 then
print ====streamt7=rows2=$rows
goto loop6
endi
if $data31 != 100 then
print ====streamt7=data31=$data31
goto loop6
endi
sql select * from streamt8 order by ts;
if $rows != 12 then
print ====streamt8=rows3=$rows
goto loop6
endi
if $data31 != 6 then
print ====streamt8=data31=$data31
goto loop6
endi
sql select * from streamt9 order by ts;
if $rows != 12 then
print ====streamt9=rows4=$rows
goto loop6
endi
if $data31 != 1 then
print ====streamt9=data31=$data31
goto loop6
endi
sql select * from streamt10 order by ts;
if $rows != 12 then
print ====streamt10=rows5=$rows
goto loop6
endi
if $data21 != 3 then
print ====streamt10=data21=$data21
return -1
endi
if $data31 != 4 then
print ====streamt10=data31=$data31
return -1
endi
if $data71 != 8 then
print ====streamt10=data71=$data71
return -1
endi
if $data91 != 10 then
print ====streamt10=data91=$data91
return -1
endi
sql drop stream if exists streams0;
sql drop stream if exists streams1;
sql drop stream if exists streams2;
sql drop stream if exists streams3;
sql drop stream if exists streams4;
sql drop stream if exists streams5;
sql drop stream if exists streams6;
sql drop stream if exists streams7;
sql drop stream if exists streams8;
sql drop stream if exists streams9;
sql drop stream if exists streams10;
sql use test1;
sql select * from t1;
print $data00
$loop_all = $loop_all + 1
print ============loop_all=$loop_all
system sh/stop_dnodes.sh
#goto looptest

695
tests/script/tsim/stream/fillIntervalLinear.sim Normal file
View File

@ -0,0 +1,695 @@
$loop_all = 0
looptest:
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/exec.sh -n dnode1 -s start
sleep 200
sql connect
sql drop stream if exists streams1;
sql drop database if exists test1;
sql create database test1 vgroups 1;
sql use test1;
sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20));
sql create stream streams1 trigger at_once into streamt1 as select _wstart as ts, max(a)+sum(c), avg(b), first(s), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(linear);
sql insert into t1 values(1648791213000,4,4,4,4.0,'aaa') (1648791216000,5,5,5,5.0,'bbb');
sql insert into t1 values(1648791210000,1,1,1,1.0,'ccc') (1648791219000,2,2,2,2.0,'ddd') (1648791222000,3,3,3,3.0,'eee');
$loop_count = 0
loop1:
sleep 200
sql use test1;
sql select * from streamt1 order by ts;
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
if $rows != 13 then
print =====rows=$rows
goto loop1
endi
if $data01 != 2.000000000 then
print =====data01=$data01
return -1
endi
if $data02 != 1.000000000 then
print =====data02=$data02
return -1
endi
if $data03 != ccc then
print =====data03=$data03
return -1
endi
if $data04 != 1 then
print =====data04=$data04
return -1
endi
if $data11 != 4.000000000 then
print =====data11=$data11
return -1
endi
if $data12 != 2.000000000 then
print =====data12=$data12
return -1
endi
if $data13 != NULL then
print =====data13=$data13
return -1
endi
if $data21 != 6.000000000 then
print =====data21=$data21
return -1
endi
if $data22 != 3.000000000 then
print =====data22=$data22
return -1
endi
if $data23 != NULL then
print =====data23=$data23
return -1
endi
if $data31 != 8.000000000 then
print =====data31=$data31
return -1
endi
if $data32 != 4.000000000 then
print =====data32=$data32
return -1
endi
if $data33 != aaa then
print =====data33=$data33
return -1
endi
if $data41 != 8.666666667 then
print =====data41=$data41
return -1
endi
if $data42 != 4.333333333 then
print =====data42=$data42
return -1
endi
if $data43 != NULL then
print =====data43=$data43
return -1
endi
if $data51 != 9.333333333 then
print =====data01=$data01
return -1
endi
if $data52 != 4.666666667 then
print =====data52=$data52
return -1
endi
if $data53 != NULL then
print =====data53=$data53
return -1
endi
if $data61 != 10.000000000 then
print =====data61=$data61
return -1
endi
if $data62 != 5.000000000 then
print =====data62=$data62
return -1
endi
if $data71 != 8.000000000 then
print =====data71=$data71
return -1
endi
if $data72 != 4.000000000 then
print =====data72=$data72
return -1
endi
if $data81 != 6.000000000 then
print =====data81=$data81
return -1
endi
if $data82 != 3.000000000 then
print =====data82=$data82
return -1
endi
if $data91 != 4.000000000 then
print =====data91=$data91
return -1
endi
if $data92 != 2.000000000 then
print =====data92=$data92
return -1
endi
if $data[10][1] != 4.666666667 then
print =====data[10][1]=$data[10][1]
return -1
endi
if $data[10][2] != 2.333333333 then
print =====data[10][2]=$data[10][2]
return -1
endi
if $data[11][1] != 5.333333333 then
print =====data[11][1]=$data[11][1]
return -1
endi
if $data[11][2] != 2.666666667 then
print =====data[11][2]=$data[11][2]
return -1
endi
if $data[12][1] != 6.000000000 then
print =====data[12][1]=$data[12][1]
return -1
endi
if $data[12][2] != 3.000000000 then
print =====data[12][2]=$data[12][2]
return -1
endi
sql drop stream if exists streams2;
sql drop database if exists test2;
sql create database test2 vgroups 1;
sql use test2;
sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20));
sql create stream streams2 trigger at_once into streamt2 as select _wstart as ts, max(a)+sum(c), avg(b), first(s), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(linear);
sql insert into t1 values(1648791210000,1,1,1,1.0,'ccc') (1648791219000,2,2,2,2.0,'ddd') (1648791222000,3,3,3,3.0,'eee');
sql insert into t1 values(1648791213000,4,4,4,4.0,'aaa') (1648791216000,5,5,5,5.0,'bbb');
$loop_count = 0
loop2:
sleep 200
sql select * from streamt2 order by ts;
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
if $rows != 13 then
print =====rows=$rows
goto loop2
endi
if $data01 != 2.000000000 then
print =====data01=$data01
return -1
endi
if $data02 != 1.000000000 then
print =====data02=$data02
return -1
endi
if $data03 != ccc then
print =====data03=$data03
return -1
endi
if $data04 != 1 then
print =====data04=$data04
return -1
endi
if $data11 != 4.000000000 then
print =====data11=$data11
return -1
endi
if $data12 != 2.000000000 then
print =====data12=$data12
return -1
endi
if $data13 != NULL then
print =====data13=$data13
return -1
endi
if $data21 != 6.000000000 then
print =====data21=$data21
return -1
endi
if $data22 != 3.000000000 then
print =====data22=$data22
return -1
endi
if $data23 != NULL then
print =====data23=$data23
return -1
endi
if $data31 != 8.000000000 then
print =====data31=$data31
return -1
endi
if $data32 != 4.000000000 then
print =====data32=$data32
return -1
endi
if $data33 != aaa then
print =====data33=$data33
return -1
endi
if $data41 != 8.666666667 then
print =====data41=$data41
return -1
endi
if $data42 != 4.333333333 then
print =====data42=$data42
return -1
endi
if $data43 != NULL then
print =====data43=$data43
return -1
endi
if $data51 != 9.333333333 then
print =====data01=$data01
return -1
endi
if $data52 != 4.666666667 then
print =====data52=$data52
return -1
endi
if $data53 != NULL then
print =====data53=$data53
return -1
endi
if $data61 != 10.000000000 then
print =====data61=$data61
return -1
endi
if $data62 != 5.000000000 then
print =====data62=$data62
return -1
endi
if $data71 != 8.000000000 then
print =====data71=$data71
return -1
endi
if $data72 != 4.000000000 then
print =====data72=$data72
return -1
endi
if $data81 != 6.000000000 then
print =====data81=$data81
return -1
endi
if $data82 != 3.000000000 then
print =====data82=$data82
return -1
endi
if $data91 != 4.000000000 then
print =====data91=$data91
return -1
endi
if $data92 != 2.000000000 then
print =====data92=$data92
return -1
endi
if $data[10][1] != 4.666666667 then
print =====data[10][1]=$data[10][1]
return -1
endi
if $data[10][2] != 2.333333333 then
print =====data[10][2]=$data[10][2]
return -1
endi
if $data[11][1] != 5.333333333 then
print =====data[11][1]=$data[11][1]
return -1
endi
if $data[11][2] != 2.666666667 then
print =====data[11][2]=$data[11][2]
return -1
endi
if $data[12][1] != 6.000000000 then
print =====data[12][1]=$data[12][1]
return -1
endi
if $data[12][2] != 3.000000000 then
print =====data[12][2]=$data[12][2]
return -1
endi
sql drop stream if exists streams3;
sql drop database if exists test3;
sql create database test3 vgroups 1;
sql use test3;
sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20));
sql create stream streams3 trigger at_once into streamt3 as select _wstart as ts, max(a), b+c, s, b+1, 1 from t1 where ts >= 1648791150000 and ts < 1648791261000 interval(1s) fill(linear);
sql insert into t1 values(1648791215000,1,1,1,1.0,'aaa');
sql insert into t1 values(1648791217000,2,2,2,2.0,'bbb');
sql insert into t1 values(1648791211000,3,3,3,3.0,'ccc');
sql insert into t1 values(1648791213000,4,4,4,4.0,'ddd');
$loop_count = 0
loop3:
sleep 300
sql select * from streamt3 order by ts;
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
if $rows != 7 then
print =====rows=$rows
goto loop3
endi
if $data01 != 3 then
print =====data01=$data01
return -1
endi
if $data02 != 6.000000000 then
print =====data02=$data02
return -1
endi
if $data03 != ccc then
print =====data03=$data03
return -1
endi
if $data11 != 3 then
print =====data11=$data11
return -1
endi
if $data12 != 7.000000000 then
print =====data12=$data12
return -1
endi
if $data13 != NULL then
print =====data13=$data13
return -1
endi
if $data21 != 4 then
print =====data21=$data21
return -1
endi
if $data22 != 8.000000000 then
print =====data22=$data22
return -1
endi
if $data23 != ddd then
print =====data23=$data23
return -1
endi
if $data31 != 2 then
print =====data31=$data31
return -1
endi
if $data32 != 5.000000000 then
print =====data32=$data32
return -1
endi
if $data33 != NULL then
print =====data33=$data33
return -1
endi
if $data41 != 1 then
print =====data41=$data41
return -1
endi
if $data42 != 2.000000000 then
print =====data42=$data42
return -1
endi
if $data43 != aaa then
print =====data43=$data43
return -1
endi
if $data51 != 1 then
print =====data51=$data51
return -1
endi
if $data52 != 3.000000000 then
print =====data52=$data52
return -1
endi
if $data53 != NULL then
print =====data53=$data53
return -1
endi
if $data61 != 2 then
print =====data61=$data61
return -1
endi
if $data62 != 4.000000000 then
print =====data62=$data62
return -1
endi
if $data63 != bbb then
print =====data63=$data63
return -1
endi
sql insert into t1 values(1648791212000,5,5,5,5.0,'eee');
sql insert into t1 values(1648791207000,6,6,6,6.0,'fff') (1648791209000,7,7,7,7.0,'ggg') (1648791219000,8,8,8,8.0,'hhh') (1648791221000,9,9,9,9.0,'iii');
$loop_count = 0
loop4:
sleep 200
sql select * from test3.streamt3 order by ts;
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
if $rows != 15 then
print =====rows=$rows
goto loop4
endi
if $data01 != 6 then
print =====data01=$data01
return -1
endi
if $data02 != 12.000000000 then
print =====data02=$data02
return -1
endi
if $data03 != fff then
print =====data03=$data03
return -1
endi
if $data11 != 6 then
print =====data11=$data11
return -1
endi
if $data12 != 13.000000000 then
print =====data12=$data12
return -1
endi
if $data13 != NULL then
print =====data13=$data13
return -1
endi
if $data21 != 7 then
print =====data21=$data21
return -1
endi
if $data22 != 14.000000000 then
print =====data22=$data22
return -1
endi
if $data23 != ggg then
print =====data23=$data23
return -1
endi
if $data31 != 5 then
print =====data31=$data31
return -1
endi
if $data32 != 10.000000000 then
print =====data32=$data32
return -1
endi
if $data33 != NULL then
print =====data33=$data33
return -1
endi
if $data51 != 5 then
print =====data51=$data51
return -1
endi
if $data52 != 10.000000000 then
print =====data52=$data52
return -1
endi
if $data53 != eee then
print =====data53=$data53
return -1
endi
if $data[11][1] != 5 then
print =====data[11][1]=$data[11][1]
return -1
endi
if $data[11][2] != 10.000000000 then
print =====data[11][2]=$data[11][2]
return -1
endi
if $data[11][3] != NULL then
print =====data[11][3]=$data[11][3]
return -1
endi
if $data[12][1] != 8 then
print =====data[12][1]=$data[12][1]
return -1
endi
if $data[12][2] != 16.000000000 then
print =====data[12][2]=$data[12][2]
return -1
endi
if $data[12][3] != hhh then
print =====data[12][3]=$data[12][3]
return -1
endi
if $data[13][1] != 8 then
print =====data[13][1]=$data[13][1]
return -1
endi
if $data[13][2] != 17.000000000 then
print =====data[13][2]=$data[13][2]
return -1
endi
if $data[13][3] != NULL then
print =====data[13][3]=$data[13][3]
return -1
endi
if $data[14][1] != 9 then
print =====data[14][1]=$data[14][1]
return -1
endi
if $data[14][2] != 18.000000000 then
print =====data[14][2]=$data[14][2]
return -1
endi
if $data[14][3] != iii then
print =====data[14][3]=$data[14][3]
return -1
endi
sql drop stream if exists streams0;
sql drop stream if exists streams1;
sql drop stream if exists streams2;
sql drop stream if exists streams3;
sql drop stream if exists streams4;
sql drop stream if exists streams5;
sql drop stream if exists streams6;
sql drop stream if exists streams7;
sql drop stream if exists streams8;
sql use test1;
sql select * from t1;
print $data00
$loop_all = $loop_all + 1
print ============loop_all=$loop_all
system sh/stop_dnodes.sh
#goto looptest

171
tests/script/tsim/stream/fillIntervalPartitionBy.sim Normal file
View File

@ -0,0 +1,171 @@
$loop_all = 0
looptest:
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/exec.sh -n dnode1 -s start
sleep 200
sql connect
sql drop stream if exists streams1;
sql drop stream if exists streams2;
sql drop stream if exists streams3;
sql drop stream if exists streams4;
sql drop stream if exists streams5;
sql drop database if exists test1;
sql create database test1 vgroups 1;
sql use test1;
sql create stable st(ts timestamp, a int, b int , c int, d double, s varchar(20)) 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 create stream streams1 trigger at_once into streamt1 as select _wstart as ts, max(a) c1, sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 partition by ta interval(1s) fill(NULL);
sql create stream streams2 trigger at_once into streamt2 as select _wstart as ts, max(a) c1, sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 partition by ta interval(1s) fill(value,100,200,300);
sql create stream streams3 trigger at_once into streamt3 as select _wstart as ts, max(a) c1, sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 partition by ta interval(1s) fill(next);
sql create stream streams4 trigger at_once into streamt4 as select _wstart as ts, max(a) c1, sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 partition by ta interval(1s) fill(prev);
sql create stream streams5 trigger at_once into streamt5 as select _wstart as ts, max(a) c1, sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 partition by ta interval(1s) fill(linear);
sql insert into t1 values(1648791210000,0,0,0,0.0,'aaa');
sql insert into t1 values(1648791213000,1,1,1,1.0,'bbb');
sql insert into t1 values(1648791215000,5,5,5,5.0,'ccc');
sql insert into t1 values(1648791216000,6,6,6,6.0,'ddd');
sql insert into t2 values(1648791210000,7,0,0,0.0,'aaa');
sql insert into t2 values(1648791213000,8,1,1,1.0,'bbb');
sql insert into t2 values(1648791215000,9,5,5,5.0,'ccc');
sql insert into t2 values(1648791216000,10,6,6,6.0,'ddd');
$loop_count = 0
loop2:
sleep 200
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
sql select * from streamt1 order by group_id, ts;
if $rows != 14 then
print ====streamt1=rows1=$rows
goto loop2
endi
sql select * from streamt2 order by group_id, ts;
if $rows != 14 then
print ====streamt2=rows2=$rows
goto loop2
endi
sql select * from streamt3 order by group_id, ts;
if $rows != 14 then
print ====streamt3=rows3=$rows
goto loop2
endi
sql select * from streamt4 order by group_id, ts;
if $rows != 14 then
print ====streamt4=rows4=$rows
goto loop2
endi
sql select * from streamt5 order by group_id, ts;
if $rows != 14 then
print ====streamt5=rows5=$rows
goto loop2
endi
sql delete from t1 where ts = 1648791216000;
print ======delete from t1 where ts = 1648791216000;
$loop_count = 0
loop3:
sleep 200
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
sql select * from streamt1 order by group_id, ts;
if $rows != 13 then
print ====streamt1=rows1=$rows
goto loop3
endi
sql select * from streamt2 order by group_id, ts;
if $rows != 13 then
print ====streamt2=rows2=$rows
goto loop3
endi
sql select * from streamt3 order by group_id, ts;
if $rows != 13 then
print ====streamt3=rows3=$rows
goto loop3
endi
sql select * from streamt4 order by group_id, ts;
if $rows != 13 then
print ====streamt4=rows4=$rows
goto loop3
endi
sql select * from streamt5 order by group_id, ts;
if $rows != 13 then
print ====streamt5=rows5=$rows
goto loop3
endi
sql drop stream if exists streams0;
sql drop stream if exists streams1;
sql drop stream if exists streams2;
sql drop stream if exists streams3;
sql drop stream if exists streams4;
sql drop stream if exists streams5;
sql drop stream if exists streams6;
sql drop stream if exists streams7;
sql drop stream if exists streams8;
sql drop stream if exists streams9;
sql drop stream if exists streams10;
sql use test1;
sql select * from t1;
print $data00
$loop_all = $loop_all + 1
print ============loop_all=$loop_all
system sh/stop_dnodes.sh
#goto looptest

1036
tests/script/tsim/stream/fillIntervalPrevNext.sim Normal file
View File

File diff suppressed because it is too large Load Diff

488
tests/script/tsim/stream/fillIntervalValue.sim Normal file
View File

@ -0,0 +1,488 @@
$loop_all = 0
looptest:
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/exec.sh -n dnode1 -s start
sleep 200
sql connect
sql drop database if exists test;
sql create database test vgroups 1;
sql use test;
sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20));;
sql create stream streams1 trigger at_once into streamt as select _wstart ts, count(*) c1 from t1 where ts > 1648791210000 and ts < 1648791413000 interval(10s) fill(value, 100);
sql insert into t1 values(1648791213000,1,2,3,1.0,'aaa');
sleep 100
sql insert into t1 values(1648791233000,1,2,3,1.0,'aaa');
sql insert into t1 values(1648791223000,1,2,3,1.0,'aaa');
sql insert into t1 values(1648791283000,1,2,3,1.0,'aaa');
sql insert into t1 values(1648791253000,1,2,3,1.0,'aaa');
$loop_count = 0
loop0:
sleep 200
sql select * from streamt order by ts;
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
if $rows != 8 then
print =====rows=$rows
goto loop0
endi
if $data01 != 1 then
print =====data01=$data01
goto loop0
endi
if $data11 != 1 then
print =====data11=$data11
goto loop0
endi
if $data21 != 1 then
print =====data21=$data21
goto loop0
endi
if $data31 != 100 then
print =====data31=$data31
goto loop0
endi
if $data41 != 1 then
print =====data41=$data41
goto loop0
endi
if $data51 != 100 then
print =====data01=$data01
goto loop0
endi
if $data61 != 100 then
print =====data61=$data61
goto loop0
endi
if $data71 != 1 then
print =====data71=$data71
goto loop0
endi
sql drop stream if exists streams2;
sql drop database if exists test2;
sql create database test2 vgroups 1;
sql use test2;
sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20));
sql create stream streams2 trigger at_once into streamt2 as select _wstart as ts, count(*) c1, max(b)+sum(a) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(value, 100,200);
sql insert into t1 values(1648791211000,1,1,1,1.0,'aaa') (1648791217000,2,2,2,2.0,'aaa') (1648791220000,3,3,3,3.0,'aaa');
sql insert into t1 values(1648791213000,4,4,4,4.0,'aaa') (1648791215000,5,5,5,5.0,'aaa');
$loop_count = 0
loop1:
sleep 200
sql select * from streamt2 order by ts;
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
if $rows != 10 then
print =====rows=$rows
goto loop1
endi
if $data01 != 1 then
print =====data01=$data01
goto loop1
endi
if $data02 != 2.000000000 then
print =====data02=$data02
goto loop1
endi
if $data11 != 100 then
print =====data11=$data11
goto loop1
endi
if $data12 != 200.000000000 then
print =====data12=$data12
goto loop1
endi
if $data21 != 1 then
print =====data21=$data21
goto loop1
endi
if $data22 != 8.000000000 then
print =====data22=$data22
goto loop1
endi
if $data31 != 100 then
print =====data31=$data31
goto loop1
endi
if $data32 != 200.000000000 then
print =====data32=$data32
goto loop1
endi
if $data41 != 1 then
print =====data41=$data41
goto loop1
endi
if $data42 != 10.000000000 then
print =====data42=$data42
goto loop1
endi
if $data51 != 100 then
print =====data01=$data01
goto loop1
endi
if $data52 != 200.000000000 then
print =====data52=$data52
goto loop1
endi
if $data61 != 1 then
print =====data61=$data61
goto loop1
endi
if $data62 != 4.000000000 then
print =====data62=$data62
goto loop1
endi
if $data71 != 100 then
print =====data71=$data71
goto loop1
endi
if $data72 != 200.000000000 then
print =====data72=$data72
goto loop1
endi
if $data81 != 100 then
print =====data81=$data81
goto loop1
endi
if $data82 != 200.000000000 then
print =====data82=$data82
goto loop1
endi
if $data91 != 1 then
print =====data91=$data91
goto loop1
endi
if $data92 != 6.000000000 then
print =====data92=$data92
goto loop1
endi
sql drop stream if exists streams3;
sql drop database if exists test3;
sql create database test3 vgroups 1;
sql use test3;
sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20));
sql create stream streams3 trigger at_once into streamt3 as select _wstart as ts, max(b), a+b, c from t1 where ts >= 1648791200000 and ts < 1648791261000 interval(10s) sliding(3s) fill(value, 100,200,300);
sql insert into t1 values(1648791220000,1,1,1,1.0,'aaa');
sleep 100
sql insert into t1 values(1648791260000,1,1,1,1.0,'aaa');
sleep 100
sql insert into t1 values(1648791200000,1,1,1,1.0,'aaa');
$loop_count = 0
loop3:
sleep 200
sql select * from streamt3 order by ts;
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
if $rows != 23 then
print =====rows=$rows
goto loop3
endi
if $data01 != 1 then
print =====data01=$data01
goto loop3
endi
if $data02 != 2.000000000 then
print =====data02=$data02
goto loop3
endi
if $data03 != 1 then
print =====data03=$data03
goto loop3
endi
if $data21 != 1 then
print =====data21=$data21
goto loop3
endi
if $data22 != 2.000000000 then
print =====data22=$data22
goto loop3
endi
if $data23 != 1 then
print =====data23=$data23
goto loop3
endi
if $data31 != 100 then
print =====data31=$data31
goto loop3
endi
if $data32 != 200.000000000 then
print =====data32=$data32
goto loop3
endi
if $data33 != 300 then
print =====data33=$data33
goto loop3
endi
if $data61 != 100 then
print =====data61=$data61
goto loop3
endi
if $data62 != 200.000000000 then
print =====data62=$data62
goto loop3
endi
if $data63 != 300 then
print =====data63=$data63
goto loop3
endi
if $data71 != 1 then
print =====data71=$data71
goto loop3
endi
if $data72 != 2.000000000 then
print =====data72=$data72
goto loop3
endi
if $data73 != 1 then
print =====data73=$data73
goto loop3
endi
if $data91 != 1 then
print =====data91=$data91
goto loop3
endi
if $data92 != 2.000000000 then
print =====data92=$data92
goto loop3
endi
if $data93 != 1 then
print =====data93=$data93
goto loop3
endi
if $data[10][1] != 100 then
print =====data[10][1]=$data[10][1]
goto loop3
endi
if $data[10][2] != 200.000000000 then
print =====data[10][2]=$data[10][2]
goto loop3
endi
if $data[10][3] != 300 then
print =====data[10][3]=$data[10][3]
goto loop3
endi
if $data[19][1] != 100 then
print =====data[19][1]=$data[19][1]
goto loop3
endi
if $data[19][2] != 200.000000000 then
print =====data[19][2]=$data[19][2]
goto loop3
endi
if $data[19][3] != 300 then
print =====data[19][3]=$data[19][3]
goto loop3
endi
if $data[20][1] != 1 then
print =====data[20][1]=$data[20][1]
goto loop3
endi
if $data[20][2] != 2.000000000 then
print =====data[20][2]=$data[20][2]
goto loop3
endi
if $data[20][3] != 1 then
print =====data[20][3]=$data[20][3]
goto loop3
endi
if $data[22][1] != 1 then
print =====data[22][1]=$data[22][1]
goto loop3
endi
if $data[22][2] != 2.000000000 then
print =====data[22][2]=$data[22][2]
goto loop3
endi
if $data[22][3] != 1 then
print =====data[22][3]=$data[22][3]
goto loop3
endi
sql drop stream if exists streams4;
sql drop database if exists test4;
sql create database test4 vgroups 1;
sql use test4;
sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20));;
sql create stream streams4 trigger at_once into streamt4 as select _wstart ts, count(*) c1 from t1 where ts > 1648791210000 and ts < 1648791413000 interval(10s) fill(NULL);
sql insert into t1 values(1648791213000,1,2,3,1.0,'aaa');
sql insert into t1 values(1648791233000,1,2,3,1.0,'aaa');
$loop_count = 0
loop4:
sleep 200
sql select * from streamt4 order by ts;
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
if $rows != 3 then
print =====rows=$rows
goto loop4
endi
if $data11 != NULL then
print =====data11=$data11
goto loop4
endi
sql drop stream if exists streams0;
sql drop stream if exists streams1;
sql drop stream if exists streams2;
sql drop stream if exists streams3;
sql drop stream if exists streams4;
sql drop stream if exists streams5;
sql drop stream if exists streams6;
sql drop stream if exists streams7;
sql drop stream if exists streams8;
sql use test;
sql select * from t1;
print $data00
$loop_all = $loop_all + 1
print ============loop_all=$loop_all
system sh/stop_dnodes.sh
#goto looptest

28
tests/script/tsim/stream/partitionby.sim
View File

@ -5,13 +5,14 @@ sleep 50
sql connect sql connect
sql create database test vgroups 4; sql create database test vgroups 4;
sql create database test0 vgroups 1;
sql use test; sql use test;
sql create stable st(ts timestamp, a int, b int , c int, d double) tags(ta int,tb int,tc int); sql create stable st(ts timestamp, a int, b int , c int, d double) tags(ta int,tb int,tc int);
sql create table ts1 using st tags(1,1,1); sql create table ts1 using st tags(1,1,1);
sql create table ts2 using st tags(2,2,2); sql create table ts2 using st tags(2,2,2);
sql create table ts3 using st tags(3,2,2); sql create table ts3 using st tags(3,2,2);
sql create table ts4 using st tags(4,2,2); sql create table ts4 using st tags(4,2,2);
sql create stream stream_t1 trigger at_once into streamtST1 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5 from st partition by ta,tb,tc interval(10s); sql create stream stream_t1 trigger at_once into test0.streamtST1 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5 from st partition by ta,tb,tc interval(10s);
sql insert into ts1 values(1648791213001,1,12,3,1.0); sql insert into ts1 values(1648791213001,1,12,3,1.0);
sql insert into ts2 values(1648791213001,1,12,3,1.0); sql insert into ts2 values(1648791213001,1,12,3,1.0);
@ -22,7 +23,7 @@ $loop_count = 0
loop0: loop0:
sleep 300 sleep 300
sql select * from streamtST1; sql select * from test0.streamtST1;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
if $loop_count == 10 then if $loop_count == 10 then
@ -34,6 +35,29 @@ print =====rows=$rows
goto loop0 goto loop0
endi endi
sql insert into ts1 values(1648791223001,1,12,3,1.0);
sql insert into ts2 values(1648791223001,1,12,3,1.0);
sql insert into ts3 values(1648791223001,1,12,3,1.0);
sql insert into ts4 values(1648791223001,1,12,3,1.0);
sleep 300
sql delete from st where ts = 1648791223001;
loop00:
sleep 300
sql select * from test0.streamtST1;
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
if $rows != 4 then
print =====rows=$rows
goto loop00
endi
print =====loop0 print =====loop0
sql create database test1 vgroups 1; sql create database test1 vgroups 1;

47
tests/script/tsim/stream/partitionbyColumnInterval.sim
View File

@ -562,6 +562,53 @@ if $data21 != 1 then
goto loop14 goto loop14
endi endi
sql drop stream if exists streams5;
sql drop database if exists test5;
sql create database test5 vgroups 4;
sql use test5;
sql create stable st(ts timestamp, a int, b int, c int, d double) 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 create table t3 using st tags(2,2,2);
sql create table t4 using st tags(2,2,2);
sql create stream streams5 trigger at_once into test.streamt5 as select _wstart c1, count(*) c2, max(a) c3 from st partition by a interval(10s);
sql insert into t1 values(1648791213000,1,2,3,1.0);
sql insert into t2 values(1648791213000,2,2,3,1.0);
sql insert into t3 values(1648791213000,3,2,3,1.0);
sql insert into t4 values(1648791213000,4,2,3,1.0);
sql insert into t1 values(1648791223000,1,2,3,1.0);
sql insert into t2 values(1648791223000,2,2,3,1.0);
sql insert into t3 values(1648791223000,3,2,3,1.0);
sql insert into t4 values(1648791223000,4,2,3,1.0);
sleep 300
sql delete from st where ts = 1648791223000;
sql select * from test.streamt5;
$loop_count = 0
loop15:
sleep 50
sql select * from test.streamt5 order by c1, c2, c3;
$loop_count = $loop_count + 1
if $loop_count == 20 then
return -1
endi
if $rows != 4 then
print =====rows=$rows
print =====rows=$rows
print =====rows=$rows
# goto loop15
endi
$loop_all = $loop_all + 1 $loop_all = $loop_all + 1
print ============loop_all=$loop_all print ============loop_all=$loop_all

2
tools/shell/src/shellArguments.c
View File

@ -358,7 +358,7 @@ static int32_t shellCheckArgs() {
return -1; return -1;
} }
if (pArgs->password != NULL && (strlen(pArgs->password) <= 0)) { if (/*pArgs->password != NULL && */ (strlen(pArgs->password) <= 0)) {
printf("Invalid password\r\n"); printf("Invalid password\r\n");
return -1; return -1;
} }

59
utils/tsim/src/simExe.c
View File

@ -224,10 +224,24 @@ char *simGetVariable(SScript *script, char *varName, int32_t varLen) {
} }
int32_t simExecuteExpression(SScript *script, char *exp) { int32_t simExecuteExpression(SScript *script, char *exp) {
char *op1, *op2, *var1, *var2, *var3, *rest; char *op1 = NULL;
int32_t op1Len, op2Len, var1Len, var2Len, var3Len, val0, val1; char *op2 = NULL;
char t0[1024], t1[1024], t2[1024], t3[2048]; char *var1 = NULL;
int32_t result; char *var2 = NULL;
char *var3 = NULL;
char *rest = NULL;
int32_t op1Len = 0;
int32_t op2Len = 0;
int32_t var1Len = 0;
int32_t var2Len = 0;
int32_t var3Len = 0;
int32_t val0 = 0;
int32_t val1 = 0;
char t0[1024] = {0};
char t1[1024] = {0};
char t2[1024] = {0};
char t3[2048] = {0};
int32_t result = 0;
rest = paGetToken(exp, &var1, &var1Len); rest = paGetToken(exp, &var1, &var1Len);
rest = paGetToken(rest, &op1, &op1Len); rest = paGetToken(rest, &op1, &op1Len);
@ -241,9 +255,9 @@ int32_t simExecuteExpression(SScript *script, char *exp) {
t0[var1Len] = 0; t0[var1Len] = 0;
} }
if (var2[0] == '$') if (var2[0] == '$') {
strcpy(t1, simGetVariable(script, var2 + 1, var2Len - 1)); tstrncpy(t1, simGetVariable(script, var2 + 1, var2Len - 1), 1024);
else { } else {
memcpy(t1, var2, var2Len); memcpy(t1, var2, var2Len);
t1[var2Len] = 0; t1[var2Len] = 0;
} }
@ -258,14 +272,21 @@ int32_t simExecuteExpression(SScript *script, char *exp) {
t2[var3Len] = 0; t2[var3Len] = 0;
} }
int64_t t1l = atoll(t1);
int64_t t2l = atoll(t2);
if (op2[0] == '+') { if (op2[0] == '+') {
sprintf(t3, "%lld", atoll(t1) + atoll(t2)); sprintf(t3, "%" PRId64, t1l + t2l);
} else if (op2[0] == '-') { } else if (op2[0] == '-') {
sprintf(t3, "%lld", atoll(t1) - atoll(t2)); sprintf(t3, "%" PRId64, t1l - t2l);
} else if (op2[0] == '*') { } else if (op2[0] == '*') {
sprintf(t3, "%lld", atoll(t1) * atoll(t2)); sprintf(t3, "%" PRId64, t1l * t2l);
} else if (op2[0] == '/') { } else if (op2[0] == '/') {
sprintf(t3, "%lld", atoll(t1) / atoll(t2)); if (t2l == 0) {
sprintf(t3, "%" PRId64, INT64_MAX);
} else {
sprintf(t3, "%" PRId64, t1l / t2l);
}
} else if (op2[0] == '.') { } else if (op2[0] == '.') {
sprintf(t3, "%s%s", t1, t2); sprintf(t3, "%s%s", t1, t2);
} }
@ -636,7 +657,7 @@ bool simCreateTaosdConnect(SScript *script, char *rest) {
} }
bool simExecuteNativeSqlCommand(SScript *script, char *rest, bool isSlow) { bool simExecuteNativeSqlCommand(SScript *script, char *rest, bool isSlow) {
char timeStr[30] = {0}; char timeStr[80] = {0};
time_t tt; time_t tt;
struct tm tp; struct tm tp;
SCmdLine *line = &script->lines[script->linePos]; SCmdLine *line = &script->lines[script->linePos];
@ -943,7 +964,7 @@ bool simExecuteSqlErrorCmd(SScript *script, char *rest) {
return true; return true;
} }
TAOS_RES *pSql = pSql = taos_query(script->taos, rest); TAOS_RES *pSql = taos_query(script->taos, rest);
int32_t ret = taos_errno(pSql); int32_t ret = taos_errno(pSql);
taos_free_result(pSql); taos_free_result(pSql);
@ -961,7 +982,7 @@ bool simExecuteSqlErrorCmd(SScript *script, char *rest) {
} }
bool simExecuteLineInsertCmd(SScript *script, char *rest) { bool simExecuteLineInsertCmd(SScript *script, char *rest) {
char buf[TSDB_MAX_BINARY_LEN]; char buf[TSDB_MAX_BINARY_LEN] = {0};
simVisuallizeOption(script, rest, buf); simVisuallizeOption(script, rest, buf);
rest = buf; rest = buf;
@ -973,10 +994,7 @@ bool simExecuteLineInsertCmd(SScript *script, char *rest) {
char *lines[] = {rest}; char *lines[] = {rest};
#if 0 #if 0
int32_t ret = taos_insert_lines(script->taos, lines, 1); int32_t ret = taos_insert_lines(script->taos, lines, 1);
#else if (ret == TSDB_CODE_SUCCESS) {
int32_t ret = 0;
#endif
if (ret == TSDB_CODE_SUCCESS) {
simDebug("script:%s, taos:%p, %s executed. success.", script->fileName, script->taos, rest); simDebug("script:%s, taos:%p, %s executed. success.", script->fileName, script->taos, rest);
script->linePos++; script->linePos++;
return true; return true;
@ -985,6 +1003,11 @@ bool simExecuteLineInsertCmd(SScript *script, char *rest) {
tstrerror(ret)); tstrerror(ret));
return false; return false;
} }
#else
simDebug("script:%s, taos:%p, %s executed. success.", script->fileName, script->taos, rest);
script->linePos++;
return true;
#endif
} }
bool simExecuteLineInsertErrorCmd(SScript *script, char *rest) { bool simExecuteLineInsertErrorCmd(SScript *script, char *rest) {

2
utils/tsim/src/simMain.c
View File

@ -33,7 +33,7 @@ int32_t main(int32_t argc, char *argv[]) {
if (strcmp(argv[i], "-c") == 0 && i < argc - 1) { if (strcmp(argv[i], "-c") == 0 && i < argc - 1) {
tstrncpy(configDir, argv[++i], 128); tstrncpy(configDir, argv[++i], 128);
} else if (strcmp(argv[i], "-f") == 0 && i < argc - 1) { } else if (strcmp(argv[i], "-f") == 0 && i < argc - 1) {
strcpy(scriptFile, argv[++i]); tstrncpy(scriptFile, argv[++i], MAX_FILE_NAME_LEN);
} else if (strcmp(argv[i], "-m") == 0) { } else if (strcmp(argv[i], "-m") == 0) {
useMultiProcess = true; useMultiProcess = true;
} else if (strcmp(argv[i], "-v") == 0) { } else if (strcmp(argv[i], "-v") == 0) {

13
utils/tsim/src/simParse.c
View File

@ -175,14 +175,17 @@ SScript *simBuildScriptObj(char *fileName) {
SScript *simParseScript(char *fileName) { SScript *simParseScript(char *fileName) {
TdFilePtr pFile; TdFilePtr pFile;
int32_t tokenLen, lineNum = 0; int32_t tokenLen, lineNum = 0;
char buffer[10*1024], name[128], *token, *rest; char buffer[10 * 1024] = {0};
SCommand *pCmd; char name[PATH_MAX] = {9};
SScript *script; char *token = NULL;
char *rest = NULL;
SCommand *pCmd = NULL;
SScript *script = NULL;
if ((fileName[0] == '.') || (fileName[0] == '/')) { if ((fileName[0] == '.') || (fileName[0] == '/')) {
strcpy(name, fileName); tstrncpy(name, fileName, PATH_MAX);
} else { } else {
sprintf(name, "%s" TD_DIRSEP "%s", simScriptDir, fileName); snprintf(name, PATH_MAX, "%s" TD_DIRSEP "%s", simScriptDir, fileName);
taosRealPath(name, NULL, sizeof(name)); taosRealPath(name, NULL, sizeof(name));
} }