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

Merge branch '3.0' into enh/TD-30554-3.0

This commit is contained in:
kailixu 2024-06-18 08:51:53 +08:00
parent e1f5e6d533 c3ae66c95c
commit d87a7d2809
29 changed files with 178 additions and 113 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cmake/libuv_CMakeLists.txt.in
View File

@ -2,7 +2,7 @@
# libuv # libuv
ExternalProject_Add(libuv ExternalProject_Add(libuv
GIT_REPOSITORY https://github.com/libuv/libuv.git GIT_REPOSITORY https://github.com/libuv/libuv.git
GIT_TAG v1.44.2 GIT_TAG v1.48.0
SOURCE_DIR "${TD_CONTRIB_DIR}/libuv" SOURCE_DIR "${TD_CONTRIB_DIR}/libuv"
BINARY_DIR "${TD_CONTRIB_DIR}/libuv" BINARY_DIR "${TD_CONTRIB_DIR}/libuv"
CONFIGURE_COMMAND "" CONFIGURE_COMMAND ""

2
docs/en/12-taos-sql/14-stream.md
View File

@ -47,7 +47,7 @@ window_clause: {
} }
``` ```
`SESSION` indicates a session window, and `tol_val` indicates the maximum range of the time interval. If the time interval between two continuous rows are within the time interval specified by `tol_val` they belong to the same session window; otherwise a new session window is started automatically. `SESSION` indicates a session window, and `tol_val` indicates the maximum range of the time interval. If the time interval between two continuous rows are within the time interval specified by `tol_val` they belong to the same session window; otherwise a new session window is started automatically.The `_wend` of this window is the time of the last data plus `tol_val`.
`EVENT_WINDOW` is determined according to the window start condition and the window close condition. The window is started when `start_trigger_condition` is evaluated to true, the window is closed when `end_trigger_condition` is evaluated to true. `start_trigger_condition` and `end_trigger_condition` can be any conditional expressions supported by TDengine and can include multiple columns. `EVENT_WINDOW` is determined according to the window start condition and the window close condition. The window is started when `start_trigger_condition` is evaluated to true, the window is closed when `end_trigger_condition` is evaluated to true. `start_trigger_condition` and `end_trigger_condition` can be any conditional expressions supported by TDengine and can include multiple columns.

4
docs/zh/12-taos-sql/14-stream.md
View File

@ -54,8 +54,10 @@ window_clause: {
} }
``` ```
其中，SESSION 是会话窗口，tol_val 是时间间隔的最大范围。在 tol_val 时间间隔范围内的数据都属于同一个窗口，如果连续的两条数据的时间超过 tol_val，则自动开启下一个窗口。 其中，SESSION 是会话窗口，tol_val 是时间间隔的最大范围。在 tol_val 时间间隔范围内的数据都属于同一个窗口，如果连续的两条数据的时间超过 tol_val，则自动开启下一个窗口。该窗口的 _wend 等于最后一条数据的时间加上 tol_val。
EVENT_WINDOW 是事件窗口，根据开始条件和结束条件来划定窗口。当 start_trigger_condition 满足时则窗口开始，直到 end_trigger_condition 满足时窗口关闭。 start_trigger_condition 和 end_trigger_condition 可以是任意 TDengine 支持的条件表达式，且可以包含不同的列。 EVENT_WINDOW 是事件窗口，根据开始条件和结束条件来划定窗口。当 start_trigger_condition 满足时则窗口开始，直到 end_trigger_condition 满足时窗口关闭。 start_trigger_condition 和 end_trigger_condition 可以是任意 TDengine 支持的条件表达式，且可以包含不同的列。
COUNT_WINDOW 是计数窗口,按固定的数据行数来划分窗口。 count_val 是常量，是正整数，必须大于等于2，小于2147483648。 count_val 表示每个 COUNT_WINDOW 包含的最大数据行数，总数据行数不能整除 count_val 时，最后一个窗口的行数会小于 count_val 。 sliding_val 是常量，表示窗口滑动的数量，类似于 INTERVAL 的 SLIDING 。 COUNT_WINDOW 是计数窗口,按固定的数据行数来划分窗口。 count_val 是常量，是正整数，必须大于等于2，小于2147483648。 count_val 表示每个 COUNT_WINDOW 包含的最大数据行数，总数据行数不能整除 count_val 时，最后一个窗口的行数会小于 count_val 。 sliding_val 是常量，表示窗口滑动的数量，类似于 INTERVAL 的 SLIDING 。
窗口的定义与时序数据特色查询中的定义完全相同，详见 [TDengine 特色查询](../distinguished) 窗口的定义与时序数据特色查询中的定义完全相同，详见 [TDengine 特色查询](../distinguished)

6
include/common/tmsg.h
View File

@ -1054,10 +1054,10 @@ typedef struct {
int8_t enable; int8_t enable;
int8_t isView; int8_t isView;
union { union {
int8_t flag; uint8_t flag;
struct { struct {
int8_t createdb : 1; uint8_t createdb : 1;
int8_t reserve : 7; uint8_t reserve : 7;
}; };
}; };
char user[TSDB_USER_LEN]; char user[TSDB_USER_LEN];

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

@ -410,7 +410,7 @@ typedef struct SStateStore {
void (*streamFileStateClear)(struct SStreamFileState* pFileState); void (*streamFileStateClear)(struct SStreamFileState* pFileState);
bool (*needClearDiskBuff)(struct SStreamFileState* pFileState); bool (*needClearDiskBuff)(struct SStreamFileState* pFileState);
SStreamState* (*streamStateOpen)(const char* path, void* pTask, bool specPath, int32_t szPage, int32_t pages); SStreamState* (*streamStateOpen)(const char* path, void* pTask, int64_t streamId, int32_t taskId, bool specPath, int32_t szPage, int32_t pages);
void (*streamStateClose)(SStreamState* pState, bool remove); void (*streamStateClose)(SStreamState* pState, bool remove);
int32_t (*streamStateBegin)(SStreamState* pState); int32_t (*streamStateBegin)(SStreamState* pState);
int32_t (*streamStateCommit)(SStreamState* pState); int32_t (*streamStateCommit)(SStreamState* pState);

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

@ -29,7 +29,8 @@ extern "C" {
#include "storageapi.h" #include "storageapi.h"
SStreamState* streamStateOpen(const char* path, void* pTask, bool specPath, int32_t szPage, int32_t pages); SStreamState* streamStateOpen(const char* path, void* pTask, int64_t streamId, int32_t taskId, bool specPath,
int32_t szPage, int32_t pages);
void streamStateClose(SStreamState* pState, bool remove); void streamStateClose(SStreamState* pState, bool remove);
int32_t streamStateBegin(SStreamState* pState); int32_t streamStateBegin(SStreamState* pState);
int32_t streamStateCommit(SStreamState* pState); int32_t streamStateCommit(SStreamState* pState);

11
include/util/taoserror.h
View File

@ -24,6 +24,14 @@ extern "C" {
// clang-format off // clang-format off
typedef struct {
int32_t val;
const char* str;
const char* macro;
} STaosError;
extern STaosError errors[];
#define TAOS_DEF_ERROR_CODE(mod, code) ((int32_t)((0x80000000 | ((mod)<<16) | (code)))) #define TAOS_DEF_ERROR_CODE(mod, code) ((int32_t)((0x80000000 | ((mod)<<16) | (code))))
#define TAOS_SYSTEM_ERROR(code) (0x80ff0000 | (code)) #define TAOS_SYSTEM_ERROR(code) (0x80ff0000 | (code))
@ -38,6 +46,7 @@ const char* terrstr();
char* taosGetErrMsgReturn(); char* taosGetErrMsgReturn();
char* taosGetErrMsg(); char* taosGetErrMsg();
int32_t* taosGetErrno(); int32_t* taosGetErrno();
int32_t taosGetErrSize();
#define terrno (*taosGetErrno()) #define terrno (*taosGetErrno())
#define terrMsg (taosGetErrMsg()) #define terrMsg (taosGetErrMsg())
@ -316,7 +325,7 @@ int32_t* taosGetErrno();
#define TSDB_CODE_MND_DB_OPTION_UNCHANGED TAOS_DEF_ERROR_CODE(0, 0x038A) // #define TSDB_CODE_MND_DB_OPTION_UNCHANGED TAOS_DEF_ERROR_CODE(0, 0x038A) //
#define TSDB_CODE_MND_DB_INDEX_NOT_EXIST TAOS_DEF_ERROR_CODE(0, 0x038B) #define TSDB_CODE_MND_DB_INDEX_NOT_EXIST TAOS_DEF_ERROR_CODE(0, 0x038B)
#define TSDB_CODE_MND_DB_RETENTION_PERIOD_ZERO TAOS_DEF_ERROR_CODE(0, 0x038C) #define TSDB_CODE_MND_DB_RETENTION_PERIOD_ZERO TAOS_DEF_ERROR_CODE(0, 0x038C)
#define TSDB_CODE_MND_INCONSIST_ENCRYPT_KEY TAOS_DEF_ERROR_CODE(0, 0x038D) // #define TSDB_CODE_MND_INCONSIST_ENCRYPT_KEY TAOS_DEF_ERROR_CODE(0, 0x038D) // unused
#define TSDB_CODE_MND_INVALID_ENCRYPT_KEY TAOS_DEF_ERROR_CODE(0, 0x038E) #define TSDB_CODE_MND_INVALID_ENCRYPT_KEY TAOS_DEF_ERROR_CODE(0, 0x038E)
// #define TSDB_CODE_MND_INVALID_DB_OPTION_DAYS TAOS_DEF_ERROR_CODE(0, 0x0390) // 2.x // #define TSDB_CODE_MND_INVALID_DB_OPTION_DAYS TAOS_DEF_ERROR_CODE(0, 0x0390) // 2.x
// #define TSDB_CODE_MND_INVALID_DB_OPTION_KEEP TAOS_DEF_ERROR_CODE(0, 0x0391) // 2.x // #define TSDB_CODE_MND_INVALID_DB_OPTION_KEEP TAOS_DEF_ERROR_CODE(0, 0x0391) // 2.x

4
source/common/src/tcol.c
View File

@ -327,7 +327,9 @@ int32_t setColCompressByOption(uint8_t type, uint8_t encode, uint16_t compressTy
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
bool useCompress(uint8_t tableType) { return TSDB_SUPER_TABLE == tableType || TSDB_NORMAL_TABLE == tableType; } bool useCompress(uint8_t tableType) {
return TSDB_SUPER_TABLE == tableType || TSDB_NORMAL_TABLE == tableType || TSDB_CHILD_TABLE == tableType;
}
int8_t validColCompressLevel(uint8_t type, uint8_t level) { int8_t validColCompressLevel(uint8_t type, uint8_t level) {
if (level == TSDB_COLVAL_LEVEL_DISABLED) return 1; if (level == TSDB_COLVAL_LEVEL_DISABLED) return 1;

4
source/common/src/tmsg.c
View File

@ -1813,7 +1813,7 @@ int32_t tSerializeSAlterUserReq(void *buf, int32_t bufLen, SAlterUserReq *pReq)
} }
if (tEncodeI64(&encoder, pReq->privileges) < 0) return -1; if (tEncodeI64(&encoder, pReq->privileges) < 0) return -1;
ENCODESQL(); ENCODESQL();
if (tEncodeI8(&encoder, pReq->flag) < 0) return -1; if (tEncodeU8(&encoder, pReq->flag) < 0) return -1;
tEndEncode(&encoder); tEndEncode(&encoder);
int32_t tlen = encoder.pos; int32_t tlen = encoder.pos;
@ -1854,7 +1854,7 @@ int32_t tDeserializeSAlterUserReq(void *buf, int32_t bufLen, SAlterUserReq *pReq
if (tDecodeI64(&decoder, &pReq->privileges) < 0) return -1; if (tDecodeI64(&decoder, &pReq->privileges) < 0) return -1;
DECODESQL(); DECODESQL();
if (!tDecodeIsEnd(&decoder)) { if (!tDecodeIsEnd(&decoder)) {
if (tDecodeI8(&decoder, &pReq->flag) < 0) return -1; if (tDecodeU8(&decoder, &pReq->flag) < 0) return -1;
} }
tEndDecode(&decoder); tEndDecode(&decoder);

6
source/dnode/mnode/impl/inc/mndDef.h
View File

@ -332,10 +332,10 @@ typedef struct {
int8_t sysInfo; int8_t sysInfo;
int8_t enable; int8_t enable;
union { union {
int8_t flag; uint8_t flag;
struct { struct {
int8_t createdb : 1; uint8_t createdb : 1;
int8_t reserve : 7; uint8_t reserve : 7;
}; };
}; };
int32_t acctId; int32_t acctId;

4
source/dnode/mnode/impl/src/mndUser.c
View File

@ -818,7 +818,7 @@ SSdbRaw *mndUserActionEncode(SUserObj *pUser) {
SDB_SET_INT8(pRaw, dataPos, pUser->superUser, _OVER) SDB_SET_INT8(pRaw, dataPos, pUser->superUser, _OVER)
SDB_SET_INT8(pRaw, dataPos, pUser->sysInfo, _OVER) SDB_SET_INT8(pRaw, dataPos, pUser->sysInfo, _OVER)
SDB_SET_INT8(pRaw, dataPos, pUser->enable, _OVER) SDB_SET_INT8(pRaw, dataPos, pUser->enable, _OVER)
SDB_SET_INT8(pRaw, dataPos, pUser->flag, _OVER) SDB_SET_UINT8(pRaw, dataPos, pUser->flag, _OVER)
SDB_SET_INT32(pRaw, dataPos, pUser->authVersion, _OVER) SDB_SET_INT32(pRaw, dataPos, pUser->authVersion, _OVER)
SDB_SET_INT32(pRaw, dataPos, pUser->passVersion, _OVER) SDB_SET_INT32(pRaw, dataPos, pUser->passVersion, _OVER)
SDB_SET_INT32(pRaw, dataPos, numOfReadDbs, _OVER) SDB_SET_INT32(pRaw, dataPos, numOfReadDbs, _OVER)
@ -1002,7 +1002,7 @@ static SSdbRow *mndUserActionDecode(SSdbRaw *pRaw) {
SDB_GET_INT8(pRaw, dataPos, &pUser->superUser, _OVER) SDB_GET_INT8(pRaw, dataPos, &pUser->superUser, _OVER)
SDB_GET_INT8(pRaw, dataPos, &pUser->sysInfo, _OVER) SDB_GET_INT8(pRaw, dataPos, &pUser->sysInfo, _OVER)
SDB_GET_INT8(pRaw, dataPos, &pUser->enable, _OVER) SDB_GET_INT8(pRaw, dataPos, &pUser->enable, _OVER)
SDB_GET_INT8(pRaw, dataPos, &pUser->flag, _OVER) SDB_GET_UINT8(pRaw, dataPos, &pUser->flag, _OVER)
if (pUser->superUser) pUser->createdb = 1; if (pUser->superUser) pUser->createdb = 1;
SDB_GET_INT32(pRaw, dataPos, &pUser->authVersion, _OVER) SDB_GET_INT32(pRaw, dataPos, &pUser->authVersion, _OVER)
if (sver >= 4) { if (sver >= 4) {

4
source/dnode/mnode/sdb/inc/sdb.h
View File

@ -57,6 +57,7 @@ extern "C" {
#define SDB_GET_INT32(pData, dataPos, val, pos) SDB_GET_VAL(pData, dataPos, val, pos, sdbGetRawInt32, int32_t) #define SDB_GET_INT32(pData, dataPos, val, pos) SDB_GET_VAL(pData, dataPos, val, pos, sdbGetRawInt32, int32_t)
#define SDB_GET_INT16(pData, dataPos, val, pos) SDB_GET_VAL(pData, dataPos, val, pos, sdbGetRawInt16, int16_t) #define SDB_GET_INT16(pData, dataPos, val, pos) SDB_GET_VAL(pData, dataPos, val, pos, sdbGetRawInt16, int16_t)
#define SDB_GET_INT8(pData, dataPos, val, pos) SDB_GET_VAL(pData, dataPos, val, pos, sdbGetRawInt8, int8_t) #define SDB_GET_INT8(pData, dataPos, val, pos) SDB_GET_VAL(pData, dataPos, val, pos, sdbGetRawInt8, int8_t)
#define SDB_GET_UINT8(pData, dataPos, val, pos) SDB_GET_VAL(pData, dataPos, val, pos, sdbGetRawUInt8, uint8_t)
#define SDB_GET_RESERVE(pRaw, dataPos, valLen, pos) \ #define SDB_GET_RESERVE(pRaw, dataPos, valLen, pos) \
{ \ { \
@ -76,6 +77,7 @@ extern "C" {
#define SDB_SET_INT32(pRaw, dataPos, val, pos) SDB_SET_VAL(pRaw, dataPos, val, pos, sdbSetRawInt32, int32_t) #define SDB_SET_INT32(pRaw, dataPos, val, pos) SDB_SET_VAL(pRaw, dataPos, val, pos, sdbSetRawInt32, int32_t)
#define SDB_SET_INT16(pRaw, dataPos, val, pos) SDB_SET_VAL(pRaw, dataPos, val, pos, sdbSetRawInt16, int16_t) #define SDB_SET_INT16(pRaw, dataPos, val, pos) SDB_SET_VAL(pRaw, dataPos, val, pos, sdbSetRawInt16, int16_t)
#define SDB_SET_INT8(pRaw, dataPos, val, pos) SDB_SET_VAL(pRaw, dataPos, val, pos, sdbSetRawInt8, int8_t) #define SDB_SET_INT8(pRaw, dataPos, val, pos) SDB_SET_VAL(pRaw, dataPos, val, pos, sdbSetRawInt8, int8_t)
#define SDB_SET_UINT8(pRaw, dataPos, val, pos) SDB_SET_VAL(pRaw, dataPos, val, pos, sdbSetRawUInt8, uint8_t)
#define SDB_SET_BINARY(pRaw, dataPos, val, valLen, pos) \ #define SDB_SET_BINARY(pRaw, dataPos, val, valLen, pos) \
{ \ { \
@ -388,6 +390,7 @@ void sdbGetCommitInfo(SSdb *pSdb, int64_t *index, int64_t *term, int64_t *config
SSdbRaw *sdbAllocRaw(ESdbType type, int8_t sver, int32_t dataLen); SSdbRaw *sdbAllocRaw(ESdbType type, int8_t sver, int32_t dataLen);
void sdbFreeRaw(SSdbRaw *pRaw); void sdbFreeRaw(SSdbRaw *pRaw);
int32_t sdbSetRawInt8(SSdbRaw *pRaw, int32_t dataPos, int8_t val); int32_t sdbSetRawInt8(SSdbRaw *pRaw, int32_t dataPos, int8_t val);
int32_t sdbSetRawUInt8(SSdbRaw *pRaw, int32_t dataPos, uint8_t val);
int32_t sdbSetRawInt16(SSdbRaw *pRaw, int32_t dataPos, int16_t val); int32_t sdbSetRawInt16(SSdbRaw *pRaw, int32_t dataPos, int16_t val);
int32_t sdbSetRawInt32(SSdbRaw *pRaw, int32_t dataPos, int32_t val); int32_t sdbSetRawInt32(SSdbRaw *pRaw, int32_t dataPos, int32_t val);
int32_t sdbSetRawInt64(SSdbRaw *pRaw, int32_t dataPos, int64_t val); int32_t sdbSetRawInt64(SSdbRaw *pRaw, int32_t dataPos, int64_t val);
@ -395,6 +398,7 @@ int32_t sdbSetRawBinary(SSdbRaw *pRaw, int32_t dataPos, const char *pVal, int32
int32_t sdbSetRawDataLen(SSdbRaw *pRaw, int32_t dataLen); int32_t sdbSetRawDataLen(SSdbRaw *pRaw, int32_t dataLen);
int32_t sdbSetRawStatus(SSdbRaw *pRaw, ESdbStatus status); int32_t sdbSetRawStatus(SSdbRaw *pRaw, ESdbStatus status);
int32_t sdbGetRawInt8(SSdbRaw *pRaw, int32_t dataPos, int8_t *val); int32_t sdbGetRawInt8(SSdbRaw *pRaw, int32_t dataPos, int8_t *val);
int32_t sdbGetRawUInt8(SSdbRaw *pRaw, int32_t dataPos, uint8_t *val);
int32_t sdbGetRawInt16(SSdbRaw *pRaw, int32_t dataPos, int16_t *val); int32_t sdbGetRawInt16(SSdbRaw *pRaw, int32_t dataPos, int16_t *val);
int32_t sdbGetRawInt32(SSdbRaw *pRaw, int32_t dataPos, int32_t *val); int32_t sdbGetRawInt32(SSdbRaw *pRaw, int32_t dataPos, int32_t *val);
int32_t sdbGetRawInt64(SSdbRaw *pRaw, int32_t dataPos, int64_t *val); int32_t sdbGetRawInt64(SSdbRaw *pRaw, int32_t dataPos, int64_t *val);

30
source/dnode/mnode/sdb/src/sdbRaw.c
View File

@ -67,6 +67,21 @@ int32_t sdbSetRawInt8(SSdbRaw *pRaw, int32_t dataPos, int8_t val) {
return 0; return 0;
} }
int32_t sdbSetRawUInt8(SSdbRaw *pRaw, int32_t dataPos, uint8_t val) {
if (pRaw == NULL) {
terrno = TSDB_CODE_INVALID_PTR;
return -1;
}
if (dataPos + sizeof(uint8_t) > pRaw->dataLen) {
terrno = TSDB_CODE_SDB_INVALID_DATA_LEN;
return -1;
}
*(uint8_t *)(pRaw->pData + dataPos) = val;
return 0;
}
int32_t sdbSetRawInt32(SSdbRaw *pRaw, int32_t dataPos, int32_t val) { int32_t sdbSetRawInt32(SSdbRaw *pRaw, int32_t dataPos, int32_t val) {
if (pRaw == NULL) { if (pRaw == NULL) {
terrno = TSDB_CODE_INVALID_PTR; terrno = TSDB_CODE_INVALID_PTR;
@ -174,6 +189,21 @@ int32_t sdbGetRawInt8(SSdbRaw *pRaw, int32_t dataPos, int8_t *val) {
return 0; return 0;
} }
int32_t sdbGetRawUInt8(SSdbRaw *pRaw, int32_t dataPos, uint8_t *val) {
if (pRaw == NULL) {
terrno = TSDB_CODE_INVALID_PTR;
return -1;
}
if (dataPos + sizeof(uint8_t) > pRaw->dataLen) {
terrno = TSDB_CODE_SDB_INVALID_DATA_LEN;
return -1;
}
*val = *(uint8_t *)(pRaw->pData + dataPos);
return 0;
}
int32_t sdbGetRawInt32(SSdbRaw *pRaw, int32_t dataPos, int32_t *val) { int32_t sdbGetRawInt32(SSdbRaw *pRaw, int32_t dataPos, int32_t *val) {
if (pRaw == NULL) { if (pRaw == NULL) {
terrno = TSDB_CODE_INVALID_PTR; terrno = TSDB_CODE_INVALID_PTR;

6
source/dnode/snode/src/snode.c
View File

@ -25,12 +25,6 @@
#define sndDebug(...) do { if (sndDebugFlag & DEBUG_DEBUG) { taosPrintLog("SND ", DEBUG_DEBUG, sndDebugFlag, __VA_ARGS__);}} while (0) #define sndDebug(...) do { if (sndDebugFlag & DEBUG_DEBUG) { taosPrintLog("SND ", DEBUG_DEBUG, sndDebugFlag, __VA_ARGS__);}} while (0)
// clang-format on // clang-format on
static void restoreStreamTaskId(SStreamTask *pTask, STaskId *pId) {
ASSERT(pTask->info.fillHistory);
pTask->id.taskId = pId->taskId;
pTask->id.streamId = pId->streamId;
}
int32_t sndExpandTask(SSnode *pSnode, SStreamTask *pTask, int64_t nextProcessVer) { int32_t sndExpandTask(SSnode *pSnode, SStreamTask *pTask, int64_t nextProcessVer) {
ASSERT(pTask->info.taskLevel == TASK_LEVEL__AGG && taosArrayGetSize(pTask->upstreamInfo.pList) != 0); ASSERT(pTask->info.taskLevel == TASK_LEVEL__AGG && taosArrayGetSize(pTask->upstreamInfo.pList) != 0);
int32_t code = streamTaskInit(pTask, pSnode->pMeta, &pSnode->msgCb, nextProcessVer); int32_t code = streamTaskInit(pTask, pSnode->pMeta, &pSnode->msgCb, nextProcessVer);

2
source/dnode/vnode/src/sma/smaRollup.c
View File

@ -299,7 +299,7 @@ static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat
tdRSmaTaskInit(pStreamTask->pMeta, pItem, &pStreamTask->id); tdRSmaTaskInit(pStreamTask->pMeta, pItem, &pStreamTask->id);
pStreamTask->status.pSM = streamCreateStateMachine(pStreamTask); pStreamTask->status.pSM = streamCreateStateMachine(pStreamTask);
pStreamTask->chkInfo.pActiveInfo = streamTaskCreateActiveChkptInfo(); pStreamTask->chkInfo.pActiveInfo = streamTaskCreateActiveChkptInfo();
pStreamState = streamStateOpen(taskInfDir, pStreamTask, true, -1, -1); pStreamState = streamStateOpen(taskInfDir, pStreamTask, pStreamTask->id.streamId, pStreamTask->id.taskId, true, -1, -1);
if (!pStreamState) { if (!pStreamState) {
terrno = TSDB_CODE_RSMA_STREAM_STATE_OPEN; terrno = TSDB_CODE_RSMA_STREAM_STATE_OPEN;
return TSDB_CODE_FAILED; return TSDB_CODE_FAILED;

7
source/dnode/vnode/src/tq/tqUtil.c
View File

@ -605,14 +605,15 @@ int32_t tqGetStreamExecInfo(SVnode* pVnode, int64_t streamId, int64_t* pDelay, b
numOfTasks = taosArrayGetSize(pMeta->pTaskList); numOfTasks = taosArrayGetSize(pMeta->pTaskList);
for (int32_t i = 0; i < numOfTasks; ++i) { for (int32_t i = 0; i < numOfTasks; ++i) {
STaskId* pId = taosArrayGet(pMeta->pTaskList, i); SStreamTaskId* pId = taosArrayGet(pMeta->pTaskList, i);
if (pId->streamId != streamId) { if (pId->streamId != streamId) {
continue; continue;
} }
SStreamTask** ppTask = taosHashGet(pMeta->pTasksMap, pId, sizeof(*pId)); STaskId id = {.streamId = pId->streamId, .taskId = pId->taskId};
SStreamTask** ppTask = taosHashGet(pMeta->pTasksMap, &id, sizeof(id));
if (ppTask == NULL) { if (ppTask == NULL) {
tqError("vgId:%d failed to acquire task:0x%" PRIx64 " in retrieving progress", pMeta->vgId, pId->taskId); tqError("vgId:%d failed to acquire task:0x%x in retrieving progress", pMeta->vgId, pId->taskId);
continue; continue;
} }

33
source/dnode/vnode/src/tqCommon/tqCommon.c
View File

@ -30,37 +30,26 @@ typedef struct SMStreamCheckpointReadyRspMsg {
static int32_t doProcessDummyRspMsg(SStreamMeta* pMeta, SRpcMsg* pMsg); static int32_t doProcessDummyRspMsg(SStreamMeta* pMeta, SRpcMsg* pMsg);
static STaskId replaceStreamTaskId(SStreamTask* pTask) {
ASSERT(pTask->info.fillHistory);
STaskId id = {.streamId = pTask->id.streamId, .taskId = pTask->id.taskId};
pTask->id.streamId = pTask->streamTaskId.streamId;
pTask->id.taskId = pTask->streamTaskId.taskId;
return id;
}
static void restoreStreamTaskId(SStreamTask* pTask, STaskId* pId) {
ASSERT(pTask->info.fillHistory);
pTask->id.taskId = pId->taskId;
pTask->id.streamId = pId->streamId;
}
int32_t tqExpandStreamTask(SStreamTask* pTask) { int32_t tqExpandStreamTask(SStreamTask* pTask) {
SStreamMeta* pMeta = pTask->pMeta; SStreamMeta* pMeta = pTask->pMeta;
int32_t vgId = pMeta->vgId; int32_t vgId = pMeta->vgId;
STaskId taskId = {0};
int64_t st = taosGetTimestampMs(); int64_t st = taosGetTimestampMs();
int64_t streamId = 0;
int32_t taskId = 0;
tqDebug("s-task:%s vgId:%d start to expand stream task", pTask->id.idStr, vgId); tqDebug("s-task:%s vgId:%d start to expand stream task", pTask->id.idStr, vgId);
if (pTask->info.fillHistory) { if (pTask->info.fillHistory) {
taskId = replaceStreamTaskId(pTask); streamId = pTask->streamTaskId.streamId;
taskId = pTask->streamTaskId.taskId;
} else {
streamId = pTask->id.streamId;
taskId = pTask->id.taskId;
} }
// sink task does not need the pState // sink task does not need the pState
if (pTask->info.taskLevel != TASK_LEVEL__SINK) { if (pTask->info.taskLevel != TASK_LEVEL__SINK) {
pTask->pState = streamStateOpen(pMeta->path, pTask, false, -1, -1); pTask->pState = streamStateOpen(pMeta->path, pTask, false, streamId, taskId, -1, -1);
if (pTask->pState == NULL) { if (pTask->pState == NULL) {
tqError("s-task:%s (vgId:%d) failed to open state for task, expand task failed", pTask->id.idStr, vgId); tqError("s-task:%s (vgId:%d) failed to open state for task, expand task failed", pTask->id.idStr, vgId);
return -1; return -1;
@ -69,10 +58,6 @@ int32_t tqExpandStreamTask(SStreamTask* pTask) {
} }
} }
if (pTask->info.fillHistory) {
restoreStreamTaskId(pTask, &taskId);
}
SReadHandle handle = { SReadHandle handle = {
.checkpointId = pTask->chkInfo.checkpointId, .checkpointId = pTask->chkInfo.checkpointId,
.pStateBackend = pTask->pState, .pStateBackend = pTask->pState,
@ -635,8 +620,8 @@ int32_t tqStreamTaskProcessDropReq(SStreamMeta* pMeta, char* msg, int32_t msgLen
// drop the related fill-history task firstly // drop the related fill-history task firstly
if (hTaskId.taskId != 0 && hTaskId.streamId != 0) { if (hTaskId.taskId != 0 && hTaskId.streamId != 0) {
streamMetaUnregisterTask(pMeta, hTaskId.streamId, hTaskId.taskId);
tqDebug("s-task:0x%x vgId:%d drop rel fill-history task:0x%x firstly", pReq->taskId, vgId, (int32_t)hTaskId.taskId); tqDebug("s-task:0x%x vgId:%d drop rel fill-history task:0x%x firstly", pReq->taskId, vgId, (int32_t)hTaskId.taskId);
streamMetaUnregisterTask(pMeta, hTaskId.streamId, hTaskId.taskId);
} }
// drop the stream task now // drop the stream task now

12
source/dnode/vnode/src/vnd/vnodeQuery.c
View File

@ -120,7 +120,8 @@ int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {
memcpy(metaRsp.pSchemas + schema.nCols, schemaTag.pSchema, sizeof(SSchema) * schemaTag.nCols); memcpy(metaRsp.pSchemas + schema.nCols, schemaTag.pSchema, sizeof(SSchema) * schemaTag.nCols);
} }
if (metaRsp.pSchemaExt) { if (metaRsp.pSchemaExt) {
code = fillTableColCmpr(&mer1, metaRsp.pSchemaExt, metaRsp.numOfColumns); SMetaReader *pReader = mer1.me.type == TSDB_CHILD_TABLE ? &mer2 : &mer1;
code = fillTableColCmpr(pReader, metaRsp.pSchemaExt, metaRsp.numOfColumns);
if (code < 0) { if (code < 0) {
code = TSDB_CODE_INVALID_MSG; code = TSDB_CODE_INVALID_MSG;
goto _exit; goto _exit;
@ -254,15 +255,18 @@ int vnodeGetTableCfg(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {
memcpy(cfgRsp.pSchemas + schema.nCols, schemaTag.pSchema, sizeof(SSchema) * schemaTag.nCols); memcpy(cfgRsp.pSchemas + schema.nCols, schemaTag.pSchema, sizeof(SSchema) * schemaTag.nCols);
} }
if (useCompress(cfgRsp.tableType)) { // if (useCompress(cfgRsp.tableType)) {
SColCmprWrapper *pColCmpr = &mer1.me.colCmpr;
SMetaReader *pReader = mer1.me.type == TSDB_CHILD_TABLE ? &mer2 : &mer1;
SColCmprWrapper *pColCmpr = &pReader->me.colCmpr;
for (int32_t i = 0; i < cfgRsp.numOfColumns; i++) { for (int32_t i = 0; i < cfgRsp.numOfColumns; i++) {
SColCmpr *pCmpr = &pColCmpr->pColCmpr[i]; SColCmpr *pCmpr = &pColCmpr->pColCmpr[i];
SSchemaExt *pSchExt = cfgRsp.pSchemaExt + i; SSchemaExt *pSchExt = cfgRsp.pSchemaExt + i;
pSchExt->colId = pCmpr->id; pSchExt->colId = pCmpr->id;
pSchExt->compress = pCmpr->alg; pSchExt->compress = pCmpr->alg;
} }
} //}
// encode and send response // encode and send response
rspLen = tSerializeSTableCfgRsp(NULL, 0, &cfgRsp); rspLen = tSerializeSTableCfgRsp(NULL, 0, &cfgRsp);

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

@ -6605,8 +6605,8 @@ static int32_t buildCreateDbReq(STranslateContext* pCxt, SCreateDatabaseStmt* pS
} }
static int32_t checkRangeOption(STranslateContext* pCxt, int32_t code, const char* pName, int64_t val, int64_t minVal, static int32_t checkRangeOption(STranslateContext* pCxt, int32_t code, const char* pName, int64_t val, int64_t minVal,
int64_t maxVal) { int64_t maxVal, bool skipUndef) {
if (val >= 0 && (val < minVal || val > maxVal)) { if (skipUndef ? ((val >= 0) && (val < minVal || val > maxVal)) : (val < minVal || val > maxVal)) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, code, return generateSyntaxErrMsgExt(&pCxt->msgBuf, code,
"Invalid option %s: %" PRId64 ", valid range: [%" PRId64 ", %" PRId64 "]", pName, "Invalid option %s: %" PRId64 ", valid range: [%" PRId64 ", %" PRId64 "]", pName,
val, minVal, maxVal); val, minVal, maxVal);
@ -6616,12 +6616,12 @@ static int32_t checkRangeOption(STranslateContext* pCxt, int32_t code, const cha
static int32_t checkDbRangeOption(STranslateContext* pCxt, const char* pName, int32_t val, int32_t minVal, static int32_t checkDbRangeOption(STranslateContext* pCxt, const char* pName, int32_t val, int32_t minVal,
int32_t maxVal) { int32_t maxVal) {
return checkRangeOption(pCxt, TSDB_CODE_PAR_INVALID_DB_OPTION, pName, val, minVal, maxVal); return checkRangeOption(pCxt, TSDB_CODE_PAR_INVALID_DB_OPTION, pName, val, minVal, maxVal, true);
} }
static int32_t checkTableRangeOption(STranslateContext* pCxt, const char* pName, int64_t val, int64_t minVal, static int32_t checkTableRangeOption(STranslateContext* pCxt, const char* pName, int64_t val, int64_t minVal,
int64_t maxVal) { int64_t maxVal) {
return checkRangeOption(pCxt, TSDB_CODE_PAR_INVALID_TABLE_OPTION, pName, val, minVal, maxVal); return checkRangeOption(pCxt, TSDB_CODE_PAR_INVALID_TABLE_OPTION, pName, val, minVal, maxVal, true);
} }
static int32_t checkDbS3KeepLocalOption(STranslateContext* pCxt, SDatabaseOptions* pOptions) { static int32_t checkDbS3KeepLocalOption(STranslateContext* pCxt, SDatabaseOptions* pOptions) {
@ -8485,7 +8485,7 @@ static int32_t translateUseDatabase(STranslateContext* pCxt, SUseDatabaseStmt* p
static int32_t translateCreateUser(STranslateContext* pCxt, SCreateUserStmt* pStmt) { static int32_t translateCreateUser(STranslateContext* pCxt, SCreateUserStmt* pStmt) {
int32_t code = 0; int32_t code = 0;
SCreateUserReq createReq = {0}; SCreateUserReq createReq = {0};
if ((code = checkRangeOption(pCxt, TSDB_CODE_INVALID_OPTION, "sysinfo", pStmt->sysinfo, 0, 1))) { if ((code = checkRangeOption(pCxt, TSDB_CODE_INVALID_OPTION, "sysinfo", pStmt->sysinfo, 0, 1, false))) {
return code; return code;
} }
strcpy(createReq.user, pStmt->userName); strcpy(createReq.user, pStmt->userName);
@ -8509,13 +8509,13 @@ static int32_t checkAlterUser(STranslateContext* pCxt, SAlterUserStmt* pStmt) {
int32_t code = 0; int32_t code = 0;
switch (pStmt->alterType) { switch (pStmt->alterType) {
case TSDB_ALTER_USER_ENABLE: case TSDB_ALTER_USER_ENABLE:
code = checkRangeOption(pCxt, TSDB_CODE_INVALID_OPTION, "enable", pStmt->enable, 0, 1); code = checkRangeOption(pCxt, TSDB_CODE_INVALID_OPTION, "enable", pStmt->enable, 0, 1, false);
break; break;
case TSDB_ALTER_USER_SYSINFO: case TSDB_ALTER_USER_SYSINFO:
code = checkRangeOption(pCxt, TSDB_CODE_INVALID_OPTION, "sysinfo", pStmt->sysinfo, 0, 1); code = checkRangeOption(pCxt, TSDB_CODE_INVALID_OPTION, "sysinfo", pStmt->sysinfo, 0, 1, false);
break; break;
case TSDB_ALTER_USER_CREATEDB: case TSDB_ALTER_USER_CREATEDB:
code = checkRangeOption(pCxt, TSDB_CODE_INVALID_OPTION, "createdb", pStmt->createdb, 0, 1); code = checkRangeOption(pCxt, TSDB_CODE_INVALID_OPTION, "createdb", pStmt->createdb, 0, 1, false);
break; break;
} }
return code; return code;

32
source/libs/stream/src/streamMeta.c
View File

@ -608,6 +608,7 @@ int32_t streamMetaRegisterTask(SStreamMeta* pMeta, int64_t ver, SStreamTask* pTa
} }
taosArrayPush(pMeta->pTaskList, &pTask->id); taosArrayPush(pMeta->pTaskList, &pTask->id);
taosHashPut(pMeta->pTasksMap, &id, sizeof(id), &pTask, POINTER_BYTES);
if (streamMetaSaveTask(pMeta, pTask) < 0) { if (streamMetaSaveTask(pMeta, pTask) < 0) {
return -1; return -1;
@ -617,7 +618,6 @@ int32_t streamMetaRegisterTask(SStreamMeta* pMeta, int64_t ver, SStreamTask* pTa
return -1; return -1;
} }
taosHashPut(pMeta->pTasksMap, &id, sizeof(id), &pTask, POINTER_BYTES);
if (pTask->info.fillHistory == 0) { if (pTask->info.fillHistory == 0) {
atomic_add_fetch_32(&pMeta->numOfStreamTasks, 1); atomic_add_fetch_32(&pMeta->numOfStreamTasks, 1);
} }
@ -672,14 +672,17 @@ void streamMetaReleaseTask(SStreamMeta* UNUSED_PARAM(pMeta), SStreamTask* pTask)
} }
} }
static void doRemoveIdFromList(SStreamMeta* pMeta, int32_t num, SStreamTaskId* id) { static void doRemoveIdFromList(SArray* pTaskList, int32_t num, SStreamTaskId* id) {
bool remove = false;
for (int32_t i = 0; i < num; ++i) { for (int32_t i = 0; i < num; ++i) {
SStreamTaskId* pTaskId = taosArrayGet(pMeta->pTaskList, i); SStreamTaskId* pTaskId = taosArrayGet(pTaskList, i);
if (pTaskId->streamId == id->streamId && pTaskId->taskId == id->taskId) { if (pTaskId->streamId == id->streamId && pTaskId->taskId == id->taskId) {
taosArrayRemove(pMeta->pTaskList, i); taosArrayRemove(pTaskList, i);
remove = true;
break; break;
} }
} }
ASSERT(remove);
} }
static int32_t streamTaskSendTransSuccessMsg(SStreamTask* pTask, void* param) { static int32_t streamTaskSendTransSuccessMsg(SStreamTask* pTask, void* param) {
@ -715,7 +718,7 @@ int32_t streamMetaUnregisterTask(SStreamMeta* pMeta, int64_t streamId, int32_t t
} }
streamMetaWUnLock(pMeta); streamMetaWUnLock(pMeta);
stDebug("s-task:0x%x set task status:dropping and start to unregister it", taskId); stDebug("s-task:0x%x vgId:%d set task status:dropping and start to unregister it", taskId, pMeta->vgId);
while (1) { while (1) {
streamMetaRLock(pMeta); streamMetaRLock(pMeta);
@ -742,18 +745,19 @@ int32_t streamMetaUnregisterTask(SStreamMeta* pMeta, int64_t streamId, int32_t t
ppTask = (SStreamTask**)taosHashGet(pMeta->pTasksMap, &id, sizeof(id)); ppTask = (SStreamTask**)taosHashGet(pMeta->pTasksMap, &id, sizeof(id));
if (ppTask) { if (ppTask) {
pTask = *ppTask; pTask = *ppTask;
// it is an fill-history task, remove the related stream task's id that points to it // it is an fill-history task, remove the related stream task's id that points to it
if (pTask->info.fillHistory == 0) {
atomic_sub_fetch_32(&pMeta->numOfStreamTasks, 1); atomic_sub_fetch_32(&pMeta->numOfStreamTasks, 1);
}
taosHashRemove(pMeta->pTasksMap, &id, sizeof(id)); taosHashRemove(pMeta->pTasksMap, &id, sizeof(id));
doRemoveIdFromList(pMeta, (int32_t)taosArrayGetSize(pMeta->pTaskList), &pTask->id); doRemoveIdFromList(pMeta->pTaskList, (int32_t)taosArrayGetSize(pMeta->pTaskList), &pTask->id);
streamMetaRemoveTask(pMeta, &id); streamMetaRemoveTask(pMeta, &id);
ASSERT(taosHashGetSize(pMeta->pTasksMap) == taosArrayGetSize(pMeta->pTaskList));
streamMetaWUnLock(pMeta); streamMetaWUnLock(pMeta);
ASSERT(pTask->status.timerActive == 0); ASSERT(pTask->status.timerActive == 0);
if (pTask->info.delaySchedParam != 0 && pTask->info.fillHistory == 0) { if (pTask->info.delaySchedParam != 0 && pTask->info.fillHistory == 0) {
stDebug("s-task:%s stop schedTimer, and (before) desc ref:%d", pTask->id.idStr, pTask->refCnt); stDebug("s-task:%s stop schedTimer, and (before) desc ref:%d", pTask->id.idStr, pTask->refCnt);
taosTmrStop(pTask->schedInfo.pDelayTimer); taosTmrStop(pTask->schedInfo.pDelayTimer);
@ -1007,9 +1011,10 @@ static int32_t metaHeartbeatToMnodeImpl(SStreamMeta* pMeta) {
hbMsg.pUpdateNodes = taosArrayInit(numOfTasks, sizeof(int32_t)); hbMsg.pUpdateNodes = taosArrayInit(numOfTasks, sizeof(int32_t));
for (int32_t i = 0; i < numOfTasks; ++i) { for (int32_t i = 0; i < numOfTasks; ++i) {
STaskId* pId = taosArrayGet(pMeta->pTaskList, i); SStreamTaskId* pId = taosArrayGet(pMeta->pTaskList, i);
SStreamTask** pTask = taosHashGet(pMeta->pTasksMap, pId, sizeof(*pId)); STaskId id = {.streamId = pId->streamId, .taskId = pId->taskId};
SStreamTask** pTask = taosHashGet(pMeta->pTasksMap, &id, sizeof(id));
if (pTask == NULL) { if (pTask == NULL) {
continue; continue;
} }
@ -1020,7 +1025,7 @@ static int32_t metaHeartbeatToMnodeImpl(SStreamMeta* pMeta) {
} }
STaskStatusEntry entry = { STaskStatusEntry entry = {
.id = *pId, .id = id,
.status = streamTaskGetStatus(*pTask)->state, .status = streamTaskGetStatus(*pTask)->state,
.nodeId = hbMsg.vgId, .nodeId = hbMsg.vgId,
.stage = pMeta->stage, .stage = pMeta->stage,
@ -1508,8 +1513,9 @@ int32_t streamMetaStopAllTasks(SStreamMeta* pMeta) {
bool streamMetaAllTasksReady(const SStreamMeta* pMeta) { bool streamMetaAllTasksReady(const SStreamMeta* pMeta) {
int32_t num = taosArrayGetSize(pMeta->pTaskList); int32_t num = taosArrayGetSize(pMeta->pTaskList);
for (int32_t i = 0; i < num; ++i) { for (int32_t i = 0; i < num; ++i) {
STaskId* pTaskId = taosArrayGet(pMeta->pTaskList, i); SStreamTaskId* pId = taosArrayGet(pMeta->pTaskList, i);
SStreamTask** ppTask = taosHashGet(pMeta->pTasksMap, pTaskId, sizeof(*pTaskId)); STaskId id = {.streamId = pId->streamId, .taskId = pId->taskId};
SStreamTask** ppTask = taosHashGet(pMeta->pTasksMap, &id, sizeof(id));
if (ppTask == NULL) { if (ppTask == NULL) {
continue; continue;
} }

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

@ -516,24 +516,25 @@ SStreamStateCur* countWinStateSeekKeyPrev(SStreamFileState* pFileState, const SS
return pBuffCur; return pBuffCur;
} }
winCount = *((COUNT_TYPE*) ((char*)pVal + (resSize - sizeof(COUNT_TYPE)))); winCount = *((COUNT_TYPE*) ((char*)pVal + (resSize - sizeof(COUNT_TYPE))));
taosMemoryFreeClear(pVal);
streamStateFreeCur(pBuffCur);
if (sessionRangeKeyCmpr(pWinKey, &key) != 0 && winCount == count) { if (sessionRangeKeyCmpr(pWinKey, &key) != 0 && winCount == count) {
streamStateFreeCur(pCur); streamStateCurNext(pFileStore, pCur);
return pBuffCur; return pCur;
} }
streamStateCurPrev(pFileStore, pCur); streamStateCurPrev(pFileStore, pCur);
while (1) { while (1) {
code = streamStateSessionGetKVByCur_rocksdb(pCur, &key, &pVal, &len); code = streamStateSessionGetKVByCur_rocksdb(pCur, &key, &pVal, &len);
if (code == TSDB_CODE_FAILED) { if (code == TSDB_CODE_FAILED) {
streamStateCurNext(pFileStore, pCur); streamStateCurNext(pFileStore, pCur);
streamStateFreeCur(pBuffCur);
return pCur; return pCur;
} }
winCount = *((COUNT_TYPE*) ((char*)pVal + (resSize - sizeof(COUNT_TYPE)))); winCount = *((COUNT_TYPE*) ((char*)pVal + (resSize - sizeof(COUNT_TYPE))));
taosMemoryFreeClear(pVal);
if (sessionRangeKeyCmpr(pWinKey, &key) == 0 || winCount < count) { if (sessionRangeKeyCmpr(pWinKey, &key) == 0 || winCount < count) {
streamStateCurPrev(pFileStore, pCur); streamStateCurPrev(pFileStore, pCur);
} else { } else {
streamStateCurNext(pFileStore, pCur); streamStateCurNext(pFileStore, pCur);
streamStateFreeCur(pBuffCur);
return pCur; return pCur;
} }
} }
@ -568,7 +569,7 @@ int32_t sessionWinStateGetKVByCur(SStreamStateCur* pCur, SSessionKey* pKey, void
void* pData = NULL; void* pData = NULL;
code = streamStateSessionGetKVByCur_rocksdb(pCur, pKey, &pData, pVLen); code = streamStateSessionGetKVByCur_rocksdb(pCur, pKey, &pData, pVLen);
if (taosArrayGetSize(pWinStates) > 0 && if (taosArrayGetSize(pWinStates) > 0 &&
(code == TSDB_CODE_FAILED || sessionStateKeyCompare(pKey, pWinStates, 0) >= 0)) { (code == TSDB_CODE_FAILED || sessionStateRangeKeyCompare(pKey, pWinStates, 0) >= 0)) {
transformCursor(pCur->pStreamFileState, pCur); transformCursor(pCur->pStreamFileState, pCur);
SRowBuffPos* pPos = taosArrayGetP(pWinStates, pCur->buffIndex); SRowBuffPos* pPos = taosArrayGetP(pWinStates, pCur->buffIndex);
if (pVal) { if (pVal) {
@ -590,6 +591,7 @@ int32_t sessionWinStateGetKVByCur(SStreamStateCur* pCur, SSessionKey* pKey, void
} }
int32_t sessionWinStateMoveToNext(SStreamStateCur* pCur) { int32_t sessionWinStateMoveToNext(SStreamStateCur* pCur) {
qTrace("move cursor to next");
if (pCur && pCur->buffIndex >= 0) { if (pCur && pCur->buffIndex >= 0) {
pCur->buffIndex++; pCur->buffIndex++;
} else { } else {

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

@ -98,7 +98,8 @@ int stateKeyCmpr(const void* pKey1, int kLen1, const void* pKey2, int kLen2) {
return winKeyCmprImpl(&pWin1->key, &pWin2->key); return winKeyCmprImpl(&pWin1->key, &pWin2->key);
} }
SStreamState* streamStateOpen(const char* path, void* pTask, bool specPath, int32_t szPage, int32_t pages) { SStreamState* streamStateOpen(const char* path, void* pTask, int64_t streamId, int32_t taskId, bool specPath,
int32_t szPage, int32_t pages) {
SStreamState* pState = taosMemoryCalloc(1, sizeof(SStreamState)); SStreamState* pState = taosMemoryCalloc(1, sizeof(SStreamState));
stDebug("open stream state %p, %s", pState, path); stDebug("open stream state %p, %s", pState, path);
if (pState == NULL) { if (pState == NULL) {
@ -114,8 +115,8 @@ SStreamState* streamStateOpen(const char* path, void* pTask, bool specPath, int3
} }
SStreamTask* pStreamTask = pTask; SStreamTask* pStreamTask = pTask;
pState->taskId = pStreamTask->id.taskId; pState->streamId = streamId;
pState->streamId = pStreamTask->id.streamId; pState->taskId = taskId;
sprintf(pState->pTdbState->idstr, "0x%" PRIx64 "-0x%x", pState->streamId, pState->taskId); sprintf(pState->pTdbState->idstr, "0x%" PRIx64 "-0x%x", pState->streamId, pState->taskId);
streamTaskSetDb(pStreamTask->pMeta, pTask, pState->pTdbState->idstr); streamTaskSetDb(pStreamTask->pMeta, pTask, pState->pTdbState->idstr);
@ -654,6 +655,7 @@ int32_t streamStateCurNext(SStreamState* pState, SStreamStateCur* pCur) {
int32_t streamStateCurPrev(SStreamState* pState, SStreamStateCur* pCur) { int32_t streamStateCurPrev(SStreamState* pState, SStreamStateCur* pCur) {
#ifdef USE_ROCKSDB #ifdef USE_ROCKSDB
qTrace("move cursor to next");
return streamStateCurPrev_rocksdb(pCur); return streamStateCurPrev_rocksdb(pCur);
#else #else
if (!pCur) { if (!pCur) {

2
source/libs/stream/test/backendTest.cpp
View File

@ -46,7 +46,7 @@ SStreamState *stateCreate(const char *path) {
SStreamMeta *pMeta = streamMetaOpen((path), NULL, NULL, NULL, 0, 0, NULL); SStreamMeta *pMeta = streamMetaOpen((path), NULL, NULL, NULL, 0, 0, NULL);
pTask->pMeta = pMeta; pTask->pMeta = pMeta;
SStreamState *p = streamStateOpen((char *)path, pTask, true, 32, 32 * 1024); SStreamState *p = streamStateOpen((char *)path, pTask, 0, 0, true, 32, 32 * 1024);
ASSERT(p != NULL); ASSERT(p != NULL);
return p; return p;
} }

2
source/libs/transport/src/transSvr.c
View File

@ -363,7 +363,7 @@ static bool uvHandleReq(SSvrConn* pConn) {
memcpy(pConn->user, pHead->user, strlen(pHead->user)); memcpy(pConn->user, pHead->user, strlen(pHead->user));
int8_t forbiddenIp = 0; int8_t forbiddenIp = 0;
if (pThrd->enableIpWhiteList) { if (pThrd->enableIpWhiteList && tsEnableWhiteList) {
forbiddenIp = !uvWhiteListCheckConn(pThrd->pWhiteList, pConn) ? 1 : 0; forbiddenIp = !uvWhiteListCheckConn(pThrd->pWhiteList, pConn) ? 1 : 0;
if (forbiddenIp == 0) { if (forbiddenIp == 0) {
uvWhiteListSetConnVer(pThrd->pWhiteList, pConn); uvWhiteListSetConnVer(pThrd->pWhiteList, pConn);

20
source/util/src/terror.c
View File

@ -21,11 +21,6 @@
#define TAOS_ERROR_C #define TAOS_ERROR_C
typedef struct {
int32_t val;
const char* str;
} STaosError;
static threadlocal int32_t tsErrno; static threadlocal int32_t tsErrno;
static threadlocal char tsErrMsgDetail[ERR_MSG_LEN] = {0}; static threadlocal char tsErrMsgDetail[ERR_MSG_LEN] = {0};
static threadlocal char tsErrMsgReturn[ERR_MSG_LEN] = {0}; static threadlocal char tsErrMsgReturn[ERR_MSG_LEN] = {0};
@ -35,7 +30,7 @@ char* taosGetErrMsg() { return tsErrMsgDetail; }
char* taosGetErrMsgReturn() { return tsErrMsgReturn; } char* taosGetErrMsgReturn() { return tsErrMsgReturn; }
#ifdef TAOS_ERROR_C #ifdef TAOS_ERROR_C
#define TAOS_DEFINE_ERROR(name, msg) {.val = (name), .str = (msg)}, #define TAOS_DEFINE_ERROR(name, msg) {.val = (name), .str = (msg), .macro = #name},
#else #else
#define TAOS_DEFINE_ERROR(name, mod, code, msg) static const int32_t name = TAOS_DEF_ERROR_CODE(mod, code); #define TAOS_DEFINE_ERROR(name, mod, code, msg) static const int32_t name = TAOS_DEF_ERROR_CODE(mod, code);
#endif #endif
@ -46,7 +41,7 @@ char* taosGetErrMsgReturn() { return tsErrMsgReturn; }
#ifdef TAOS_ERROR_C #ifdef TAOS_ERROR_C
STaosError errors[] = { STaosError errors[] = {
{.val = 0, .str = "success"}, TAOS_DEFINE_ERROR(TSDB_CODE_SUCCESS, "success")
#endif #endif
// rpc // rpc
@ -250,15 +245,14 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_DB, "Invalid database name
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_DATABASES, "Too many databases for account") TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_DATABASES, "Too many databases for account")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_DB_IN_DROPPING, "Database in dropping status") TAOS_DEFINE_ERROR(TSDB_CODE_MND_DB_IN_DROPPING, "Database in dropping status")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_DB_NOT_EXIST, "Database not exist") TAOS_DEFINE_ERROR(TSDB_CODE_MND_DB_NOT_EXIST, "Database not exist")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_DB_RETENTION_PERIOD_ZERO, "WAL retention period is zero")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_DB_ACCT, "Invalid database account") TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_DB_ACCT, "Invalid database account")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_DB_OPTION_UNCHANGED, "Database options not changed") TAOS_DEFINE_ERROR(TSDB_CODE_MND_DB_OPTION_UNCHANGED, "Database options not changed")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_DB_INDEX_NOT_EXIST, "Index not exist") TAOS_DEFINE_ERROR(TSDB_CODE_MND_DB_INDEX_NOT_EXIST, "Index not exist")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_SYS_TABLENAME, "Invalid system table name") TAOS_DEFINE_ERROR(TSDB_CODE_MND_DB_RETENTION_PERIOD_ZERO, "WAL retention period is zero")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_DB_IN_CREATING, "Database in creating status")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_ENCRYPT_NOT_ALLOW_CHANGE, "Encryption is not allowed to be changed after database is created")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INCONSIST_ENCRYPT_KEY, "Inconsistent encryption key")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_ENCRYPT_KEY, "The cluster has not been set properly for database encryption") TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_ENCRYPT_KEY, "The cluster has not been set properly for database encryption")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_DB_IN_CREATING, "Database in creating status")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_SYS_TABLENAME, "Invalid system table name")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_ENCRYPT_NOT_ALLOW_CHANGE, "Encryption is not allowed to be changed after database is created")
// mnode-node // mnode-node
TAOS_DEFINE_ERROR(TSDB_CODE_MND_MNODE_ALREADY_EXIST, "Mnode already exists") TAOS_DEFINE_ERROR(TSDB_CODE_MND_MNODE_ALREADY_EXIST, "Mnode already exists")
@ -837,3 +831,5 @@ const char* tstrerror(int32_t err) {
} }
const char* terrstr() { return tstrerror(terrno); } const char* terrstr() { return tstrerror(terrno); }
int32_t taosGetErrSize() { return sizeof(errors)/sizeof(errors[0]); }

8
source/util/test/CMakeLists.txt
View File

@ -124,3 +124,11 @@ add_test(
# NAME decompressTest # NAME decompressTest
# COMMAND decompressTest # COMMAND decompressTest
#) #)
# terrorTest
add_executable(terrorTest "terrorTest.cpp")
target_link_libraries(terrorTest os util common gtest_main)
add_test(
NAME terrorTest
COMMAND terrorTest
)

15
source/util/test/terrorTest.cpp Normal file
View File

@ -0,0 +1,15 @@
#include <gtest/gtest.h>
#include <cassert>
#include <iostream>
#include "taoserror.h"
using namespace std;
TEST(TAOS_ERROR_TEST, terror_test) {
int32_t errSize = taosGetErrSize();
for (int32_t i = 0; i < errSize; ++i) {
STaosError *pInfo = &errors[i];
std::cout << i + 1 << " " << pInfo->macro << " " << pInfo->val << std::endl;
}
}

4
tests/script/tsim/user/basic.sim
View File

@ -217,14 +217,18 @@ endi
sql_error CREATE USER u100 PASS 'taosdata' SYSINFO -1; sql_error CREATE USER u100 PASS 'taosdata' SYSINFO -1;
sql_error CREATE USER u101 PASS 'taosdata' SYSINFO 2; sql_error CREATE USER u101 PASS 'taosdata' SYSINFO 2;
sql_error CREATE USER u102 PASS 'taosdata' SYSINFO 20000; sql_error CREATE USER u102 PASS 'taosdata' SYSINFO 20000;
sql_error CREATE USER u103 PASS 'taosdata' SYSINFO 1000;
sql_error ALTER USER u1 enable -1 sql_error ALTER USER u1 enable -1
sql_error ALTER USER u1 enable 2 sql_error ALTER USER u1 enable 2
sql_error ALTER USER u1 enable 1000
sql_error ALTER USER u1 enable 10000 sql_error ALTER USER u1 enable 10000
sql_error ALTER USER u1 sysinfo -1 sql_error ALTER USER u1 sysinfo -1
sql_error ALTER USER u1 sysinfo 2 sql_error ALTER USER u1 sysinfo 2
sql_error ALTER USER u1 sysinfo 1000
sql_error ALTER USER u1 sysinfo -20000 sql_error ALTER USER u1 sysinfo -20000
sql_error ALTER USER u1 createdb -1 sql_error ALTER USER u1 createdb -1
sql_error ALTER USER u1 createdb 3 sql_error ALTER USER u1 createdb 3
sql_error ALTER USER u1 createdb 1000
sql_error ALTER USER u1 createdb 100000 sql_error ALTER USER u1 createdb 100000
system sh/exec.sh -n dnode1 -s stop -x SIGINT system sh/exec.sh -n dnode1 -s stop -x SIGINT

2
tests/system-test/0-others/user_control.py
View File

@ -514,7 +514,7 @@ class TDTestCase:
def test_alter_user(self): def test_alter_user(self):
options = ["enable", "sysinfo", "createdb"] options = ["enable", "sysinfo", "createdb"]
optionErrVals = [-10000, -128, -1, 2, 127, 10000] optionErrVals = [-10000, -128, -1, 2, 127, 1000, 10000]
for optionErrVal in optionErrVals: for optionErrVal in optionErrVals:
tdSql.error("create user user_alter pass 'taosdata' sysinfo %d" % optionErrVal) tdSql.error("create user user_alter pass 'taosdata' sysinfo %d" % optionErrVal)
tdSql.execute("create user user_alter pass 'taosdata'") tdSql.execute("create user user_alter pass 'taosdata'")