Skip to content

T
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
提交 9025a5d7 编写于 作者: H Haojun Liao
浏览文件
操作

Merge branch '3.0' into feature/3.0_liaohj

上级 0bdabe6a 4e392c9f
展开全部 隐藏空白更改
内联 并排
Showing with 7627 addition and 5150 deletion
cmake/cmake.define
浏览文件 @ 9025a5d7
...@@ -14,6 +14,25 @@ MESSAGE(STATUS "Project binary files output path: " ${PROJECT_BINARY_DIR}) ...@@ -14,6 +14,25 @@ MESSAGE(STATUS "Project binary files output path: " ${PROJECT_BINARY_DIR})
MESSAGE(STATUS "Project executable files output path: " ${EXECUTABLE_OUTPUT_PATH}) MESSAGE(STATUS "Project executable files output path: " ${EXECUTABLE_OUTPUT_PATH})
MESSAGE(STATUS "Project library files output path: " ${LIBRARY_OUTPUT_PATH}) MESSAGE(STATUS "Project library files output path: " ${LIBRARY_OUTPUT_PATH})
find_package(Git QUIET)
if(GIT_FOUND AND EXISTS "${PROJECT_SOURCE_DIR}/.git")
# Update submodules as needed
option(GIT_SUBMODULE "Check submodules during build" ON)
if(GIT_SUBMODULE)
message(STATUS "Submodule update")
execute_process(COMMAND ${GIT_EXECUTABLE} submodule update --init --recursive
WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}
RESULT_VARIABLE GIT_SUBMOD_RESULT)
if(NOT GIT_SUBMOD_RESULT EQUAL "0")
message(WARNING "git submodule update --init --recursive failed with ${GIT_SUBMOD_RESULT}, please checkout submodules")
endif()
endif()
endif()
if(NOT EXISTS "${PROJECT_SOURCE_DIR}/tools/taos-tools/CMakeLists.txt")
message(WARNING "The submodules were not downloaded! GIT_SUBMODULE was turned off or failed. Please update submodules manually if you need build them.")
endif()
if (NOT DEFINED TD_GRANT) if (NOT DEFINED TD_GRANT)
SET(TD_GRANT FALSE) SET(TD_GRANT FALSE)
endif() endif()
......
include/common/tcommon.h
浏览文件 @ 9025a5d7
...@@ -45,6 +45,12 @@ enum { ...@@ -45,6 +45,12 @@ enum {
STREAM_TRIGGER__BY_EVENT_TIME, STREAM_TRIGGER__BY_EVENT_TIME,
}; };
typedef enum EStreamType {
STREAM_NORMAL = 1,
STREAM_INVERT,
STREAM_INVALID,
} EStreamType;
typedef struct { typedef struct {
uint32_t numOfTables; uint32_t numOfTables;
SArray* pGroupList; SArray* pGroupList;
...@@ -71,6 +77,7 @@ typedef struct SDataBlockInfo { ...@@ -71,6 +77,7 @@ typedef struct SDataBlockInfo {
int16_t numOfCols; int16_t numOfCols;
int16_t hasVarCol; int16_t hasVarCol;
int32_t capacity; int32_t capacity;
EStreamType type;
} SDataBlockInfo; } SDataBlockInfo;
typedef struct SSDataBlock { typedef struct SSDataBlock {
......
include/common/tmsg.h
浏览文件 @ 9025a5d7
...@@ -135,6 +135,8 @@ typedef enum _mgmt_table { ...@@ -135,6 +135,8 @@ typedef enum _mgmt_table {
#define TSDB_ALTER_USER_ADD_WRITE_DB 0x6 #define TSDB_ALTER_USER_ADD_WRITE_DB 0x6
#define TSDB_ALTER_USER_REMOVE_WRITE_DB 0x7 #define TSDB_ALTER_USER_REMOVE_WRITE_DB 0x7
#define TSDB_ALTER_USER_CLEAR_WRITE_DB 0x8 #define TSDB_ALTER_USER_CLEAR_WRITE_DB 0x8
#define TSDB_ALTER_USER_ADD_ALL_DB 0x9
#define TSDB_ALTER_USER_REMOVE_ALL_DB 0xA
#define TSDB_ALTER_USER_PRIVILEGES 0x2 #define TSDB_ALTER_USER_PRIVILEGES 0x2
...@@ -338,8 +340,8 @@ typedef struct SEpSet { ...@@ -338,8 +340,8 @@ typedef struct SEpSet {
SEp eps[TSDB_MAX_REPLICA]; SEp eps[TSDB_MAX_REPLICA];
} SEpSet; } SEpSet;
int32_t tEncodeSEpSet(SCoder* pEncoder, const SEpSet* pEp); int32_t tEncodeSEpSet(SEncoder* pEncoder, const SEpSet* pEp);
int32_t tDecodeSEpSet(SCoder* pDecoder, SEpSet* pEp); int32_t tDecodeSEpSet(SDecoder* pDecoder, SEpSet* pEp);
int32_t taosEncodeSEpSet(void** buf, const SEpSet* pEp); int32_t taosEncodeSEpSet(void** buf, const SEpSet* pEp);
void* taosDecodeSEpSet(const void* buf, SEpSet* pEp); void* taosDecodeSEpSet(const void* buf, SEpSet* pEp);
...@@ -611,8 +613,8 @@ typedef struct { ...@@ -611,8 +613,8 @@ typedef struct {
int32_t tSerializeSUseDbRsp(void* buf, int32_t bufLen, const SUseDbRsp* pRsp); int32_t tSerializeSUseDbRsp(void* buf, int32_t bufLen, const SUseDbRsp* pRsp);
int32_t tDeserializeSUseDbRsp(void* buf, int32_t bufLen, SUseDbRsp* pRsp); int32_t tDeserializeSUseDbRsp(void* buf, int32_t bufLen, SUseDbRsp* pRsp);
int32_t tSerializeSUseDbRspImp(SCoder* pEncoder, const SUseDbRsp* pRsp); int32_t tSerializeSUseDbRspImp(SEncoder* pEncoder, const SUseDbRsp* pRsp);
int32_t tDeserializeSUseDbRspImp(SCoder* pDecoder, SUseDbRsp* pRsp); int32_t tDeserializeSUseDbRspImp(SDecoder* pDecoder, SUseDbRsp* pRsp);
void tFreeSUsedbRsp(SUseDbRsp* pRsp); void tFreeSUsedbRsp(SUseDbRsp* pRsp);
typedef struct { typedef struct {
...@@ -1527,8 +1529,8 @@ typedef struct { ...@@ -1527,8 +1529,8 @@ typedef struct {
char* qmsg2; // pAst2:qmsg2:SRetention2 => trigger aggr task2 char* qmsg2; // pAst2:qmsg2:SRetention2 => trigger aggr task2
} SRSmaParam; } SRSmaParam;
int32_t tEncodeSRSmaParam(SCoder* pCoder, const SRSmaParam* pRSmaParam); int32_t tEncodeSRSmaParam(SEncoder* pCoder, const SRSmaParam* pRSmaParam);
int32_t tDecodeSRSmaParam(SCoder* pCoder, SRSmaParam* pRSmaParam); int32_t tDecodeSRSmaParam(SDecoder* pCoder, SRSmaParam* pRSmaParam);
// TDMT_VND_CREATE_STB ============== // TDMT_VND_CREATE_STB ==============
typedef struct SVCreateStbReq { typedef struct SVCreateStbReq {
...@@ -1540,8 +1542,8 @@ typedef struct SVCreateStbReq { ...@@ -1540,8 +1542,8 @@ typedef struct SVCreateStbReq {
SRSmaParam pRSmaParam; SRSmaParam pRSmaParam;
} SVCreateStbReq; } SVCreateStbReq;
int tEncodeSVCreateStbReq(SCoder* pCoder, const SVCreateStbReq* pReq); int tEncodeSVCreateStbReq(SEncoder* pCoder, const SVCreateStbReq* pReq);
int tDecodeSVCreateStbReq(SCoder* pCoder, SVCreateStbReq* pReq); int tDecodeSVCreateStbReq(SDecoder* pCoder, SVCreateStbReq* pReq);
// TDMT_VND_DROP_STB ============== // TDMT_VND_DROP_STB ==============
typedef struct SVDropStbReq { typedef struct SVDropStbReq {
...@@ -1549,8 +1551,8 @@ typedef struct SVDropStbReq { ...@@ -1549,8 +1551,8 @@ typedef struct SVDropStbReq {
tb_uid_t suid; tb_uid_t suid;
} SVDropStbReq; } SVDropStbReq;
int32_t tEncodeSVDropStbReq(SCoder* pCoder, const SVDropStbReq* pReq); int32_t tEncodeSVDropStbReq(SEncoder* pCoder, const SVDropStbReq* pReq);
int32_t tDecodeSVDropStbReq(SCoder* pCoder, SVDropStbReq* pReq); int32_t tDecodeSVDropStbReq(SDecoder* pCoder, SVDropStbReq* pReq);
#define TD_CREATE_IF_NOT_EXISTS 0x1 #define TD_CREATE_IF_NOT_EXISTS 0x1
typedef struct SVCreateTbReq { typedef struct SVCreateTbReq {
...@@ -1562,8 +1564,8 @@ typedef struct SVCreateTbReq { ...@@ -1562,8 +1564,8 @@ typedef struct SVCreateTbReq {
int8_t type; int8_t type;
union { union {
struct { struct {
tb_uid_t suid; tb_uid_t suid;
const void* pTag; const uint8_t* pTag;
} ctb; } ctb;
struct { struct {
SSchemaWrapper schema; SSchemaWrapper schema;
...@@ -1571,8 +1573,8 @@ typedef struct SVCreateTbReq { ...@@ -1571,8 +1573,8 @@ typedef struct SVCreateTbReq {
}; };
} SVCreateTbReq; } SVCreateTbReq;
int tEncodeSVCreateTbReq(SCoder* pCoder, const SVCreateTbReq* pReq); int tEncodeSVCreateTbReq(SEncoder* pCoder, const SVCreateTbReq* pReq);
int tDecodeSVCreateTbReq(SCoder* pCoder, SVCreateTbReq* pReq); int tDecodeSVCreateTbReq(SDecoder* pCoder, SVCreateTbReq* pReq);
typedef struct { typedef struct {
int32_t nReqs; int32_t nReqs;
...@@ -1582,15 +1584,15 @@ typedef struct { ...@@ -1582,15 +1584,15 @@ typedef struct {
}; };
} SVCreateTbBatchReq; } SVCreateTbBatchReq;
int tEncodeSVCreateTbBatchReq(SCoder* pCoder, const SVCreateTbBatchReq* pReq); int tEncodeSVCreateTbBatchReq(SEncoder* pCoder, const SVCreateTbBatchReq* pReq);
int tDecodeSVCreateTbBatchReq(SCoder* pCoder, SVCreateTbBatchReq* pReq); int tDecodeSVCreateTbBatchReq(SDecoder* pCoder, SVCreateTbBatchReq* pReq);
typedef struct { typedef struct {
int32_t code; int32_t code;
} SVCreateTbRsp, SVUpdateTbRsp; } SVCreateTbRsp, SVUpdateTbRsp;
int tEncodeSVCreateTbRsp(SCoder* pCoder, const SVCreateTbRsp* pRsp); int tEncodeSVCreateTbRsp(SEncoder* pCoder, const SVCreateTbRsp* pRsp);
int tDecodeSVCreateTbRsp(SCoder* pCoder, SVCreateTbRsp* pRsp); int tDecodeSVCreateTbRsp(SDecoder* pCoder, SVCreateTbRsp* pRsp);
int32_t tSerializeSVCreateTbReq(void** buf, SVCreateTbReq* pReq); int32_t tSerializeSVCreateTbReq(void** buf, SVCreateTbReq* pReq);
void* tDeserializeSVCreateTbReq(void* buf, SVCreateTbReq* pReq); void* tDeserializeSVCreateTbReq(void* buf, SVCreateTbReq* pReq);
...@@ -1603,8 +1605,8 @@ typedef struct { ...@@ -1603,8 +1605,8 @@ typedef struct {
}; };
} SVCreateTbBatchRsp; } SVCreateTbBatchRsp;
int tEncodeSVCreateTbBatchRsp(SCoder* pCoder, const SVCreateTbBatchRsp* pRsp); int tEncodeSVCreateTbBatchRsp(SEncoder* pCoder, const SVCreateTbBatchRsp* pRsp);
int tDecodeSVCreateTbBatchRsp(SCoder* pCoder, SVCreateTbBatchRsp* pRsp); int tDecodeSVCreateTbBatchRsp(SDecoder* pCoder, SVCreateTbBatchRsp* pRsp);
int32_t tSerializeSVCreateTbBatchRsp(void* buf, int32_t bufLen, SVCreateTbBatchRsp* pRsp); int32_t tSerializeSVCreateTbBatchRsp(void* buf, int32_t bufLen, SVCreateTbBatchRsp* pRsp);
int32_t tDeserializeSVCreateTbBatchRsp(void* buf, int32_t bufLen, SVCreateTbBatchRsp* pRsp); int32_t tDeserializeSVCreateTbBatchRsp(void* buf, int32_t bufLen, SVCreateTbBatchRsp* pRsp);
...@@ -1627,8 +1629,8 @@ typedef struct { ...@@ -1627,8 +1629,8 @@ typedef struct {
}; };
} SVDropTbBatchReq; } SVDropTbBatchReq;
int32_t tEncodeSVDropTbBatchReq(SCoder* pCoder, const SVDropTbBatchReq* pReq); int32_t tEncodeSVDropTbBatchReq(SEncoder* pCoder, const SVDropTbBatchReq* pReq);
int32_t tDecodeSVDropTbBatchReq(SCoder* pCoder, SVDropTbBatchReq* pReq); int32_t tDecodeSVDropTbBatchReq(SDecoder* pCoder, SVDropTbBatchReq* pReq);
typedef struct { typedef struct {
int32_t nRsps; int32_t nRsps;
...@@ -1638,8 +1640,8 @@ typedef struct { ...@@ -1638,8 +1640,8 @@ typedef struct {
}; };
} SVDropTbBatchRsp; } SVDropTbBatchRsp;
int32_t tEncodeSVDropTbBatchRsp(SCoder* pCoder, const SVDropTbBatchRsp* pRsp); int32_t tEncodeSVDropTbBatchRsp(SEncoder* pCoder, const SVDropTbBatchRsp* pRsp);
int32_t tDecodeSVDropTbBatchRsp(SCoder* pCoder, SVDropTbBatchRsp* pRsp); int32_t tDecodeSVDropTbBatchRsp(SDecoder* pCoder, SVDropTbBatchRsp* pRsp);
typedef struct { typedef struct {
SMsgHead head; SMsgHead head;
...@@ -1821,14 +1823,14 @@ static FORCE_INLINE void tFreeClientHbBatchRsp(void* pRsp) { ...@@ -1821,14 +1823,14 @@ static FORCE_INLINE void tFreeClientHbBatchRsp(void* pRsp) {
int32_t tSerializeSClientHbBatchRsp(void* buf, int32_t bufLen, const SClientHbBatchRsp* pBatchRsp); int32_t tSerializeSClientHbBatchRsp(void* buf, int32_t bufLen, const SClientHbBatchRsp* pBatchRsp);
int32_t tDeserializeSClientHbBatchRsp(void* buf, int32_t bufLen, SClientHbBatchRsp* pBatchRsp); int32_t tDeserializeSClientHbBatchRsp(void* buf, int32_t bufLen, SClientHbBatchRsp* pBatchRsp);
static FORCE_INLINE int32_t tEncodeSKv(SCoder* pEncoder, const SKv* pKv) { static FORCE_INLINE int32_t tEncodeSKv(SEncoder* pEncoder, const SKv* pKv) {
if (tEncodeI32(pEncoder, pKv->key) < 0) return -1; if (tEncodeI32(pEncoder, pKv->key) < 0) return -1;
if (tEncodeI32(pEncoder, pKv->valueLen) < 0) return -1; if (tEncodeI32(pEncoder, pKv->valueLen) < 0) return -1;
if (tEncodeBinary(pEncoder, (const char*)pKv->value, pKv->valueLen) < 0) return -1; if (tEncodeBinary(pEncoder, (uint8_t*)pKv->value, pKv->valueLen) < 0) return -1;
return 0; return 0;
} }
static FORCE_INLINE int32_t tDecodeSKv(SCoder* pDecoder, SKv* pKv) { static FORCE_INLINE int32_t tDecodeSKv(SDecoder* pDecoder, SKv* pKv) {
if (tDecodeI32(pDecoder, &pKv->key) < 0) return -1; if (tDecodeI32(pDecoder, &pKv->key) < 0) return -1;
if (tDecodeI32(pDecoder, &pKv->valueLen) < 0) return -1; if (tDecodeI32(pDecoder, &pKv->valueLen) < 0) return -1;
pKv->value = taosMemoryMalloc(pKv->valueLen + 1); pKv->value = taosMemoryMalloc(pKv->valueLen + 1);
...@@ -1837,13 +1839,13 @@ static FORCE_INLINE int32_t tDecodeSKv(SCoder* pDecoder, SKv* pKv) { ...@@ -1837,13 +1839,13 @@ static FORCE_INLINE int32_t tDecodeSKv(SCoder* pDecoder, SKv* pKv) {
return 0; return 0;
} }
static FORCE_INLINE int32_t tEncodeSClientHbKey(SCoder* pEncoder, const SClientHbKey* pKey) { static FORCE_INLINE int32_t tEncodeSClientHbKey(SEncoder* pEncoder, const SClientHbKey* pKey) {
if (tEncodeI64(pEncoder, pKey->tscRid) < 0) return -1; if (tEncodeI64(pEncoder, pKey->tscRid) < 0) return -1;
if (tEncodeI8(pEncoder, pKey->connType) < 0) return -1; if (tEncodeI8(pEncoder, pKey->connType) < 0) return -1;
return 0; return 0;
} }
static FORCE_INLINE int32_t tDecodeSClientHbKey(SCoder* pDecoder, SClientHbKey* pKey) { static FORCE_INLINE int32_t tDecodeSClientHbKey(SDecoder* pDecoder, SClientHbKey* pKey) {
if (tDecodeI64(pDecoder, &pKey->tscRid) < 0) return -1; if (tDecodeI64(pDecoder, &pKey->tscRid) < 0) return -1;
if (tDecodeI8(pDecoder, &pKey->connType) < 0) return -1; if (tDecodeI8(pDecoder, &pKey->connType) < 0) return -1;
return 0; return 0;
...@@ -2046,10 +2048,10 @@ typedef struct { ...@@ -2046,10 +2048,10 @@ typedef struct {
int32_t reserved; int32_t reserved;
} SMqCMCommitOffsetRsp; } SMqCMCommitOffsetRsp;
int32_t tEncodeSMqOffset(SCoder* encoder, const SMqOffset* pOffset); int32_t tEncodeSMqOffset(SEncoder* encoder, const SMqOffset* pOffset);
int32_t tDecodeSMqOffset(SCoder* decoder, SMqOffset* pOffset); int32_t tDecodeSMqOffset(SDecoder* decoder, SMqOffset* pOffset);
int32_t tEncodeSMqCMCommitOffsetReq(SCoder* encoder, const SMqCMCommitOffsetReq* pReq); int32_t tEncodeSMqCMCommitOffsetReq(SEncoder* encoder, const SMqCMCommitOffsetReq* pReq);
int32_t tDecodeSMqCMCommitOffsetReq(SCoder* decoder, SMqCMCommitOffsetReq* pReq); int32_t tDecodeSMqCMCommitOffsetReq(SDecoder* decoder, SMqCMCommitOffsetReq* pReq);
static FORCE_INLINE SSchemaWrapper* tCloneSSchemaWrapper(const SSchemaWrapper* pSchemaWrapper) { static FORCE_INLINE SSchemaWrapper* tCloneSSchemaWrapper(const SSchemaWrapper* pSchemaWrapper) {
SSchemaWrapper* pSW = (SSchemaWrapper*)taosMemoryMalloc(sizeof(SSchemaWrapper)); SSchemaWrapper* pSW = (SSchemaWrapper*)taosMemoryMalloc(sizeof(SSchemaWrapper));
...@@ -2089,7 +2091,7 @@ static FORCE_INLINE void* taosDecodeSSchema(const void* buf, SSchema* pSchema) { ...@@ -2089,7 +2091,7 @@ static FORCE_INLINE void* taosDecodeSSchema(const void* buf, SSchema* pSchema) {
return (void*)buf; return (void*)buf;
} }
static FORCE_INLINE int32_t tEncodeSSchema(SCoder* pEncoder, const SSchema* pSchema) { static FORCE_INLINE int32_t tEncodeSSchema(SEncoder* pEncoder, const SSchema* pSchema) {
if (tEncodeI8(pEncoder, pSchema->type) < 0) return -1; if (tEncodeI8(pEncoder, pSchema->type) < 0) return -1;
if (tEncodeI8(pEncoder, pSchema->flags) < 0) return -1; if (tEncodeI8(pEncoder, pSchema->flags) < 0) return -1;
if (tEncodeI32v(pEncoder, pSchema->bytes) < 0) return -1; if (tEncodeI32v(pEncoder, pSchema->bytes) < 0) return -1;
...@@ -2098,7 +2100,7 @@ static FORCE_INLINE int32_t tEncodeSSchema(SCoder* pEncoder, const SSchema* pSch ...@@ -2098,7 +2100,7 @@ static FORCE_INLINE int32_t tEncodeSSchema(SCoder* pEncoder, const SSchema* pSch
return 0; return 0;
} }
static FORCE_INLINE int32_t tDecodeSSchema(SCoder* pDecoder, SSchema* pSchema) { static FORCE_INLINE int32_t tDecodeSSchema(SDecoder* pDecoder, SSchema* pSchema) {
if (tDecodeI8(pDecoder, &pSchema->type) < 0) return -1; if (tDecodeI8(pDecoder, &pSchema->type) < 0) return -1;
if (tDecodeI8(pDecoder, &pSchema->flags) < 0) return -1; if (tDecodeI8(pDecoder, &pSchema->flags) < 0) return -1;
if (tDecodeI32v(pDecoder, &pSchema->bytes) < 0) return -1; if (tDecodeI32v(pDecoder, &pSchema->bytes) < 0) return -1;
...@@ -2131,7 +2133,7 @@ static FORCE_INLINE void* taosDecodeSSchemaWrapper(const void* buf, SSchemaWrapp ...@@ -2131,7 +2133,7 @@ static FORCE_INLINE void* taosDecodeSSchemaWrapper(const void* buf, SSchemaWrapp
return (void*)buf; return (void*)buf;
} }
static FORCE_INLINE int32_t tEncodeSSchemaWrapper(SCoder* pEncoder, const SSchemaWrapper* pSW) { static FORCE_INLINE int32_t tEncodeSSchemaWrapper(SEncoder* pEncoder, const SSchemaWrapper* pSW) {
if (tEncodeI32v(pEncoder, pSW->nCols) < 0) return -1; if (tEncodeI32v(pEncoder, pSW->nCols) < 0) return -1;
if (tEncodeI32v(pEncoder, pSW->sver) < 0) return -1; if (tEncodeI32v(pEncoder, pSW->sver) < 0) return -1;
for (int32_t i = 0; i < pSW->nCols; i++) { for (int32_t i = 0; i < pSW->nCols; i++) {
...@@ -2141,7 +2143,7 @@ static FORCE_INLINE int32_t tEncodeSSchemaWrapper(SCoder* pEncoder, const SSchem ...@@ -2141,7 +2143,7 @@ static FORCE_INLINE int32_t tEncodeSSchemaWrapper(SCoder* pEncoder, const SSchem
return 0; return 0;
} }
static FORCE_INLINE int32_t tDecodeSSchemaWrapper(SCoder* pDecoder, SSchemaWrapper* pSW) { static FORCE_INLINE int32_t tDecodeSSchemaWrapper(SDecoder* pDecoder, SSchemaWrapper* pSW) {
if (tDecodeI32v(pDecoder, &pSW->nCols) < 0) return -1; if (tDecodeI32v(pDecoder, &pSW->nCols) < 0) return -1;
if (tDecodeI32v(pDecoder, &pSW->sver) < 0) return -1; if (tDecodeI32v(pDecoder, &pSW->sver) < 0) return -1;
...@@ -2593,12 +2595,12 @@ static FORCE_INLINE void tDeleteSMqAskEpRsp(SMqAskEpRsp* pRsp) { ...@@ -2593,12 +2595,12 @@ static FORCE_INLINE void tDeleteSMqAskEpRsp(SMqAskEpRsp* pRsp) {
#define TD_AUTO_CREATE_TABLE 0x1 #define TD_AUTO_CREATE_TABLE 0x1
typedef struct { typedef struct {
int64_t suid; int64_t suid;
int64_t uid; int64_t uid;
int32_t sver; int32_t sver;
uint64_t nData; uint32_t nData;
const void* pData; const uint8_t* pData;
SVCreateTbReq cTbReq; SVCreateTbReq cTbReq;
} SVSubmitBlk; } SVSubmitBlk;
typedef struct { typedef struct {
...@@ -2610,8 +2612,8 @@ typedef struct { ...@@ -2610,8 +2612,8 @@ typedef struct {
}; };
} SVSubmitReq; } SVSubmitReq;
int32_t tEncodeSVSubmitReq(SCoder* pCoder, const SVSubmitReq* pReq); int32_t tEncodeSVSubmitReq(SEncoder* pCoder, const SVSubmitReq* pReq);
int32_t tDecodeSVSubmitReq(SCoder* pCoder, SVSubmitReq* pReq); int32_t tDecodeSVSubmitReq(SDecoder* pCoder, SVSubmitReq* pReq);
#pragma pack(pop) #pragma pack(pop)
......
include/common/ttokendef.h
浏览文件 @ 9025a5d7
...@@ -51,201 +51,208 @@ ...@@ -51,201 +51,208 @@
#define TK_USER 33 #define TK_USER 33
#define TK_PRIVILEGE 34 #define TK_PRIVILEGE 34
#define TK_DROP 35 #define TK_DROP 35
#define TK_DNODE 36 #define TK_GRANT 36
#define TK_PORT 37 #define TK_ON 37
#define TK_NK_INTEGER 38 #define TK_TO 38
#define TK_DNODES 39 #define TK_REVOKE 39
#define TK_NK_IPTOKEN 40 #define TK_FROM 40
#define TK_LOCAL 41 #define TK_NK_COMMA 41
#define TK_QNODE 42 #define TK_READ 42
#define TK_ON 43 #define TK_WRITE 43
#define TK_BNODE 44 #define TK_NK_DOT 44
#define TK_SNODE 45 #define TK_DNODE 45
#define TK_MNODE 46 #define TK_PORT 46
#define TK_DATABASE 47 #define TK_NK_INTEGER 47
#define TK_USE 48 #define TK_DNODES 48
#define TK_IF 49 #define TK_NK_IPTOKEN 49
#define TK_NOT 50 #define TK_LOCAL 50
#define TK_EXISTS 51 #define TK_QNODE 51
#define TK_BUFFER 52 #define TK_BNODE 52
#define TK_CACHELAST 53 #define TK_SNODE 53
#define TK_COMP 54 #define TK_MNODE 54
#define TK_DAYS 55 #define TK_DATABASE 55
#define TK_NK_VARIABLE 56 #define TK_USE 56
#define TK_FSYNC 57 #define TK_IF 57
#define TK_MAXROWS 58 #define TK_NOT 58
#define TK_MINROWS 59 #define TK_EXISTS 59
#define TK_KEEP 60 #define TK_BUFFER 60
#define TK_PAGES 61 #define TK_CACHELAST 61
#define TK_PAGESIZE 62 #define TK_COMP 62
#define TK_PRECISION 63 #define TK_DAYS 63
#define TK_REPLICA 64 #define TK_NK_VARIABLE 64
#define TK_STRICT 65 #define TK_FSYNC 65
#define TK_WAL 66 #define TK_MAXROWS 66
#define TK_VGROUPS 67 #define TK_MINROWS 67
#define TK_SINGLE_STABLE 68 #define TK_KEEP 68
#define TK_RETENTIONS 69 #define TK_PAGES 69
#define TK_NK_COMMA 70 #define TK_PAGESIZE 70
#define TK_NK_COLON 71 #define TK_PRECISION 71
#define TK_TABLE 72 #define TK_REPLICA 72
#define TK_NK_LP 73 #define TK_STRICT 73
#define TK_NK_RP 74 #define TK_WAL 74
#define TK_STABLE 75 #define TK_VGROUPS 75
#define TK_ADD 76 #define TK_SINGLE_STABLE 76
#define TK_COLUMN 77 #define TK_RETENTIONS 77
#define TK_MODIFY 78 #define TK_NK_COLON 78
#define TK_RENAME 79 #define TK_TABLE 79
#define TK_TAG 80 #define TK_NK_LP 80
#define TK_SET 81 #define TK_NK_RP 81
#define TK_NK_EQ 82 #define TK_STABLE 82
#define TK_USING 83 #define TK_ADD 83
#define TK_TAGS 84 #define TK_COLUMN 84
#define TK_NK_DOT 85 #define TK_MODIFY 85
#define TK_COMMENT 86 #define TK_RENAME 86
#define TK_BOOL 87 #define TK_TAG 87
#define TK_TINYINT 88 #define TK_SET 88
#define TK_SMALLINT 89 #define TK_NK_EQ 89
#define TK_INT 90 #define TK_USING 90
#define TK_INTEGER 91 #define TK_TAGS 91
#define TK_BIGINT 92 #define TK_COMMENT 92
#define TK_FLOAT 93 #define TK_BOOL 93
#define TK_DOUBLE 94 #define TK_TINYINT 94
#define TK_BINARY 95 #define TK_SMALLINT 95
#define TK_TIMESTAMP 96 #define TK_INT 96
#define TK_NCHAR 97 #define TK_INTEGER 97
#define TK_UNSIGNED 98 #define TK_BIGINT 98
#define TK_JSON 99 #define TK_FLOAT 99
#define TK_VARCHAR 100 #define TK_DOUBLE 100
#define TK_MEDIUMBLOB 101 #define TK_BINARY 101
#define TK_BLOB 102 #define TK_TIMESTAMP 102
#define TK_VARBINARY 103 #define TK_NCHAR 103
#define TK_DECIMAL 104 #define TK_UNSIGNED 104
#define TK_DELAY 105 #define TK_JSON 105
#define TK_FILE_FACTOR 106 #define TK_VARCHAR 106
#define TK_NK_FLOAT 107 #define TK_MEDIUMBLOB 107
#define TK_ROLLUP 108 #define TK_BLOB 108
#define TK_TTL 109 #define TK_VARBINARY 109
#define TK_SMA 110 #define TK_DECIMAL 110
#define TK_SHOW 111 #define TK_DELAY 111
#define TK_DATABASES 112 #define TK_FILE_FACTOR 112
#define TK_TABLES 113 #define TK_NK_FLOAT 113
#define TK_STABLES 114 #define TK_ROLLUP 114
#define TK_MNODES 115 #define TK_TTL 115
#define TK_MODULES 116 #define TK_SMA 116
#define TK_QNODES 117 #define TK_SHOW 117
#define TK_FUNCTIONS 118 #define TK_DATABASES 118
#define TK_INDEXES 119 #define TK_TABLES 119
#define TK_FROM 120 #define TK_STABLES 120
#define TK_ACCOUNTS 121 #define TK_MNODES 121
#define TK_APPS 122 #define TK_MODULES 122
#define TK_CONNECTIONS 123 #define TK_QNODES 123
#define TK_LICENCE 124 #define TK_FUNCTIONS 124
#define TK_GRANTS 125 #define TK_INDEXES 125
#define TK_QUERIES 126 #define TK_ACCOUNTS 126
#define TK_SCORES 127 #define TK_APPS 127
#define TK_TOPICS 128 #define TK_CONNECTIONS 128
#define TK_VARIABLES 129 #define TK_LICENCE 129
#define TK_BNODES 130 #define TK_GRANTS 130
#define TK_SNODES 131 #define TK_QUERIES 131
#define TK_CLUSTER 132 #define TK_SCORES 132
#define TK_LIKE 133 #define TK_TOPICS 133
#define TK_INDEX 134 #define TK_VARIABLES 134
#define TK_FULLTEXT 135 #define TK_BNODES 135
#define TK_FUNCTION 136 #define TK_SNODES 136
#define TK_INTERVAL 137 #define TK_CLUSTER 137
#define TK_TOPIC 138 #define TK_TRANSACTIONS 138
#define TK_AS 139 #define TK_LIKE 139
#define TK_WITH 140 #define TK_INDEX 140
#define TK_SCHEMA 141 #define TK_FULLTEXT 141
#define TK_DESC 142 #define TK_FUNCTION 142
#define TK_DESCRIBE 143 #define TK_INTERVAL 143
#define TK_RESET 144 #define TK_TOPIC 144
#define TK_QUERY 145 #define TK_AS 145
#define TK_CACHE 146 #define TK_WITH 146
#define TK_EXPLAIN 147 #define TK_SCHEMA 147
#define TK_ANALYZE 148 #define TK_DESC 148
#define TK_VERBOSE 149 #define TK_DESCRIBE 149
#define TK_NK_BOOL 150 #define TK_RESET 150
#define TK_RATIO 151 #define TK_QUERY 151
#define TK_COMPACT 152 #define TK_CACHE 152
#define TK_VNODES 153 #define TK_EXPLAIN 153
#define TK_IN 154 #define TK_ANALYZE 154
#define TK_OUTPUTTYPE 155 #define TK_VERBOSE 155
#define TK_AGGREGATE 156 #define TK_NK_BOOL 156
#define TK_BUFSIZE 157 #define TK_RATIO 157
#define TK_STREAM 158 #define TK_COMPACT 158
#define TK_INTO 159 #define TK_VNODES 159
#define TK_TRIGGER 160 #define TK_IN 160
#define TK_AT_ONCE 161 #define TK_OUTPUTTYPE 161
#define TK_WINDOW_CLOSE 162 #define TK_AGGREGATE 162
#define TK_WATERMARK 163 #define TK_BUFSIZE 163
#define TK_KILL 164 #define TK_STREAM 164
#define TK_CONNECTION 165 #define TK_INTO 165
#define TK_MERGE 166 #define TK_TRIGGER 166
#define TK_VGROUP 167 #define TK_AT_ONCE 167
#define TK_REDISTRIBUTE 168 #define TK_WINDOW_CLOSE 168
#define TK_SPLIT 169 #define TK_WATERMARK 169
#define TK_SYNCDB 170 #define TK_KILL 170
#define TK_NULL 171 #define TK_CONNECTION 171
#define TK_NK_QUESTION 172 #define TK_TRANSACTION 172
#define TK_NK_ARROW 173 #define TK_MERGE 173
#define TK_ROWTS 174 #define TK_VGROUP 174
#define TK_TBNAME 175 #define TK_REDISTRIBUTE 175
#define TK_QSTARTTS 176 #define TK_SPLIT 176
#define TK_QENDTS 177 #define TK_SYNCDB 177
#define TK_WSTARTTS 178 #define TK_NULL 178
#define TK_WENDTS 179 #define TK_NK_QUESTION 179
#define TK_WDURATION 180 #define TK_NK_ARROW 180
#define TK_CAST 181 #define TK_ROWTS 181
#define TK_NOW 182 #define TK_TBNAME 182
#define TK_TODAY 183 #define TK_QSTARTTS 183
#define TK_TIMEZONE 184 #define TK_QENDTS 184
#define TK_COUNT 185 #define TK_WSTARTTS 185
#define TK_FIRST 186 #define TK_WENDTS 186
#define TK_LAST 187 #define TK_WDURATION 187
#define TK_LAST_ROW 188 #define TK_CAST 188
#define TK_BETWEEN 189 #define TK_NOW 189
#define TK_IS 190 #define TK_TODAY 190
#define TK_NK_LT 191 #define TK_TIMEZONE 191
#define TK_NK_GT 192 #define TK_COUNT 192
#define TK_NK_LE 193 #define TK_FIRST 193
#define TK_NK_GE 194 #define TK_LAST 194
#define TK_NK_NE 195 #define TK_LAST_ROW 195
#define TK_MATCH 196 #define TK_BETWEEN 196
#define TK_NMATCH 197 #define TK_IS 197
#define TK_CONTAINS 198 #define TK_NK_LT 198
#define TK_JOIN 199 #define TK_NK_GT 199
#define TK_INNER 200 #define TK_NK_LE 200
#define TK_SELECT 201 #define TK_NK_GE 201
#define TK_DISTINCT 202 #define TK_NK_NE 202
#define TK_WHERE 203 #define TK_MATCH 203
#define TK_PARTITION 204 #define TK_NMATCH 204
#define TK_BY 205 #define TK_CONTAINS 205
#define TK_SESSION 206 #define TK_JOIN 206
#define TK_STATE_WINDOW 207 #define TK_INNER 207
#define TK_SLIDING 208 #define TK_SELECT 208
#define TK_FILL 209 #define TK_DISTINCT 209
#define TK_VALUE 210 #define TK_WHERE 210
#define TK_NONE 211 #define TK_PARTITION 211
#define TK_PREV 212 #define TK_BY 212
#define TK_LINEAR 213 #define TK_SESSION 213
#define TK_NEXT 214 #define TK_STATE_WINDOW 214
#define TK_GROUP 215 #define TK_SLIDING 215
#define TK_HAVING 216 #define TK_FILL 216
#define TK_ORDER 217 #define TK_VALUE 217
#define TK_SLIMIT 218 #define TK_NONE 218
#define TK_SOFFSET 219 #define TK_PREV 219
#define TK_LIMIT 220 #define TK_LINEAR 220
#define TK_OFFSET 221 #define TK_NEXT 221
#define TK_ASC 222 #define TK_GROUP 222
#define TK_NULLS 223 #define TK_HAVING 223
#define TK_ID 224 #define TK_ORDER 224
#define TK_NK_BITNOT 225 #define TK_SLIMIT 225
#define TK_INSERT 226 #define TK_SOFFSET 226
#define TK_VALUES 227 #define TK_LIMIT 227
#define TK_IMPORT 228 #define TK_OFFSET 228
#define TK_NK_SEMI 229 #define TK_ASC 229
#define TK_FILE 230 #define TK_NULLS 230
#define TK_ID 231
#define TK_NK_BITNOT 232
#define TK_INSERT 233
#define TK_VALUES 234
#define TK_IMPORT 235
#define TK_NK_SEMI 236
#define TK_FILE 237
#define TK_NK_SPACE 300 #define TK_NK_SPACE 300
#define TK_NK_COMMENT 301 #define TK_NK_COMMENT 301
......
include/libs/function/functionMgt.h
浏览文件 @ 9025a5d7
...@@ -168,6 +168,9 @@ EFuncDataRequired fmFuncDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWin ...@@ -168,6 +168,9 @@ EFuncDataRequired fmFuncDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWin
int32_t fmGetFuncExecFuncs(int32_t funcId, SFuncExecFuncs* pFpSet); int32_t fmGetFuncExecFuncs(int32_t funcId, SFuncExecFuncs* pFpSet);
int32_t fmGetScalarFuncExecFuncs(int32_t funcId, SScalarFuncExecFuncs* pFpSet); int32_t fmGetScalarFuncExecFuncs(int32_t funcId, SScalarFuncExecFuncs* pFpSet);
int32_t fmGetUdafExecFuncs(int32_t funcId, SFuncExecFuncs* pFpSet); int32_t fmGetUdafExecFuncs(int32_t funcId, SFuncExecFuncs* pFpSet);
int32_t fmSetInvertFunc(int32_t funcId, SFuncExecFuncs* pFpSet);
int32_t fmSetNormalFunc(int32_t funcId, SFuncExecFuncs* pFpSet);
bool fmIsInvertible(int32_t funcId);
#ifdef __cplusplus #ifdef __cplusplus
} }
......
include/libs/function/tudf.h
浏览文件 @ 9025a5d7
...@@ -44,7 +44,8 @@ enum { ...@@ -44,7 +44,8 @@ enum {
UDFC_CODE_PIPE_READ_ERR = -2, UDFC_CODE_PIPE_READ_ERR = -2,
UDFC_CODE_CONNECT_PIPE_ERR = -3, UDFC_CODE_CONNECT_PIPE_ERR = -3,
UDFC_CODE_LOAD_UDF_FAILURE = -4, UDFC_CODE_LOAD_UDF_FAILURE = -4,
UDFC_CODE_INVALID_STATE = -5 UDFC_CODE_INVALID_STATE = -5,
UDFC_CODE_NO_PIPE = -6,
}; };
typedef void *UdfcFuncHandle; typedef void *UdfcFuncHandle;
...@@ -140,6 +141,44 @@ typedef int32_t (*TUdfDestroyFunc)(); ...@@ -140,6 +141,44 @@ typedef int32_t (*TUdfDestroyFunc)();
#define UDF_MEMORY_EXP_GROWTH 1.5 #define UDF_MEMORY_EXP_GROWTH 1.5
#define udfColDataIsNull_var(pColumn, row) ((pColumn->colData.varLenCol.varOffsets)[row] == -1)
#define udfColDataIsNull_f(pColumn, row) ((BMCharPos(pColumn->colData.fixLenCol.nullBitmap, row) & (1u << (7u - BitPos(row)))) == (1u << (7u - BitPos(row))))
#define udfColDataSetNull_f(pColumn, row) \
do { \
BMCharPos(pColumn->colData.fixLenCol.nullBitmap, row) |= (1u << (7u - BitPos(row))); \
} while (0)
#define udfColDataSetNotNull_f(pColumn, r_) \
do { \
BMCharPos(pColumn->colData.fixLenCol.nullBitmap, r_) &= ~(1u << (7u - BitPos(r_))); \
} while (0)
#define udfColDataSetNull_var(pColumn, row) ((pColumn->colData.varLenCol.varOffsets)[row] = -1)
static FORCE_INLINE char* udfColDataGetData(const SUdfColumn* pColumn, int32_t row) {
if (IS_VAR_DATA_TYPE(pColumn->colMeta.type)) {
return pColumn->colData.varLenCol.payload + pColumn->colData.varLenCol.varOffsets[row];
} else {
return pColumn->colData.fixLenCol.data + pColumn->colMeta.bytes * row;
}
}
static FORCE_INLINE bool udfColDataIsNull(const SUdfColumn* pColumn, int32_t row) {
if (IS_VAR_DATA_TYPE(pColumn->colMeta.type)) {
if (pColumn->colMeta.type == TSDB_DATA_TYPE_JSON) {
if (udfColDataIsNull_var(pColumn, row)) {
return true;
}
char* data = udfColDataGetData(pColumn, row);
return (*data == TSDB_DATA_TYPE_NULL);
} else {
return udfColDataIsNull_var(pColumn, row);
}
} else {
return udfColDataIsNull_f(pColumn, row);
}
}
static FORCE_INLINE int32_t udfColEnsureCapacity(SUdfColumn* pColumn, int32_t newCapacity) { static FORCE_INLINE int32_t udfColEnsureCapacity(SUdfColumn* pColumn, int32_t newCapacity) {
SUdfColumnMeta *meta = &pColumn->colMeta; SUdfColumnMeta *meta = &pColumn->colMeta;
SUdfColumnData *data = &pColumn->colData; SUdfColumnData *data = &pColumn->colData;
...@@ -186,17 +225,22 @@ static FORCE_INLINE int32_t udfColEnsureCapacity(SUdfColumn* pColumn, int32_t ne ...@@ -186,17 +225,22 @@ static FORCE_INLINE int32_t udfColEnsureCapacity(SUdfColumn* pColumn, int32_t ne
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static FORCE_INLINE int32_t udfColSetRow(SUdfColumn* pColumn, uint32_t currentRow, const char* pData, bool isNull) { static FORCE_INLINE void udfColDataSetNull(SUdfColumn* pColumn, int32_t row) {
udfColEnsureCapacity(pColumn, row+1);
if (IS_VAR_DATA_TYPE(pColumn->colMeta.type)) {
udfColDataSetNull_var(pColumn, row);
} else {
udfColDataSetNull_f(pColumn, row);
}
}
static FORCE_INLINE int32_t udfColDataSet(SUdfColumn* pColumn, uint32_t currentRow, const char* pData, bool isNull) {
SUdfColumnMeta *meta = &pColumn->colMeta; SUdfColumnMeta *meta = &pColumn->colMeta;
SUdfColumnData *data = &pColumn->colData; SUdfColumnData *data = &pColumn->colData;
udfColEnsureCapacity(pColumn, currentRow+1); udfColEnsureCapacity(pColumn, currentRow+1);
bool isVarCol = IS_VAR_DATA_TYPE(meta->type); bool isVarCol = IS_VAR_DATA_TYPE(meta->type);
if (isNull) { if (isNull) {
if (isVarCol) { udfColDataSetNull(pColumn, currentRow);
data->varLenCol.varOffsets[currentRow] = -1;
} else {
colDataSetNull_f(data->fixLenCol.nullBitmap, currentRow);
}
} else { } else {
if (!isVarCol) { if (!isVarCol) {
colDataSetNotNull_f(data->fixLenCol.nullBitmap, currentRow); colDataSetNotNull_f(data->fixLenCol.nullBitmap, currentRow);
......
include/libs/nodes/cmdnodes.h
浏览文件 @ 9025a5d7
...@@ -310,6 +310,29 @@ typedef struct SCreateFunctionStmt { ...@@ -310,6 +310,29 @@ typedef struct SCreateFunctionStmt {
int32_t bufSize; int32_t bufSize;
} SCreateFunctionStmt; } SCreateFunctionStmt;
typedef struct SDropFunctionStmt {
ENodeType type;
char funcName[TSDB_FUNC_NAME_LEN];
bool ignoreNotExists;
} SDropFunctionStmt;
#define PRIVILEGE_TYPE_MASK(n) (1 << n)
#define PRIVILEGE_TYPE_ALL PRIVILEGE_TYPE_MASK(0)
#define PRIVILEGE_TYPE_READ PRIVILEGE_TYPE_MASK(1)
#define PRIVILEGE_TYPE_WRITE PRIVILEGE_TYPE_MASK(2)
#define PRIVILEGE_TYPE_TEST_MASK(val, mask) (((val) & (mask)) != 0)
typedef struct SGrantStmt {
ENodeType type;
char userName[TSDB_USER_LEN];
char dbName[TSDB_DB_NAME_LEN];
int64_t privileges;
} SGrantStmt;
typedef SGrantStmt SRevokeStmt;
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif
......
include/libs/nodes/nodes.h
浏览文件 @ 9025a5d7
...@@ -143,6 +143,8 @@ typedef enum ENodeType { ...@@ -143,6 +143,8 @@ typedef enum ENodeType {
QUERY_NODE_REDISTRIBUTE_VGROUP_STMT, QUERY_NODE_REDISTRIBUTE_VGROUP_STMT,
QUERY_NODE_SPLIT_VGROUP_STMT, QUERY_NODE_SPLIT_VGROUP_STMT,
QUERY_NODE_SYNCDB_STMT, QUERY_NODE_SYNCDB_STMT,
QUERY_NODE_GRANT_STMT,
QUERY_NODE_REVOKE_STMT,
QUERY_NODE_SHOW_DNODES_STMT, QUERY_NODE_SHOW_DNODES_STMT,
QUERY_NODE_SHOW_MNODES_STMT, QUERY_NODE_SHOW_MNODES_STMT,
QUERY_NODE_SHOW_MODULES_STMT, QUERY_NODE_SHOW_MODULES_STMT,
...@@ -174,8 +176,10 @@ typedef enum ENodeType { ...@@ -174,8 +176,10 @@ typedef enum ENodeType {
QUERY_NODE_SHOW_CREATE_DATABASE_STMT, QUERY_NODE_SHOW_CREATE_DATABASE_STMT,
QUERY_NODE_SHOW_CREATE_TABLE_STMT, QUERY_NODE_SHOW_CREATE_TABLE_STMT,
QUERY_NODE_SHOW_CREATE_STABLE_STMT, QUERY_NODE_SHOW_CREATE_STABLE_STMT,
QUERY_NODE_SHOW_TRANSACTIONS_STMT,
QUERY_NODE_KILL_CONNECTION_STMT, QUERY_NODE_KILL_CONNECTION_STMT,
QUERY_NODE_KILL_QUERY_STMT, QUERY_NODE_KILL_QUERY_STMT,
QUERY_NODE_KILL_TRANSACTION_STMT,
// logic plan node // logic plan node
QUERY_NODE_LOGIC_PLAN_SCAN, QUERY_NODE_LOGIC_PLAN_SCAN,
......
include/libs/planner/planner.h
浏览文件 @ 9025a5d7
...@@ -25,20 +25,18 @@ extern "C" { ...@@ -25,20 +25,18 @@ extern "C" {
typedef struct SPlanContext { typedef struct SPlanContext {
uint64_t queryId; uint64_t queryId;
int32_t acctId; int32_t acctId;
SEpSet mgmtEpSet; SEpSet mgmtEpSet;
SNode* pAstRoot; SNode* pAstRoot;
bool topicQuery; bool topicQuery;
bool streamQuery; bool streamQuery;
bool rSmaQuery; bool rSmaQuery;
bool showRewrite; bool showRewrite;
int8_t triggerType; int8_t triggerType;
int64_t watermark; int64_t watermark;
int32_t placeholderNum; int32_t placeholderNum;
void* pTransporter; char* pMsg;
struct SCatalog* pCatalog; int32_t msgLen;
char* pMsg;
int32_t msgLen;
} SPlanContext; } SPlanContext;
// Create the physical plan for the query, according to the AST. // Create the physical plan for the query, according to the AST.
...@@ -47,7 +45,7 @@ int32_t qCreateQueryPlan(SPlanContext* pCxt, SQueryPlan** pPlan, SArray* pExecNo ...@@ -47,7 +45,7 @@ int32_t qCreateQueryPlan(SPlanContext* pCxt, SQueryPlan** pPlan, SArray* pExecNo
// Set datasource of this subplan, multiple calls may be made to a subplan. // Set datasource of this subplan, multiple calls may be made to a subplan.
// @pSubplan subplan to be schedule // @pSubplan subplan to be schedule
// @groupId id of a group of datasource subplans of this @pSubplan // @groupId id of a group of datasource subplans of this @pSubplan
// @pSource one execution location of this group of datasource subplans // @pSource one execution location of this group of datasource subplans
int32_t qSetSubplanExecutionNode(SSubplan* pSubplan, int32_t groupId, SDownstreamSourceNode* pSource); int32_t qSetSubplanExecutionNode(SSubplan* pSubplan, int32_t groupId, SDownstreamSourceNode* pSource);
int32_t qStmtBindParam(SQueryPlan* pPlan, TAOS_MULTI_BIND* pParams, int32_t colIdx, uint64_t queryId); int32_t qStmtBindParam(SQueryPlan* pPlan, TAOS_MULTI_BIND* pParams, int32_t colIdx, uint64_t queryId);
...@@ -56,7 +54,7 @@ int32_t qStmtBindParam(SQueryPlan* pPlan, TAOS_MULTI_BIND* pParams, int32_t colI ...@@ -56,7 +54,7 @@ int32_t qStmtBindParam(SQueryPlan* pPlan, TAOS_MULTI_BIND* pParams, int32_t colI
int32_t qSubPlanToString(const SSubplan* pSubplan, char** pStr, int32_t* pLen); int32_t qSubPlanToString(const SSubplan* pSubplan, char** pStr, int32_t* pLen);
int32_t qStringToSubplan(const char* pStr, SSubplan** pSubplan); int32_t qStringToSubplan(const char* pStr, SSubplan** pSubplan);
char* qQueryPlanToString(const SQueryPlan* pPlan); char* qQueryPlanToString(const SQueryPlan* pPlan);
SQueryPlan* qStringToQueryPlan(const char* pStr); SQueryPlan* qStringToQueryPlan(const char* pStr);
void qDestroyQueryPlan(SQueryPlan* pPlan); void qDestroyQueryPlan(SQueryPlan* pPlan);
......
include/libs/stream/tstream.h
浏览文件 @ 9025a5d7
...@@ -154,8 +154,8 @@ typedef struct { ...@@ -154,8 +154,8 @@ typedef struct {
} SStreamTask; } SStreamTask;
SStreamTask* tNewSStreamTask(int64_t streamId); SStreamTask* tNewSStreamTask(int64_t streamId);
int32_t tEncodeSStreamTask(SCoder* pEncoder, const SStreamTask* pTask); int32_t tEncodeSStreamTask(SEncoder* pEncoder, const SStreamTask* pTask);
int32_t tDecodeSStreamTask(SCoder* pDecoder, SStreamTask* pTask); int32_t tDecodeSStreamTask(SDecoder* pDecoder, SStreamTask* pTask);
void tFreeSStreamTask(SStreamTask* pTask); void tFreeSStreamTask(SStreamTask* pTask);
typedef struct { typedef struct {
......
include/libs/stream/tstreamUpdate.h
浏览文件 @ 9025a5d7
...@@ -36,7 +36,7 @@ typedef struct SUpdateInfo { ...@@ -36,7 +36,7 @@ typedef struct SUpdateInfo {
SUpdateInfo *updateInfoInitP(SInterval* pInterval, int64_t watermark); SUpdateInfo *updateInfoInitP(SInterval* pInterval, int64_t watermark);
SUpdateInfo *updateInfoInit(int64_t interval, int32_t precision, int64_t watermark); SUpdateInfo *updateInfoInit(int64_t interval, int32_t precision, int64_t watermark);
bool isUpdated(SUpdateInfo *pInfo, tb_uid_t tableId, TSKEY ts); bool updateInfoIsUpdated(SUpdateInfo *pInfo, tb_uid_t tableId, TSKEY ts);
void updateInfoDestroy(SUpdateInfo *pInfo); void updateInfoDestroy(SUpdateInfo *pInfo);
#ifdef __cplusplus #ifdef __cplusplus
......
include/util/taoserror.h
浏览文件 @ 9025a5d7
...@@ -632,6 +632,7 @@ int32_t* taosGetErrno(); ...@@ -632,6 +632,7 @@ int32_t* taosGetErrno();
#define TSDB_CODE_PAR_INVALID_FIRST_COLUMN TAOS_DEF_ERROR_CODE(0, 0x2641) #define TSDB_CODE_PAR_INVALID_FIRST_COLUMN TAOS_DEF_ERROR_CODE(0, 0x2641)
#define TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN TAOS_DEF_ERROR_CODE(0, 0x2642) #define TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN TAOS_DEF_ERROR_CODE(0, 0x2642)
#define TSDB_CODE_PAR_INVALID_TAGS_NUM TAOS_DEF_ERROR_CODE(0, 0x2643) #define TSDB_CODE_PAR_INVALID_TAGS_NUM TAOS_DEF_ERROR_CODE(0, 0x2643)
#define TSDB_CODE_PAR_PERMISSION_DENIED TAOS_DEF_ERROR_CODE(0, 0x2644)
//planner //planner
#define TSDB_CODE_PLAN_INTERNAL_ERROR TAOS_DEF_ERROR_CODE(0, 0x2700) #define TSDB_CODE_PLAN_INTERNAL_ERROR TAOS_DEF_ERROR_CODE(0, 0x2700)
......
include/util/tencode.h
浏览文件 @ 9025a5d7
此差异已折叠。
source/client/src/clientImpl.c
浏览文件 @ 9025a5d7
...@@ -162,18 +162,17 @@ int32_t buildRequest(STscObj* pTscObj, const char* sql, int sqlLen, SRequestObj* ...@@ -162,18 +162,17 @@ int32_t buildRequest(STscObj* pTscObj, const char* sql, int sqlLen, SRequestObj*
int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery, SStmtCallback* pStmtCb) { int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery, SStmtCallback* pStmtCb) {
STscObj* pTscObj = pRequest->pTscObj; STscObj* pTscObj = pRequest->pTscObj;
SParseContext cxt = { SParseContext cxt = {.requestId = pRequest->requestId,
.requestId = pRequest->requestId, .acctId = pTscObj->acctId,
.acctId = pTscObj->acctId, .db = pRequest->pDb,
.db = pRequest->pDb, .topicQuery = topicQuery,
.topicQuery = topicQuery, .pSql = pRequest->sqlstr,
.pSql = pRequest->sqlstr, .sqlLen = pRequest->sqlLen,
.sqlLen = pRequest->sqlLen, .pMsg = pRequest->msgBuf,
.pMsg = pRequest->msgBuf, .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
.msgLen = ERROR_MSG_BUF_DEFAULT_SIZE, .pTransporter = pTscObj->pAppInfo->pTransporter,
.pTransporter = pTscObj->pAppInfo->pTransporter, .pStmtCb = pStmtCb,
.pStmtCb = pStmtCb, .pUser = pTscObj->user};
};
cxt.mgmtEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp); cxt.mgmtEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
int32_t code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &cxt.pCatalog); int32_t code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &cxt.pCatalog);
...@@ -232,11 +231,15 @@ int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArra ...@@ -232,11 +231,15 @@ int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArra
.mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp), .mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp),
.pAstRoot = pQuery->pRoot, .pAstRoot = pQuery->pRoot,
.showRewrite = pQuery->showRewrite, .showRewrite = pQuery->showRewrite,
.pTransporter = pRequest->pTscObj->pAppInfo->pTransporter,
.pMsg = pRequest->msgBuf, .pMsg = pRequest->msgBuf,
.msgLen = ERROR_MSG_BUF_DEFAULT_SIZE, .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
.placeholderNum = pQuery->placeholderNum}; .placeholderNum = pQuery->placeholderNum};
int32_t code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &cxt.pCatalog); SEpSet mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp);
SCatalog* pCatalog = NULL;
int32_t code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog);
if (TSDB_CODE_SUCCESS == code) {
code = catalogGetQnodeList(pCatalog, pRequest->pTscObj->pAppInfo->pTransporter, &mgmtEpSet, pNodeList);
}
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = qCreateQueryPlan(&cxt, pPlan, pNodeList); code = qCreateQueryPlan(&cxt, pPlan, pNodeList);
} }
......
source/client/src/tmq.c
浏览文件 @ 9025a5d7
...@@ -547,21 +547,21 @@ tmq_resp_err_t tmq_commit(tmq_t* tmq, const tmq_topic_vgroup_list_t* offsets, in ...@@ -547,21 +547,21 @@ tmq_resp_err_t tmq_commit(tmq_t* tmq, const tmq_topic_vgroup_list_t* offsets, in
req.offsets = (SMqOffset*)offsets->container.pData; req.offsets = (SMqOffset*)offsets->container.pData;
} }
SCoder encoder; SEncoder encoder;
tCoderInit(&encoder, TD_LITTLE_ENDIAN, NULL, 0, TD_ENCODER); tEncoderInit(&encoder, NULL, 0);
tEncodeSMqCMCommitOffsetReq(&encoder, &req); tEncodeSMqCMCommitOffsetReq(&encoder, &req);
int32_t tlen = encoder.pos; int32_t tlen = encoder.pos;
void* buf = taosMemoryMalloc(tlen); void* buf = taosMemoryMalloc(tlen);
if (buf == NULL) { if (buf == NULL) {
tCoderClear(&encoder); tEncoderClear(&encoder);
return -1; return -1;
} }
tCoderClear(&encoder); tEncoderClear(&encoder);
tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, tlen, TD_ENCODER); tEncoderInit(&encoder, buf, tlen);
tEncodeSMqCMCommitOffsetReq(&encoder, &req); tEncodeSMqCMCommitOffsetReq(&encoder, &req);
tCoderClear(&encoder); tEncoderClear(&encoder);
pRequest = createRequest(tmq->pTscObj, NULL, NULL, TDMT_MND_MQ_COMMIT_OFFSET); pRequest = createRequest(tmq->pTscObj, NULL, NULL, TDMT_MND_MQ_COMMIT_OFFSET);
if (pRequest == NULL) { if (pRequest == NULL) {
......
source/common/src/tmsg.c
浏览文件 @ 9025a5d7
此差异已折叠。
source/dnode/mnode/impl/inc/mndDef.h
浏览文件 @ 9025a5d7
...@@ -594,8 +594,8 @@ typedef struct { ...@@ -594,8 +594,8 @@ typedef struct {
SSchemaWrapper outputSchema; SSchemaWrapper outputSchema;
} SStreamObj; } SStreamObj;
int32_t tEncodeSStreamObj(SCoder* pEncoder, const SStreamObj* pObj); int32_t tEncodeSStreamObj(SEncoder* pEncoder, const SStreamObj* pObj);
int32_t tDecodeSStreamObj(SCoder* pDecoder, SStreamObj* pObj); int32_t tDecodeSStreamObj(SDecoder* pDecoder, SStreamObj* pObj);
#ifdef __cplusplus #ifdef __cplusplus
} }
......
source/dnode/mnode/impl/src/mndDef.c
浏览文件 @ 9025a5d7
...@@ -411,7 +411,7 @@ void *tDecodeSMqSubActionLogObj(const void *buf, SMqSubActionLogObj *pLog) { ...@@ -411,7 +411,7 @@ void *tDecodeSMqSubActionLogObj(const void *buf, SMqSubActionLogObj *pLog) {
return (void *)buf; return (void *)buf;
} }
int32_t tEncodeSStreamObj(SCoder *pEncoder, const SStreamObj *pObj) { int32_t tEncodeSStreamObj(SEncoder *pEncoder, const SStreamObj *pObj) {
int32_t sz = 0; int32_t sz = 0;
/*int32_t outputNameSz = 0;*/ /*int32_t outputNameSz = 0;*/
if (tEncodeCStr(pEncoder, pObj->name) < 0) return -1; if (tEncodeCStr(pEncoder, pObj->name) < 0) return -1;
...@@ -462,7 +462,7 @@ int32_t tEncodeSStreamObj(SCoder *pEncoder, const SStreamObj *pObj) { ...@@ -462,7 +462,7 @@ int32_t tEncodeSStreamObj(SCoder *pEncoder, const SStreamObj *pObj) {
return pEncoder->pos; return pEncoder->pos;
} }
int32_t tDecodeSStreamObj(SCoder *pDecoder, SStreamObj *pObj) { int32_t tDecodeSStreamObj(SDecoder *pDecoder, SStreamObj *pObj) {
if (tDecodeCStrTo(pDecoder, pObj->name) < 0) return -1; if (tDecodeCStrTo(pDecoder, pObj->name) < 0) return -1;
if (tDecodeCStrTo(pDecoder, pObj->sourceDb) < 0) return -1; if (tDecodeCStrTo(pDecoder, pObj->sourceDb) < 0) return -1;
if (tDecodeI64(pDecoder, &pObj->createTime) < 0) return -1; if (tDecodeI64(pDecoder, &pObj->createTime) < 0) return -1;
......
source/dnode/mnode/impl/src/mndOffset.c
浏览文件 @ 9025a5d7
...@@ -157,8 +157,8 @@ static int32_t mndProcessCommitOffsetReq(SNodeMsg *pMsg) { ...@@ -157,8 +157,8 @@ static int32_t mndProcessCommitOffsetReq(SNodeMsg *pMsg) {
SMnode *pMnode = pMsg->pNode; SMnode *pMnode = pMsg->pNode;
char *msgStr = pMsg->rpcMsg.pCont; char *msgStr = pMsg->rpcMsg.pCont;
SMqCMCommitOffsetReq commitOffsetReq; SMqCMCommitOffsetReq commitOffsetReq;
SCoder decoder; SDecoder decoder;
tCoderInit(&decoder, TD_LITTLE_ENDIAN, msgStr, pMsg->rpcMsg.contLen, TD_DECODER); tDecoderInit(&decoder, msgStr, pMsg->rpcMsg.contLen);
tDecodeSMqCMCommitOffsetReq(&decoder, &commitOffsetReq); tDecodeSMqCMCommitOffsetReq(&decoder, &commitOffsetReq);
......
source/dnode/mnode/impl/src/mndScheduler.c
浏览文件 @ 9025a5d7
...@@ -83,12 +83,12 @@ END: ...@@ -83,12 +83,12 @@ END:
} }
int32_t mndPersistTaskDeployReq(STrans* pTrans, SStreamTask* pTask, const SEpSet* pEpSet, tmsg_t type, int32_t nodeId) { int32_t mndPersistTaskDeployReq(STrans* pTrans, SStreamTask* pTask, const SEpSet* pEpSet, tmsg_t type, int32_t nodeId) {
SCoder encoder; SEncoder encoder;
tCoderInit(&encoder, TD_LITTLE_ENDIAN, NULL, 0, TD_ENCODER); tEncoderInit(&encoder, NULL, 0);
tEncodeSStreamTask(&encoder, pTask); tEncodeSStreamTask(&encoder, pTask);
int32_t size = encoder.pos; int32_t size = encoder.pos;
int32_t tlen = sizeof(SMsgHead) + size; int32_t tlen = sizeof(SMsgHead) + size;
tCoderClear(&encoder); tEncoderClear(&encoder);
void* buf = taosMemoryMalloc(tlen); void* buf = taosMemoryMalloc(tlen);
if (buf == NULL) { if (buf == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
...@@ -96,9 +96,9 @@ int32_t mndPersistTaskDeployReq(STrans* pTrans, SStreamTask* pTask, const SEpSet ...@@ -96,9 +96,9 @@ int32_t mndPersistTaskDeployReq(STrans* pTrans, SStreamTask* pTask, const SEpSet
} }
((SMsgHead*)buf)->vgId = htonl(nodeId); ((SMsgHead*)buf)->vgId = htonl(nodeId);
void* abuf = POINTER_SHIFT(buf, sizeof(SMsgHead)); void* abuf = POINTER_SHIFT(buf, sizeof(SMsgHead));
tCoderInit(&encoder, TD_LITTLE_ENDIAN, abuf, size, TD_ENCODER); tEncoderInit(&encoder, abuf, size);
tEncodeSStreamTask(&encoder, pTask); tEncodeSStreamTask(&encoder, pTask);
tCoderClear(&encoder); tEncoderClear(&encoder);
STransAction action = {0}; STransAction action = {0};
memcpy(&action.epSet, pEpSet, sizeof(SEpSet)); memcpy(&action.epSet, pEpSet, sizeof(SEpSet));
......
source/dnode/mnode/impl/src/mndStb.c
浏览文件 @ 9025a5d7
...@@ -369,7 +369,7 @@ static FORCE_INLINE int schemaExColIdCompare(const void *colId, const void *pSch ...@@ -369,7 +369,7 @@ static FORCE_INLINE int schemaExColIdCompare(const void *colId, const void *pSch
} }
static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb, int32_t *pContLen) { static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb, int32_t *pContLen) {
SCoder coder = {0}; SEncoder encoder = {0};
int32_t contLen; int32_t contLen;
SName name = {0}; SName name = {0};
SVCreateStbReq req = {0}; SVCreateStbReq req = {0};
...@@ -422,11 +422,11 @@ static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pSt ...@@ -422,11 +422,11 @@ static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pSt
pHead->vgId = htonl(pVgroup->vgId); pHead->vgId = htonl(pVgroup->vgId);
void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead)); void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));
tCoderInit(&coder, TD_LITTLE_ENDIAN, pBuf, contLen - sizeof(SMsgHead), TD_ENCODER); tEncoderInit(&encoder, pBuf, contLen - sizeof(SMsgHead));
if (tEncodeSVCreateStbReq(&coder, &req) < 0) { if (tEncodeSVCreateStbReq(&encoder, &req) < 0) {
return NULL; return NULL;
} }
tCoderClear(&coder); tEncoderClear(&encoder);
*pContLen = contLen; *pContLen = contLen;
taosMemoryFreeClear(req.pRSmaParam.qmsg1); taosMemoryFreeClear(req.pRSmaParam.qmsg1);
...@@ -440,7 +440,7 @@ static void *mndBuildVDropStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb, ...@@ -440,7 +440,7 @@ static void *mndBuildVDropStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb,
int32_t contLen = 0; int32_t contLen = 0;
int32_t ret = 0; int32_t ret = 0;
SMsgHead *pHead = NULL; SMsgHead *pHead = NULL;
SCoder coder = {0}; SEncoder encoder = {0};
tNameFromString(&name, pStb->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE); tNameFromString(&name, pStb->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
...@@ -462,9 +462,9 @@ static void *mndBuildVDropStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb, ...@@ -462,9 +462,9 @@ static void *mndBuildVDropStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb,
void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead)); void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));
tCoderInit(&coder, TD_LITTLE_ENDIAN, pBuf, contLen - sizeof(SMsgHead), TD_ENCODER); tEncoderInit(&encoder, pBuf, contLen - sizeof(SMsgHead));
tEncodeSVDropStbReq(&coder, &req); tEncodeSVDropStbReq(&encoder, &req);
tCoderClear(&coder); tEncoderClear(&encoder);
*pContLen = contLen; *pContLen = contLen;
return pHead; return pHead;
......
source/dnode/mnode/impl/src/mndStream.c
浏览文件 @ 9025a5d7
...@@ -70,14 +70,14 @@ SSdbRaw *mndStreamActionEncode(SStreamObj *pStream) { ...@@ -70,14 +70,14 @@ SSdbRaw *mndStreamActionEncode(SStreamObj *pStream) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
void *buf = NULL; void *buf = NULL;
SCoder encoder; SEncoder encoder;
tCoderInit(&encoder, TD_LITTLE_ENDIAN, NULL, 0, TD_ENCODER); tEncoderInit(&encoder, NULL, 0);
if (tEncodeSStreamObj(&encoder, pStream) < 0) { if (tEncodeSStreamObj(&encoder, pStream) < 0) {
tCoderClear(&encoder); tEncoderClear(&encoder);
goto STREAM_ENCODE_OVER; goto STREAM_ENCODE_OVER;
} }
int32_t tlen = encoder.pos; int32_t tlen = encoder.pos;
tCoderClear(&encoder); tEncoderClear(&encoder);
int32_t size = sizeof(int32_t) + tlen + MND_STREAM_RESERVE_SIZE; int32_t size = sizeof(int32_t) + tlen + MND_STREAM_RESERVE_SIZE;
SSdbRaw *pRaw = sdbAllocRaw(SDB_STREAM, MND_STREAM_VER_NUMBER, size); SSdbRaw *pRaw = sdbAllocRaw(SDB_STREAM, MND_STREAM_VER_NUMBER, size);
...@@ -86,12 +86,12 @@ SSdbRaw *mndStreamActionEncode(SStreamObj *pStream) { ...@@ -86,12 +86,12 @@ SSdbRaw *mndStreamActionEncode(SStreamObj *pStream) {
buf = taosMemoryMalloc(tlen); buf = taosMemoryMalloc(tlen);
if (buf == NULL) goto STREAM_ENCODE_OVER; if (buf == NULL) goto STREAM_ENCODE_OVER;
tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, tlen, TD_ENCODER); tEncoderInit(&encoder, buf, tlen);
if (tEncodeSStreamObj(&encoder, pStream) < 0) { if (tEncodeSStreamObj(&encoder, pStream) < 0) {
tCoderClear(&encoder); tEncoderClear(&encoder);
goto STREAM_ENCODE_OVER; goto STREAM_ENCODE_OVER;
} }
tCoderClear(&encoder); tEncoderClear(&encoder);
int32_t dataPos = 0; int32_t dataPos = 0;
SDB_SET_INT32(pRaw, dataPos, tlen, STREAM_ENCODE_OVER); SDB_SET_INT32(pRaw, dataPos, tlen, STREAM_ENCODE_OVER);
...@@ -138,8 +138,8 @@ SSdbRow *mndStreamActionDecode(SSdbRaw *pRaw) { ...@@ -138,8 +138,8 @@ SSdbRow *mndStreamActionDecode(SSdbRaw *pRaw) {
if (buf == NULL) goto STREAM_DECODE_OVER; if (buf == NULL) goto STREAM_DECODE_OVER;
SDB_GET_BINARY(pRaw, dataPos, buf, tlen, STREAM_DECODE_OVER); SDB_GET_BINARY(pRaw, dataPos, buf, tlen, STREAM_DECODE_OVER);
SCoder decoder; SDecoder decoder;
tCoderInit(&decoder, TD_LITTLE_ENDIAN, buf, tlen + 1, TD_DECODER); tDecoderInit(&decoder, buf, tlen + 1);
if (tDecodeSStreamObj(&decoder, pStream) < 0) { if (tDecodeSStreamObj(&decoder, pStream) < 0) {
goto STREAM_DECODE_OVER; goto STREAM_DECODE_OVER;
} }
......
source/dnode/mnode/impl/src/mndUser.c
浏览文件 @ 9025a5d7
...@@ -412,7 +412,7 @@ static int32_t mndProcessAlterUserReq(SNodeMsg *pReq) { ...@@ -412,7 +412,7 @@ static int32_t mndProcessAlterUserReq(SNodeMsg *pReq) {
goto _OVER; goto _OVER;
} }
if (alterReq.pass[0] == 0) { if (TSDB_ALTER_USER_PASSWD == alterReq.alterType && alterReq.pass[0] == 0) {
terrno = TSDB_CODE_MND_INVALID_PASS_FORMAT; terrno = TSDB_CODE_MND_INVALID_PASS_FORMAT;
goto _OVER; goto _OVER;
} }
......
source/dnode/snode/src/snode.c
浏览文件 @ 9025a5d7
...@@ -99,10 +99,10 @@ void sndProcessUMsg(SSnode *pSnode, SRpcMsg *pMsg) { ...@@ -99,10 +99,10 @@ void sndProcessUMsg(SSnode *pSnode, SRpcMsg *pMsg) {
ASSERT(0); ASSERT(0);
return; return;
} }
SCoder decoder; SDecoder decoder;
tCoderInit(&decoder, TD_LITTLE_ENDIAN, msg, pMsg->contLen - sizeof(SMsgHead), TD_DECODER); tDecoderInit(&decoder, msg, pMsg->contLen - sizeof(SMsgHead));
tDecodeSStreamTask(&decoder, pTask); tDecodeSStreamTask(&decoder, pTask);
tCoderClear(&decoder); tDecoderClear(&decoder);
sndMetaDeployTask(pSnode->pMeta, pTask); sndMetaDeployTask(pSnode->pMeta, pTask);
} else if (pMsg->msgType == TDMT_SND_TASK_EXEC) { } else if (pMsg->msgType == TDMT_SND_TASK_EXEC) {
......
source/dnode/vnode/inc/vnode.h
浏览文件 @ 9025a5d7
...@@ -90,7 +90,7 @@ int metaTbCursorNext(SMTbCursor *pTbCur); ...@@ -90,7 +90,7 @@ int metaTbCursorNext(SMTbCursor *pTbCur);
// tsdb // tsdb
// typedef struct STsdb STsdb; // typedef struct STsdb STsdb;
typedef void *tsdbReaderT; typedef void *tsdbReaderT;
#define BLOCK_LOAD_OFFSET_SEQ_ORDER 1 #define BLOCK_LOAD_OFFSET_SEQ_ORDER 1
#define BLOCK_LOAD_TABLE_SEQ_ORDER 2 #define BLOCK_LOAD_TABLE_SEQ_ORDER 2
...@@ -108,12 +108,12 @@ int32_t tsdbQuerySTableByTagCond(void *pMeta, uint64_t uid, TSKEY skey, con ...@@ -108,12 +108,12 @@ int32_t tsdbQuerySTableByTagCond(void *pMeta, uint64_t uid, TSKEY skey, con
int64_t tsdbGetNumOfRowsInMemTable(tsdbReaderT *pHandle); int64_t tsdbGetNumOfRowsInMemTable(tsdbReaderT *pHandle);
bool tsdbNextDataBlock(tsdbReaderT pTsdbReadHandle); bool tsdbNextDataBlock(tsdbReaderT pTsdbReadHandle);
void tsdbRetrieveDataBlockInfo(tsdbReaderT *pTsdbReadHandle, SDataBlockInfo *pBlockInfo); void tsdbRetrieveDataBlockInfo(tsdbReaderT *pTsdbReadHandle, SDataBlockInfo *pBlockInfo);
int32_t tsdbRetrieveDataBlockStatisInfo(tsdbReaderT *pTsdbReadHandle, SColumnDataAgg ***pBlockStatis, bool* allHave); int32_t tsdbRetrieveDataBlockStatisInfo(tsdbReaderT *pTsdbReadHandle, SColumnDataAgg ***pBlockStatis, bool *allHave);
SArray *tsdbRetrieveDataBlock(tsdbReaderT *pTsdbReadHandle, SArray *pColumnIdList); SArray *tsdbRetrieveDataBlock(tsdbReaderT *pTsdbReadHandle, SArray *pColumnIdList);
void tsdbResetReadHandle(tsdbReaderT queryHandle, SQueryTableDataCond *pCond); void tsdbResetReadHandle(tsdbReaderT queryHandle, SQueryTableDataCond *pCond);
void tsdbDestroyTableGroup(STableGroupInfo *pGroupList); void tsdbDestroyTableGroup(STableGroupInfo *pGroupList);
int32_t tsdbGetOneTableGroup(void *pMeta, uint64_t uid, TSKEY startKey, STableGroupInfo *pGroupInfo); int32_t tsdbGetOneTableGroup(void *pMeta, uint64_t uid, TSKEY startKey, STableGroupInfo *pGroupInfo);
int32_t tsdbGetTableGroupFromIdList(SVnode *pVnode, SArray *pTableIdList, STableGroupInfo *pGroupInfo); int32_t tsdbGetTableGroupFromIdList(SVnode *pVnode, SArray *pTableIdList, STableGroupInfo *pGroupInfo);
// tq // tq
...@@ -126,8 +126,8 @@ int tqReadHandleSetTbUidList(STqReadHandle *pHandle, const SArray *tbUidList ...@@ -126,8 +126,8 @@ int tqReadHandleSetTbUidList(STqReadHandle *pHandle, const SArray *tbUidList
int tqReadHandleAddTbUidList(STqReadHandle *pHandle, const SArray *tbUidList); int tqReadHandleAddTbUidList(STqReadHandle *pHandle, const SArray *tbUidList);
int32_t tqReadHandleSetMsg(STqReadHandle *pHandle, SSubmitReq *pMsg, int64_t ver); int32_t tqReadHandleSetMsg(STqReadHandle *pHandle, SSubmitReq *pMsg, int64_t ver);
bool tqNextDataBlock(STqReadHandle *pHandle); bool tqNextDataBlock(STqReadHandle *pHandle);
int32_t tqRetrieveDataBlock(SArray **ppCols, STqReadHandle *pHandle, uint64_t *pGroupId, uint64_t* pUid, int32_t *pNumOfRows, int32_t tqRetrieveDataBlock(SArray **ppCols, STqReadHandle *pHandle, uint64_t *pGroupId, uint64_t *pUid,
int16_t *pNumOfCols); int32_t *pNumOfRows, int16_t *pNumOfCols);
// need to reposition // need to reposition
...@@ -189,10 +189,10 @@ struct SMetaEntry { ...@@ -189,10 +189,10 @@ struct SMetaEntry {
SSchemaWrapper schemaTag; SSchemaWrapper schemaTag;
} stbEntry; } stbEntry;
struct { struct {
int64_t ctime; int64_t ctime;
int32_t ttlDays; int32_t ttlDays;
tb_uid_t suid; tb_uid_t suid;
const void *pTags; const uint8_t *pTags;
} ctbEntry; } ctbEntry;
struct { struct {
int64_t ctime; int64_t ctime;
...@@ -205,7 +205,7 @@ struct SMetaEntry { ...@@ -205,7 +205,7 @@ struct SMetaEntry {
struct SMetaReader { struct SMetaReader {
int32_t flags; int32_t flags;
SMeta *pMeta; SMeta *pMeta;
SCoder coder; SDecoder coder;
SMetaEntry me; SMetaEntry me;
void *pBuf; void *pBuf;
int szBuf; int szBuf;
......
source/dnode/vnode/src/inc/meta.h
浏览文件 @ 9025a5d7
...@@ -39,8 +39,8 @@ typedef struct SMSmaCursor SMSmaCursor; ...@@ -39,8 +39,8 @@ typedef struct SMSmaCursor SMSmaCursor;
// metaOpen ================== // metaOpen ==================
// metaEntry ================== // metaEntry ==================
int metaEncodeEntry(SCoder* pCoder, const SMetaEntry* pME); int metaEncodeEntry(SEncoder* pCoder, const SMetaEntry* pME);
int metaDecodeEntry(SCoder* pCoder, SMetaEntry* pME); int metaDecodeEntry(SDecoder* pCoder, SMetaEntry* pME);
// metaTable ================== // metaTable ==================
......
source/dnode/vnode/src/meta/metaEntry.c
浏览文件 @ 9025a5d7
...@@ -15,7 +15,7 @@ ...@@ -15,7 +15,7 @@
#include "meta.h" #include "meta.h"
int metaEncodeEntry(SCoder *pCoder, const SMetaEntry *pME) { int metaEncodeEntry(SEncoder *pCoder, const SMetaEntry *pME) {
if (tStartEncode(pCoder) < 0) return -1; if (tStartEncode(pCoder) < 0) return -1;
if (tEncodeI64(pCoder, pME->version) < 0) return -1; if (tEncodeI64(pCoder, pME->version) < 0) return -1;
...@@ -43,8 +43,8 @@ int metaEncodeEntry(SCoder *pCoder, const SMetaEntry *pME) { ...@@ -43,8 +43,8 @@ int metaEncodeEntry(SCoder *pCoder, const SMetaEntry *pME) {
return 0; return 0;
} }
int metaDecodeEntry(SCoder *pCoder, SMetaEntry *pME) { int metaDecodeEntry(SDecoder *pCoder, SMetaEntry *pME) {
uint64_t len; uint32_t len;
if (tStartDecode(pCoder) < 0) return -1; if (tStartDecode(pCoder) < 0) return -1;
if (tDecodeI64(pCoder, &pME->version) < 0) return -1; if (tDecodeI64(pCoder, &pME->version) < 0) return -1;
......
source/dnode/vnode/src/meta/metaQuery.c
浏览文件 @ 9025a5d7
...@@ -22,7 +22,7 @@ void metaReaderInit(SMetaReader *pReader, SMeta *pMeta, int32_t flags) { ...@@ -22,7 +22,7 @@ void metaReaderInit(SMetaReader *pReader, SMeta *pMeta, int32_t flags) {
} }
void metaReaderClear(SMetaReader *pReader) { void metaReaderClear(SMetaReader *pReader) {
tCoderClear(&pReader->coder); tDecoderClear(&pReader->coder);
tdbFree(pReader->pBuf); tdbFree(pReader->pBuf);
} }
...@@ -37,7 +37,7 @@ int metaGetTableEntryByVersion(SMetaReader *pReader, int64_t version, tb_uid_t u ...@@ -37,7 +37,7 @@ int metaGetTableEntryByVersion(SMetaReader *pReader, int64_t version, tb_uid_t u
} }
// decode the entry // decode the entry
tCoderInit(&pReader->coder, TD_LITTLE_ENDIAN, pReader->pBuf, pReader->szBuf, TD_DECODER); tDecoderInit(&pReader->coder, pReader->pBuf, pReader->szBuf);
if (metaDecodeEntry(&pReader->coder, &pReader->me) < 0) { if (metaDecodeEntry(&pReader->coder, &pReader->me) < 0) {
goto _err; goto _err;
...@@ -147,7 +147,7 @@ SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, boo ...@@ -147,7 +147,7 @@ SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, boo
SSchemaWrapper *pSW = NULL; SSchemaWrapper *pSW = NULL;
SSchema *pSchema = NULL; SSchema *pSchema = NULL;
void *pBuf; void *pBuf;
SCoder coder = {0}; SDecoder coder = {0};
// fetch // fetch
skmDbKey.uid = uid; skmDbKey.uid = uid;
...@@ -163,11 +163,11 @@ SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, boo ...@@ -163,11 +163,11 @@ SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, boo
pBuf = pVal; pBuf = pVal;
pSW = taosMemoryMalloc(sizeof(SSchemaWrapper)); pSW = taosMemoryMalloc(sizeof(SSchemaWrapper));
tCoderInit(&coder, TD_LITTLE_ENDIAN, pVal, vLen, TD_DECODER); tDecoderInit(&coder, pVal, vLen);
tDecodeSSchemaWrapper(&coder, pSW); tDecodeSSchemaWrapper(&coder, pSW);
pSchema = taosMemoryMalloc(sizeof(SSchema) * pSW->nCols); pSchema = taosMemoryMalloc(sizeof(SSchema) * pSW->nCols);
memcpy(pSchema, pSW->pSchema, sizeof(SSchema) * pSW->nCols); memcpy(pSchema, pSW->pSchema, sizeof(SSchema) * pSW->nCols);
tCoderClear(&coder); tDecoderClear(&coder);
pSW->pSchema = pSchema; pSW->pSchema = pSchema;
......
source/dnode/vnode/src/meta/metaTDBImpl.c 已删除 100644 → 0
浏览文件 @ 0bdabe6a
此差异已折叠。
source/dnode/vnode/src/meta/metaTable.c
浏览文件 @ 9025a5d7
...@@ -33,7 +33,6 @@ int metaCreateSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) { ...@@ -33,7 +33,6 @@ int metaCreateSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
void *pBuf = NULL; void *pBuf = NULL;
int32_t szBuf = 0; int32_t szBuf = 0;
void *p = NULL; void *p = NULL;
SCoder coder = {0};
SMetaReader mr = {0}; SMetaReader mr = {0};
// validate req // validate req
...@@ -192,7 +191,7 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq) { ...@@ -192,7 +191,7 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq) {
tb_uid_t uid; tb_uid_t uid;
int64_t tver; int64_t tver;
SMetaEntry me = {0}; SMetaEntry me = {0};
SCoder coder = {0}; SDecoder coder = {0};
int8_t type; int8_t type;
int64_t ctime; int64_t ctime;
tb_uid_t suid; tb_uid_t suid;
...@@ -253,7 +252,7 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq) { ...@@ -253,7 +252,7 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq) {
// decode entry // decode entry
void *pDataCopy = taosMemoryMalloc(nData); // remove the copy (todo) void *pDataCopy = taosMemoryMalloc(nData); // remove the copy (todo)
memcpy(pDataCopy, pData, nData); memcpy(pDataCopy, pData, nData);
tCoderInit(&coder, TD_LITTLE_ENDIAN, pDataCopy, nData, TD_DECODER); tDecoderInit(&coder, pDataCopy, nData);
ret = metaDecodeEntry(&coder, &me); ret = metaDecodeEntry(&coder, &me);
if (ret < 0) { if (ret < 0) {
ASSERT(0); ASSERT(0);
...@@ -272,7 +271,7 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq) { ...@@ -272,7 +271,7 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq) {
} }
taosMemoryFree(pDataCopy); taosMemoryFree(pDataCopy);
tCoderClear(&coder); tDecoderClear(&coder);
tdbDbcClose(pTbDbc); tdbDbcClose(pTbDbc);
if (type == TSDB_CHILD_TABLE) { if (type == TSDB_CHILD_TABLE) {
...@@ -309,7 +308,7 @@ static int metaSaveToTbDb(SMeta *pMeta, const SMetaEntry *pME) { ...@@ -309,7 +308,7 @@ static int metaSaveToTbDb(SMeta *pMeta, const SMetaEntry *pME) {
void *pVal = NULL; void *pVal = NULL;
int kLen = 0; int kLen = 0;
int vLen = 0; int vLen = 0;
SCoder coder = {0}; SEncoder coder = {0};
// set key and value // set key and value
tbDbKey.version = pME->version; tbDbKey.version = pME->version;
...@@ -330,13 +329,13 @@ static int metaSaveToTbDb(SMeta *pMeta, const SMetaEntry *pME) { ...@@ -330,13 +329,13 @@ static int metaSaveToTbDb(SMeta *pMeta, const SMetaEntry *pME) {
goto _err; goto _err;
} }
tCoderInit(&coder, TD_LITTLE_ENDIAN, pVal, vLen, TD_ENCODER); tEncoderInit(&coder, pVal, vLen);
if (metaEncodeEntry(&coder, pME) < 0) { if (metaEncodeEntry(&coder, pME) < 0) {
goto _err; goto _err;
} }
tCoderClear(&coder); tEncoderClear(&coder);
// write to table.db // write to table.db
if (tdbDbInsert(pMeta->pTbDb, pKey, kLen, pVal, vLen, &pMeta->txn) < 0) { if (tdbDbInsert(pMeta->pTbDb, pKey, kLen, pVal, vLen, &pMeta->txn) < 0) {
...@@ -393,7 +392,7 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pME) { ...@@ -393,7 +392,7 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pME) {
} }
static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) { static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) {
SCoder coder = {0}; SEncoder coder = {0};
void *pVal = NULL; void *pVal = NULL;
int vLen = 0; int vLen = 0;
int rcode = 0; int rcode = 0;
...@@ -422,7 +421,7 @@ static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) { ...@@ -422,7 +421,7 @@ static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) {
goto _exit; goto _exit;
} }
tCoderInit(&coder, TD_LITTLE_ENDIAN, pVal, vLen, TD_ENCODER); tEncoderInit(&coder, pVal, vLen);
tEncodeSSchemaWrapper(&coder, pSW); tEncodeSSchemaWrapper(&coder, pSW);
if (tdbDbInsert(pMeta->pSkmDb, &skmDbKey, sizeof(skmDbKey), pVal, vLen, &pMeta->txn) < 0) { if (tdbDbInsert(pMeta->pSkmDb, &skmDbKey, sizeof(skmDbKey), pVal, vLen, &pMeta->txn) < 0) {
...@@ -432,7 +431,7 @@ static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) { ...@@ -432,7 +431,7 @@ static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) {
_exit: _exit:
taosMemoryFree(pVal); taosMemoryFree(pVal);
tCoderClear(&coder); tEncoderClear(&coder);
return rcode; return rcode;
} }
......
source/dnode/vnode/src/tq/tq.c
浏览文件 @ 9025a5d7
...@@ -910,12 +910,12 @@ int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) { ...@@ -910,12 +910,12 @@ int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) {
if (pTask == NULL) { if (pTask == NULL) {
return -1; return -1;
} }
SCoder decoder; SDecoder decoder;
tCoderInit(&decoder, TD_LITTLE_ENDIAN, (uint8_t*)msg, msgLen, TD_DECODER); tDecoderInit(&decoder, (uint8_t*)msg, msgLen);
if (tDecodeSStreamTask(&decoder, pTask) < 0) { if (tDecodeSStreamTask(&decoder, pTask) < 0) {
ASSERT(0); ASSERT(0);
} }
tCoderClear(&decoder); tDecoderClear(&decoder);
// exec // exec
if (tqExpandTask(pTq, pTask, 4) < 0) { if (tqExpandTask(pTq, pTask, 4) < 0) {
......
source/dnode/vnode/src/tsdb/tsdbMemTable.c
浏览文件 @ 9025a5d7
...@@ -249,8 +249,10 @@ int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey ...@@ -249,8 +249,10 @@ int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey
pMergeInfo->keyFirst = TMIN(pMergeInfo->keyFirst, rowKey); pMergeInfo->keyFirst = TMIN(pMergeInfo->keyFirst, rowKey);
pMergeInfo->keyLast = TMAX(pMergeInfo->keyLast, rowKey); pMergeInfo->keyLast = TMAX(pMergeInfo->keyLast, rowKey);
lastKey = rowKey; lastKey = rowKey;
++pCols->numOfRows; if (pCols) {
tsdbAppendTableRowToCols(pTable, pCols, &pSchema, row, false); ++pCols->numOfRows;
tsdbAppendTableRowToCols(pTable, pCols, &pSchema, row, false);
}
} else { } else {
tsdbAppendTableRowToCols(pTable, pCols, &pSchema, row, true); tsdbAppendTableRowToCols(pTable, pCols, &pSchema, row, true);
} }
...@@ -279,7 +281,7 @@ int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey ...@@ -279,7 +281,7 @@ int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey
} }
#endif #endif
} }
if (lastKey != TSKEY_INITIAL_VAL) { if (pCols && (lastKey != TSKEY_INITIAL_VAL)) {
++pCols->numOfRows; ++pCols->numOfRows;
} }
......
source/dnode/vnode/src/tsdb/tsdbMemTable2.c
浏览文件 @ 9025a5d7
...@@ -63,14 +63,18 @@ struct SMemSkipListCurosr { ...@@ -63,14 +63,18 @@ struct SMemSkipListCurosr {
SMemSkipListNode *pNodeC; SMemSkipListNode *pNodeC;
}; };
#define HASH_BUCKET(SUID, UID, NBUCKET) (TABS((SUID) + (UID)) % (NBUCKET))
#define SL_NODE_SIZE(l) (sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l)*2) #define SL_NODE_SIZE(l) (sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l)*2)
#define SL_NODE_HALF_SIZE(l) (sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l)) #define SL_NODE_HALF_SIZE(l) (sizeof(SMemSkipListNode) + sizeof(SMemSkipListNode *) * (l))
#define SL_NODE_FORWARD(n, l) ((n)->forwards[l]) #define SL_NODE_FORWARD(n, l) ((n)->forwards[l])
#define SL_NODE_BACKWARD(n, l) ((n)->forwards[(n)->level + (l)]) #define SL_NODE_BACKWARD(n, l) ((n)->forwards[(n)->level + (l)])
#define SL_NODE_DATA(n) (&SL_NODE_BACKWARD(n, (n)->level)) #define SL_NODE_DATA(n) (&SL_NODE_BACKWARD(n, (n)->level))
#define SL_HEAD_NODE(sl) ((sl)->pHead) #define SL_HEAD_NODE(sl) ((sl)->pHead)
#define SL_TAIL_NODE(sl) ((SMemSkipListNode *)&SL_NODE_FORWARD(SL_HEAD_NODE(sl), (sl)->maxLevel)) #define SL_TAIL_NODE(sl) ((SMemSkipListNode *)&SL_NODE_FORWARD(SL_HEAD_NODE(sl), (sl)->maxLevel))
#define SL_HEAD_NODE_FORWARD(n, l) SL_NODE_FORWARD(n, l)
#define SL_TAIL_NODE_BACKWARD(n, l) SL_NODE_FORWARD(n, l)
// SMemTable // SMemTable
int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable **ppMemTb) { int32_t tsdbMemTableCreate2(STsdb *pTsdb, SMemTable **ppMemTb) {
...@@ -111,23 +115,18 @@ int32_t tsdbMemTableDestroy2(STsdb *pTsdb, SMemTable *pMemTb) { ...@@ -111,23 +115,18 @@ int32_t tsdbMemTableDestroy2(STsdb *pTsdb, SMemTable *pMemTb) {
} }
int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *pSubmitBlk) { int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *pSubmitBlk) {
SMemData *pMemData; SMemData *pMemData;
STsdb *pTsdb = pMemTb->pTsdb; STsdb *pTsdb = pMemTb->pTsdb;
SVnode *pVnode = pTsdb->pVnode; SVnode *pVnode = pTsdb->pVnode;
SVBufPool *pPool = pVnode->inUse; SVBufPool *pPool = pVnode->inUse;
int32_t hash; tb_uid_t suid = pSubmitBlk->suid;
int32_t tlen; tb_uid_t uid = pSubmitBlk->uid;
uint8_t buf[16]; int32_t iBucket;
int32_t rlen;
const uint8_t *p; // search SMemData by hash
SMemSkipListNode *pSlNode; iBucket = HASH_BUCKET(suid, uid, pMemTb->nBucket);
const STSRow *pTSRow; for (pMemData = pMemTb->pBuckets[iBucket]; pMemData; pMemData = pMemData->pHashNext) {
SMemSkipListCurosr slc = {0}; if (pMemData->suid == suid && pMemData->uid == uid) break;
// search hash
hash = (pSubmitBlk->suid + pSubmitBlk->uid) % pMemTb->nBucket;
for (pMemData = pMemTb->pBuckets[hash]; pMemData; pMemData = pMemData->pHashNext) {
if (pMemData->suid == pSubmitBlk->suid && pMemData->uid == pSubmitBlk->uid) break;
} }
// create pMemData if need // create pMemData if need
...@@ -143,8 +142,8 @@ int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *p ...@@ -143,8 +142,8 @@ int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *p
} }
pMemData->pHashNext = NULL; pMemData->pHashNext = NULL;
pMemData->suid = pSubmitBlk->suid; pMemData->suid = suid;
pMemData->uid = pSubmitBlk->uid; pMemData->uid = uid;
pMemData->minKey = TSKEY_MAX; pMemData->minKey = TSKEY_MAX;
pMemData->maxKey = TSKEY_MIN; pMemData->maxKey = TSKEY_MIN;
pMemData->minVer = -1; pMemData->minVer = -1;
...@@ -159,55 +158,67 @@ int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *p ...@@ -159,55 +158,67 @@ int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *p
pHead->level = maxLevel; pHead->level = maxLevel;
pTail->level = maxLevel; pTail->level = maxLevel;
for (int iLevel = 0; iLevel < maxLevel; iLevel++) { for (int iLevel = 0; iLevel < maxLevel; iLevel++) {
SL_NODE_FORWARD(pHead, iLevel) = pTail; SL_HEAD_NODE_FORWARD(pHead, iLevel) = pTail;
SL_NODE_FORWARD(pTail, iLevel) = pHead; SL_TAIL_NODE_BACKWARD(pTail, iLevel) = pHead;
} }
// add to MemTable // add to hash
hash = (pMemData->suid + pMemData->uid) % pMemTb->nBucket; if (pMemTb->nHash >= pMemTb->nBucket) {
pMemData->pHashNext = pMemTb->pBuckets[hash]; // rehash (todo)
pMemTb->pBuckets[hash] = pMemData; }
iBucket = HASH_BUCKET(suid, uid, pMemTb->nBucket);
pMemData->pHashNext = pMemTb->pBuckets[iBucket];
pMemTb->pBuckets[iBucket] = pMemData;
pMemTb->nHash++; pMemTb->nHash++;
// sort organize (todo)
} }
// loop to insert data to skiplist // do insert data to SMemData
#if 0 SMemSkipListCurosr slc = {0};
tsdbMemSkipListCursorOpen(&slc, &pMemData->sl); const uint8_t *p = pSubmitBlk->pData;
p = pSubmitBlk->pData; const uint8_t *pt;
for (;;) { const STSRow *pRow;
if (p - (uint8_t *)pSubmitBlk->pData >= pSubmitBlk->nData) break; uint64_t szRow;
SDecoder decoder = {0};
const uint8_t *pt = p; // tCoderInit(&coder, TD_LITTLE_ENDIAN, pSubmitBlk->pData, pSubmitBlk->nData, TD_DECODER);
p = tGetBinary(p, &pTSRow, &rlen); for (;;) {
// if (tDecodeIsEnd(&coder)) break;
// if (tDecodeBinary(&coder, (const uint8_t **)&pRow, &szRow) < 0) {
// terrno = TSDB_CODE_INVALID_MSG;
// return -1;
// }
// check the row (todo) // check the row (todo)
// move the cursor to position to write (todo) // // move the cursor to position to write (todo)
int32_t c; // int32_t c;
tsdbMemSkipListCursorMoveTo(&slc, pTSRow, version, &c); // tsdbMemSkipListCursorMoveTo(&slc, pTSRow, version, &c);
ASSERT(c); // ASSERT(c);
// encode row // // encode row
int8_t level = tsdbMemSkipListRandLevel(&pMemData->sl); // int8_t level = tsdbMemSkipListRandLevel(&pMemData->sl);
int32_t tsize = SL_NODE_SIZE(level) + sizeof(version) + (p - pt); // int32_t tsize = SL_NODE_SIZE(level) + sizeof(version) + (p - pt);
pSlNode = vnodeBufPoolMalloc(pPool, tsize); // pSlNode = vnodeBufPoolMalloc(pPool, tsize);
pSlNode->level = level; // pSlNode->level = level;
uint8_t *pData = SL_NODE_DATA(pSlNode); // uint8_t *pData = SL_NODE_DATA(pSlNode);
*(int64_t *)pData = version; // *(int64_t *)pData = version;
pData += sizeof(version); // pData += sizeof(version);
memcpy(pData, pt, p - pt); // memcpy(pData, pt, p - pt);
// insert row // // insert row
tsdbMemSkipListCursorPut(&slc, pSlNode); // tsdbMemSkipListCursorPut(&slc, pSlNode);
// update status // update status
if (pTSRow->ts < pMemData->minKey) pMemData->minKey = pTSRow->ts; if (pRow->ts < pMemData->minKey) pMemData->minKey = pRow->ts;
if (pTSRow->ts > pMemData->maxKey) pMemData->maxKey = pTSRow->ts; if (pRow->ts > pMemData->maxKey) pMemData->maxKey = pRow->ts;
} }
tsdbMemSkipListCursorClose(&slc); // tCoderClear(&coder);
#endif // tsdbMemSkipListCursorClose(&slc);
// update status
if (pMemData->minVer == -1) pMemData->minVer = version; if (pMemData->minVer == -1) pMemData->minVer = version;
if (pMemData->maxVer == -1 || pMemData->maxVer < version) pMemData->maxVer = version; if (pMemData->maxVer == -1 || pMemData->maxVer < version) pMemData->maxVer = version;
...@@ -217,8 +228,4 @@ int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *p ...@@ -217,8 +228,4 @@ int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *p
if (pMemTb->maxVer == -1 || pMemTb->maxVer < version) pMemTb->maxVer = version; if (pMemTb->maxVer == -1 || pMemTb->maxVer < version) pMemTb->maxVer = version;
return 0; return 0;
} }
\ No newline at end of file
// SMemData
// SMemSkipList
\ No newline at end of file
source/dnode/vnode/src/tsdb/tsdbSma.c
浏览文件 @ 9025a5d7
...@@ -1638,7 +1638,7 @@ int32_t tsdbCreateTSma(STsdb *pTsdb, char *pMsg) { ...@@ -1638,7 +1638,7 @@ int32_t tsdbCreateTSma(STsdb *pTsdb, char *pMsg) {
tsdbWarn("vgId:%d tsma create msg received but deserialize failed since %s", REPO_ID(pTsdb), terrstr(terrno)); tsdbWarn("vgId:%d tsma create msg received but deserialize failed since %s", REPO_ID(pTsdb), terrstr(terrno));
return -1; return -1;
} }
tsdbDebug("vgId:%d tsma create msg %s:%" PRIi64 " for table %" PRIi64 " received", REPO_ID(pTsdb), tsdbDebug("vgId:%d tsma create msg %s:%" PRIi64 " for table %" PRIi64 " received", REPO_ID(pTsdb),
vCreateSmaReq.tSma.indexName, vCreateSmaReq.tSma.indexUid, vCreateSmaReq.tSma.tableUid); vCreateSmaReq.tSma.indexName, vCreateSmaReq.tSma.indexUid, vCreateSmaReq.tSma.tableUid);
...@@ -2006,6 +2006,12 @@ static FORCE_INLINE int32_t tsdbExecuteRSmaImpl(STsdb *pTsdb, const void *pMsg, ...@@ -2006,6 +2006,12 @@ static FORCE_INLINE int32_t tsdbExecuteRSmaImpl(STsdb *pTsdb, const void *pMsg,
qTaskInfo_t *taskInfo, STSchema *pTSchema, tb_uid_t suid, tb_uid_t uid, qTaskInfo_t *taskInfo, STSchema *pTSchema, tb_uid_t suid, tb_uid_t uid,
int8_t level) { int8_t level) {
SArray *pResult = NULL; SArray *pResult = NULL;
if (!taskInfo) {
tsdbDebug("vgId:%d no qTaskInfo to execute rsma %" PRIi8 " task for suid:%" PRIu64, REPO_ID(pTsdb), level, suid);
return TSDB_CODE_SUCCESS;
}
tsdbDebug("vgId:%d execute rsma %" PRIi8 " task for qTaskInfo:%p suid:%" PRIu64, REPO_ID(pTsdb), level, taskInfo, tsdbDebug("vgId:%d execute rsma %" PRIi8 " task for qTaskInfo:%p suid:%" PRIu64, REPO_ID(pTsdb), level, taskInfo,
suid); suid);
...@@ -2071,10 +2077,18 @@ static int32_t tsdbExecuteRSma(STsdb *pTsdb, const void *pMsg, int32_t inputType ...@@ -2071,10 +2077,18 @@ static int32_t tsdbExecuteRSma(STsdb *pTsdb, const void *pMsg, int32_t inputType
tsdbDebug("vgId:%d no rsma info for suid:%" PRIu64, REPO_ID(pTsdb), suid); tsdbDebug("vgId:%d no rsma info for suid:%" PRIu64, REPO_ID(pTsdb), suid);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
if (!pRSmaInfo->taskInfo[0]) {
tsdbDebug("vgId:%d no rsma qTaskInfo for suid:%" PRIu64, REPO_ID(pTsdb), suid);
return TSDB_CODE_SUCCESS;
}
if (inputType == STREAM_DATA_TYPE_SUBMIT_BLOCK) { if (inputType == STREAM_DATA_TYPE_SUBMIT_BLOCK) {
// TODO: use the proper schema instead of 0, and cache STSchema in cache // TODO: use the proper schema instead of 0, and cache STSchema in cache
STSchema *pTSchema = metaGetTbTSchema(pTsdb->pVnode->pMeta, suid, 0); STSchema *pTSchema = metaGetTbTSchema(pTsdb->pVnode->pMeta, suid, 0);
if (!pTSchema) {
terrno = TSDB_CODE_TDB_IVD_TB_SCHEMA_VERSION;
return TSDB_CODE_FAILED;
}
tsdbExecuteRSmaImpl(pTsdb, pMsg, inputType, pRSmaInfo->taskInfo[0], pTSchema, suid, uid, TSDB_RETENTION_L1); tsdbExecuteRSmaImpl(pTsdb, pMsg, inputType, pRSmaInfo->taskInfo[0], pTSchema, suid, uid, TSDB_RETENTION_L1);
tsdbExecuteRSmaImpl(pTsdb, pMsg, inputType, pRSmaInfo->taskInfo[1], pTSchema, suid, uid, TSDB_RETENTION_L2); tsdbExecuteRSmaImpl(pTsdb, pMsg, inputType, pRSmaInfo->taskInfo[1], pTSchema, suid, uid, TSDB_RETENTION_L2);
taosMemoryFree(pTSchema); taosMemoryFree(pTSchema);
......
source/dnode/vnode/src/vnd/vnodeCfg.c
浏览文件 @ 9025a5d7
...@@ -25,7 +25,7 @@ const SVnodeCfg vnodeCfgDefault = { ...@@ -25,7 +25,7 @@ const SVnodeCfg vnodeCfgDefault = {
.isHeap = false, .isHeap = false,
.isWeak = 0, .isWeak = 0,
.tsdbCfg = {.precision = TSDB_TIME_PRECISION_MILLI, .tsdbCfg = {.precision = TSDB_TIME_PRECISION_MILLI,
.update = 0, .update = 1,
.compression = 2, .compression = 2,
.slLevel = 5, .slLevel = 5,
.days = 10, .days = 10,
......
source/dnode/vnode/src/vnd/vnodeSvr.c
浏览文件 @ 9025a5d7
...@@ -286,7 +286,7 @@ int vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) { ...@@ -286,7 +286,7 @@ int vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
static int vnodeProcessCreateStbReq(SVnode *pVnode, int64_t version, void *pReq, int len, SRpcMsg *pRsp) { static int vnodeProcessCreateStbReq(SVnode *pVnode, int64_t version, void *pReq, int len, SRpcMsg *pRsp) {
SVCreateStbReq req = {0}; SVCreateStbReq req = {0};
SCoder coder; SDecoder coder;
pRsp->msgType = TDMT_VND_CREATE_STB_RSP; pRsp->msgType = TDMT_VND_CREATE_STB_RSP;
pRsp->code = TSDB_CODE_SUCCESS; pRsp->code = TSDB_CODE_SUCCESS;
...@@ -294,7 +294,7 @@ static int vnodeProcessCreateStbReq(SVnode *pVnode, int64_t version, void *pReq, ...@@ -294,7 +294,7 @@ static int vnodeProcessCreateStbReq(SVnode *pVnode, int64_t version, void *pReq,
pRsp->contLen = 0; pRsp->contLen = 0;
// decode and process req // decode and process req
tCoderInit(&coder, TD_LITTLE_ENDIAN, pReq, len, TD_DECODER); tDecoderInit(&coder, pReq, len);
if (tDecodeSVCreateStbReq(&coder, &req) < 0) { if (tDecodeSVCreateStbReq(&coder, &req) < 0) {
pRsp->code = terrno; pRsp->code = terrno;
...@@ -308,16 +308,16 @@ static int vnodeProcessCreateStbReq(SVnode *pVnode, int64_t version, void *pReq, ...@@ -308,16 +308,16 @@ static int vnodeProcessCreateStbReq(SVnode *pVnode, int64_t version, void *pReq,
tsdbRegisterRSma(pVnode->pTsdb, pVnode->pMeta, &req); tsdbRegisterRSma(pVnode->pTsdb, pVnode->pMeta, &req);
tCoderClear(&coder); tDecoderClear(&coder);
return 0; return 0;
_err: _err:
tCoderClear(&coder); tDecoderClear(&coder);
return -1; return -1;
} }
static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq, int len, SRpcMsg *pRsp) { static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq, int len, SRpcMsg *pRsp) {
SCoder coder = {0}; SDecoder decoder = {0};
int rcode = 0; int rcode = 0;
SVCreateTbBatchReq req = {0}; SVCreateTbBatchReq req = {0};
SVCreateTbReq *pCreateReq; SVCreateTbReq *pCreateReq;
...@@ -332,8 +332,8 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq, ...@@ -332,8 +332,8 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq,
pRsp->contLen = 0; pRsp->contLen = 0;
// decode // decode
tCoderInit(&coder, TD_LITTLE_ENDIAN, pReq, len, TD_DECODER); tDecoderInit(&decoder, pReq, len);
if (tDecodeSVCreateTbBatchReq(&coder, &req) < 0) { if (tDecodeSVCreateTbBatchReq(&decoder, &req) < 0) {
rcode = -1; rcode = -1;
terrno = TSDB_CODE_INVALID_MSG; terrno = TSDB_CODE_INVALID_MSG;
goto _exit; goto _exit;
...@@ -373,13 +373,14 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq, ...@@ -373,13 +373,14 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq,
taosArrayPush(rsp.pArray, &cRsp); taosArrayPush(rsp.pArray, &cRsp);
} }
tCoderClear(&coder); tDecoderClear(&decoder);
tsdbUpdateTbUidList(pVnode->pTsdb, pStore); tsdbUpdateTbUidList(pVnode->pTsdb, pStore);
tsdbUidStoreFree(pStore); tsdbUidStoreFree(pStore);
// prepare rsp // prepare rsp
int32_t ret = 0; SEncoder encoder = {0};
int32_t ret = 0;
tEncodeSize(tEncodeSVCreateTbBatchRsp, &rsp, pRsp->contLen, ret); tEncodeSize(tEncodeSVCreateTbBatchRsp, &rsp, pRsp->contLen, ret);
pRsp->pCont = rpcMallocCont(pRsp->contLen); pRsp->pCont = rpcMallocCont(pRsp->contLen);
if (pRsp->pCont == NULL) { if (pRsp->pCont == NULL) {
...@@ -387,12 +388,14 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq, ...@@ -387,12 +388,14 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq,
rcode = -1; rcode = -1;
goto _exit; goto _exit;
} }
tCoderInit(&coder, TD_LITTLE_ENDIAN, pRsp->pCont, pRsp->contLen, TD_ENCODER); tEncoderInit(&encoder, pRsp->pCont, pRsp->contLen);
tEncodeSVCreateTbBatchRsp(&coder, &rsp); tEncodeSVCreateTbBatchRsp(&encoder, &rsp);
tEncoderClear(&encoder);
_exit: _exit:
taosArrayClear(rsp.pArray); taosArrayClear(rsp.pArray);
tCoderClear(&coder); tDecoderClear(&decoder);
tEncoderClear(&encoder);
return rcode; return rcode;
} }
...@@ -416,15 +419,15 @@ static int vnodeProcessAlterStbReq(SVnode *pVnode, void *pReq, int32_t len, SRpc ...@@ -416,15 +419,15 @@ static int vnodeProcessAlterStbReq(SVnode *pVnode, void *pReq, int32_t len, SRpc
static int vnodeProcessDropStbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) { static int vnodeProcessDropStbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
SVDropStbReq req = {0}; SVDropStbReq req = {0};
int rcode = TSDB_CODE_SUCCESS; int rcode = TSDB_CODE_SUCCESS;
SCoder coder = {0}; SDecoder decoder = {0};
pRsp->msgType = TDMT_VND_CREATE_STB_RSP; pRsp->msgType = TDMT_VND_CREATE_STB_RSP;
pRsp->pCont = NULL; pRsp->pCont = NULL;
pRsp->contLen = 0; pRsp->contLen = 0;
// decode request // decode request
tCoderInit(&coder, TD_LITTLE_ENDIAN, pReq, len, TD_DECODER); tDecoderInit(&decoder, pReq, len);
if (tDecodeSVDropStbReq(&coder, &req) < 0) { if (tDecodeSVDropStbReq(&decoder, &req) < 0) {
rcode = TSDB_CODE_INVALID_MSG; rcode = TSDB_CODE_INVALID_MSG;
goto _exit; goto _exit;
} }
...@@ -438,7 +441,7 @@ static int vnodeProcessDropStbReq(SVnode *pVnode, int64_t version, void *pReq, i ...@@ -438,7 +441,7 @@ static int vnodeProcessDropStbReq(SVnode *pVnode, int64_t version, void *pReq, i
// return rsp // return rsp
_exit: _exit:
pRsp->code = rcode; pRsp->code = rcode;
tCoderClear(&coder); tDecoderClear(&decoder);
return 0; return 0;
} }
...@@ -451,7 +454,7 @@ static int vnodeProcessAlterTbReq(SVnode *pVnode, void *pReq, int32_t len, SRpcM ...@@ -451,7 +454,7 @@ static int vnodeProcessAlterTbReq(SVnode *pVnode, void *pReq, int32_t len, SRpcM
static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) { static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
SVDropTbBatchReq req = {0}; SVDropTbBatchReq req = {0};
SVDropTbBatchRsp rsp = {0}; SVDropTbBatchRsp rsp = {0};
SCoder coder = {0}; SDecoder decoder = {0};
int ret; int ret;
pRsp->msgType = TDMT_VND_DROP_TABLE_RSP; pRsp->msgType = TDMT_VND_DROP_TABLE_RSP;
...@@ -460,8 +463,8 @@ static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, in ...@@ -460,8 +463,8 @@ static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, in
pRsp->code = TSDB_CODE_SUCCESS; pRsp->code = TSDB_CODE_SUCCESS;
// decode req // decode req
tCoderInit(&coder, TD_LITTLE_ENDIAN, pReq, len, TD_DECODER); tDecoderInit(&decoder, pReq, len);
ret = tDecodeSVDropTbBatchReq(&coder, &req); ret = tDecodeSVDropTbBatchReq(&decoder, &req);
if (ret < 0) { if (ret < 0) {
terrno = TSDB_CODE_INVALID_MSG; terrno = TSDB_CODE_INVALID_MSG;
pRsp->code = terrno; pRsp->code = terrno;
...@@ -490,7 +493,7 @@ static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, in ...@@ -490,7 +493,7 @@ static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, in
} }
_exit: _exit:
tCoderClear(&coder); tDecoderClear(&decoder);
// encode rsp (TODO) // encode rsp (TODO)
return 0; return 0;
} }
...@@ -501,7 +504,7 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in ...@@ -501,7 +504,7 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
SSubmitBlk *pBlock; SSubmitBlk *pBlock;
SSubmitRsp rsp = {0}; SSubmitRsp rsp = {0};
SVCreateTbReq createTbReq = {0}; SVCreateTbReq createTbReq = {0};
SCoder coder = {0}; SDecoder decoder = {0};
int32_t nRows; int32_t nRows;
pRsp->code = 0; pRsp->code = 0;
...@@ -518,17 +521,17 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in ...@@ -518,17 +521,17 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
// create table for auto create table mode // create table for auto create table mode
if (msgIter.schemaLen > 0) { if (msgIter.schemaLen > 0) {
tCoderInit(&coder, TD_LITTLE_ENDIAN, pBlock->data, msgIter.schemaLen, TD_DECODER); tDecoderInit(&decoder, pBlock->data, msgIter.schemaLen);
if (tDecodeSVCreateTbReq(&coder, &createTbReq) < 0) { if (tDecodeSVCreateTbReq(&decoder, &createTbReq) < 0) {
pRsp->code = TSDB_CODE_INVALID_MSG; pRsp->code = TSDB_CODE_INVALID_MSG;
tCoderClear(&coder); tDecoderClear(&decoder);
goto _exit; goto _exit;
} }
if (metaCreateTable(pVnode->pMeta, version, &createTbReq) < 0) { if (metaCreateTable(pVnode->pMeta, version, &createTbReq) < 0) {
if (terrno != TSDB_CODE_TDB_TABLE_ALREADY_EXIST) { if (terrno != TSDB_CODE_TDB_TABLE_ALREADY_EXIST) {
pRsp->code = terrno; pRsp->code = terrno;
tCoderClear(&coder); tDecoderClear(&decoder);
goto _exit; goto _exit;
} }
} }
...@@ -540,7 +543,7 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in ...@@ -540,7 +543,7 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
msgIter.suid = 0; msgIter.suid = 0;
} }
tCoderClear(&coder); tDecoderClear(&decoder);
} }
if (tsdbInsertTableData(pVnode->pTsdb, &msgIter, pBlock, &nRows) < 0) { if (tsdbInsertTableData(pVnode->pTsdb, &msgIter, pBlock, &nRows) < 0) {
......
source/libs/executor/CMakeLists.txt
浏览文件 @ 9025a5d7
此差异已折叠。
source/libs/executor/src/executorimpl.c
浏览文件 @ 9025a5d7
...@@ -198,6 +198,7 @@ SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode) { ...@@ -198,6 +198,7 @@ SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode) {
pBlock->info.blockId = pNode->dataBlockId; pBlock->info.blockId = pNode->dataBlockId;
pBlock->info.rowSize = pNode->totalRowSize; // todo ?? pBlock->info.rowSize = pNode->totalRowSize; // todo ??
pBlock->info.type = STREAM_INVALID;
for (int32_t i = 0; i < numOfCols; ++i) { for (int32_t i = 0; i < numOfCols; ++i) {
SColumnInfoData idata = {{0}}; SColumnInfoData idata = {{0}};
......
source/libs/function/inc/builtins.h
浏览文件 @ 9025a5d7
此差异已折叠。
source/libs/function/src/builtins.c
浏览文件 @ 9025a5d7
此差异已折叠。
source/libs/function/src/tudf.c
浏览文件 @ 9025a5d7
此差异已折叠。
source/libs/function/test/udf1.c
浏览文件 @ 9025a5d7
此差异已折叠。
source/libs/function/test/udf2.c
浏览文件 @ 9025a5d7
此差异已折叠。
source/libs/monitor/src/monMsg.c
浏览文件 @ 9025a5d7
此差异已折叠。
source/libs/parser/inc/parAst.h
浏览文件 @ 9025a5d7
此差异已折叠。
source/libs/parser/inc/parInt.h
浏览文件 @ 9025a5d7
此差异已折叠。
source/libs/parser/inc/sql.y
浏览文件 @ 9025a5d7
此差异已折叠。
source/libs/parser/src/parAuthenticator.c 0 → 100644
浏览文件 @ 9025a5d7
此差异已折叠。
source/libs/parser/src/parser.c
浏览文件 @ 9025a5d7
此差异已折叠。
source/libs/parser/src/sql.c
浏览文件 @ 9025a5d7
此差异已折叠。
source/libs/scalar/src/sclfunc.c
浏览文件 @ 9025a5d7
此差异已折叠。
source/libs/stream/src/tstream.c
浏览文件 @ 9025a5d7
此差异已折叠。
source/libs/sync/src/syncMessage.c
浏览文件 @ 9025a5d7
此差异已折叠。
source/os/src/osSocket.c
浏览文件 @ 9025a5d7
此差异已折叠。
source/util/src/tencode.c
浏览文件 @ 9025a5d7
此差异已折叠。
source/util/src/terror.c
浏览文件 @ 9025a5d7
此差异已折叠。
source/util/test/CMakeLists.txt
浏览文件 @ 9025a5d7
此差异已折叠。
source/util/test/encodeTest.cpp
浏览文件 @ 9025a5d7
此差异已折叠。
tests/script/tsim/query/udf.sim
浏览文件 @ 9025a5d7
此差异已折叠。
tests/system-test/2-query/diff.py 0 → 100644
浏览文件 @ 9025a5d7
此差异已折叠。
tests/system-test/2-query/last.py
浏览文件 @ 9025a5d7
此差异已折叠。
tests/system-test/2-query/query_cols_tags_and_or.py 0 → 100644
浏览文件 @ 9025a5d7
此差异已折叠。
tests/system-test/fulltest.sh
浏览文件 @ 9025a5d7
此差异已折叠。
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或