Skip to content

T
TDengine

taosdata / TDengine
12 个月 前同步成功

通知 1180
Star 22014
Fork 4786
T
TDengine

体验新版 GitCode，发现更多精彩内容 >>

提交 842b75cd 编写于 作者: H Haojun Liao
浏览文件
操作

Merge remote-tracking branch 'origin/3.0' into feature/3.0_liaohj

上级 ab390ece 56200d6c
展开全部 隐藏空白更改
内联 并排
Showing with 6687 addition and 4763 deletion
CMakeLists.txt
浏览文件 @ 842b75cd
......@@ -7,7 +7,7 @@ project(
)
if (NOT DEFINED TD_SOURCE_DIR)
set( TD_SOURCE_DIR ${CMAKE_SOURCE_DIR} )
set( TD_SOURCE_DIR ${PROJECT_SOURCE_DIR} )
endif()
set(TD_SUPPORT_DIR "${TD_SOURCE_DIR}/cmake")
......
example/src/tmq.c
浏览文件 @ 842b75cd
......@@ -60,7 +60,6 @@ int32_t init_env() {
pRes =
taos_query(pConn, "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(10)) tags(t1 int)");
if (taos_errno(pRes) != 0) {
assert(0);
printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes));
return -1;
}
......@@ -104,8 +103,8 @@ int32_t create_topic() {
}
taos_free_result(pRes);
/*pRes = taos_query(pConn, "create topic topic_ctb_column as abc1");*/
pRes = taos_query(pConn, "create topic topic_ctb_column as select ts, c1, c2, c3 from ct1");
pRes = taos_query(pConn, "create topic topic_ctb_column as abc1");
/*pRes = taos_query(pConn, "create topic topic_ctb_column as select ts, c1, c2, c3 from ct1");*/
if (taos_errno(pRes) != 0) {
printf("failed to create topic topic_ctb_column, reason:%s\n", taos_errstr(pRes));
return -1;
......@@ -163,9 +162,10 @@ tmq_t* build_consumer() {
tmq_conf_set(conf, "group.id", "tg2");
tmq_conf_set(conf, "td.connect.user", "root");
tmq_conf_set(conf, "td.connect.pass", "taosdata");
tmq_conf_set(conf, "td.connect.db", "abc1");
/*tmq_conf_set(conf, "td.connect.db", "abc1");*/
tmq_conf_set_offset_commit_cb(conf, tmq_commit_cb_print);
tmq_t* tmq = tmq_consumer_new(conf, NULL, 0);
assert(tmq);
return tmq;
}
......
examples/c/CMakeLists.txt
浏览文件 @ 842b75cd
PROJECT(TDengine)
IF (TD_LINUX)
INCLUDE_DIRECTORIES(. ${TD_COMMUNITY_DIR}/src/inc ${TD_COMMUNITY_DIR}/src/client/inc ${TD_COMMUNITY_DIR}/inc)
INCLUDE_DIRECTORIES(. ${TD_SOURCE_DIR}/src/inc ${TD_SOURCE_DIR}/src/client/inc ${TD_SOURCE_DIR}/inc)
AUX_SOURCE_DIRECTORY(. SRC)
ADD_EXECUTABLE(demo apitest.c)
TARGET_LINK_LIBRARIES(demo taos_static trpc tutil pthread )
......@@ -13,7 +13,7 @@ IF (TD_LINUX)
TARGET_LINK_LIBRARIES(epoll taos_static trpc tutil pthread lua)
ENDIF ()
IF (TD_DARWIN)
INCLUDE_DIRECTORIES(. ${TD_COMMUNITY_DIR}/src/inc ${TD_COMMUNITY_DIR}/src/client/inc ${TD_COMMUNITY_DIR}/inc)
INCLUDE_DIRECTORIES(. ${TD_SOURCE_DIR}/src/inc ${TD_SOURCE_DIR}/src/client/inc ${TD_SOURCE_DIR}/inc)
AUX_SOURCE_DIRECTORY(. SRC)
ADD_EXECUTABLE(demo demo.c)
TARGET_LINK_LIBRARIES(demo taos_static trpc tutil pthread lua)
......
include/client/taos.h
浏览文件 @ 842b75cd
......@@ -93,12 +93,12 @@ typedef struct taosField {
typedef void (*__taos_async_fn_t)(void *param, TAOS_RES *, int code);
typedef struct TAOS_BIND_v2 {
int buffer_type;
void *buffer;
int32_t buffer_length;
int32_t *length;
char *is_null;
int num;
int buffer_type;
void *buffer;
int32_t buffer_length;
int32_t *length;
char *is_null;
int num;
} TAOS_BIND_v2;
typedef enum {
......@@ -128,35 +128,35 @@ DLL_EXPORT void taos_close(TAOS *taos);
const char *taos_data_type(int type);
DLL_EXPORT TAOS_STMT *taos_stmt_init(TAOS *taos);
DLL_EXPORT int taos_stmt_prepare(TAOS_STMT *stmt, const char *sql, unsigned long length);
DLL_EXPORT int taos_stmt_set_tbname_tags(TAOS_STMT *stmt, const char *name, TAOS_BIND_v2 *tags);
DLL_EXPORT int taos_stmt_set_tbname(TAOS_STMT *stmt, const char *name);
DLL_EXPORT int taos_stmt_set_sub_tbname(TAOS_STMT *stmt, const char *name);
DLL_EXPORT int taos_stmt_is_insert(TAOS_STMT *stmt, int *insert);
DLL_EXPORT int taos_stmt_num_params(TAOS_STMT *stmt, int *nums);
DLL_EXPORT int taos_stmt_get_param(TAOS_STMT *stmt, int idx, int *type, int *bytes);
DLL_EXPORT int taos_stmt_bind_param(TAOS_STMT *stmt, TAOS_BIND_v2 *bind);
DLL_EXPORT int taos_stmt_bind_param_batch(TAOS_STMT *stmt, TAOS_BIND_v2 *bind);
DLL_EXPORT int taos_stmt_bind_single_param_batch(TAOS_STMT *stmt, TAOS_BIND_v2 *bind, int colIdx);
DLL_EXPORT int taos_stmt_add_batch(TAOS_STMT *stmt);
DLL_EXPORT int taos_stmt_execute(TAOS_STMT *stmt);
DLL_EXPORT TAOS_RES *taos_stmt_use_result(TAOS_STMT *stmt);
DLL_EXPORT int taos_stmt_close(TAOS_STMT *stmt);
DLL_EXPORT char *taos_stmt_errstr(TAOS_STMT *stmt);
DLL_EXPORT int taos_stmt_affected_rows(TAOS_STMT *stmt);
DLL_EXPORT int taos_stmt_affected_rows_once(TAOS_STMT *stmt);
DLL_EXPORT TAOS_RES *taos_query(TAOS *taos, const char *sql);
DLL_EXPORT TAOS_RES *taos_query_l(TAOS *taos, const char *sql, int sqlLen);
DLL_EXPORT TAOS_ROW taos_fetch_row(TAOS_RES *res);
DLL_EXPORT int taos_result_precision(TAOS_RES *res); // get the time precision of result
DLL_EXPORT void taos_free_result(TAOS_RES *res);
DLL_EXPORT int taos_field_count(TAOS_RES *res);
DLL_EXPORT int taos_num_fields(TAOS_RES *res);
DLL_EXPORT int taos_affected_rows(TAOS_RES *res);
DLL_EXPORT TAOS_STMT *taos_stmt_init(TAOS *taos);
DLL_EXPORT int taos_stmt_prepare(TAOS_STMT *stmt, const char *sql, unsigned long length);
DLL_EXPORT int taos_stmt_set_tbname_tags(TAOS_STMT *stmt, const char *name, TAOS_BIND_v2 *tags);
DLL_EXPORT int taos_stmt_set_tbname(TAOS_STMT *stmt, const char *name);
DLL_EXPORT int taos_stmt_set_sub_tbname(TAOS_STMT *stmt, const char *name);
DLL_EXPORT int taos_stmt_is_insert(TAOS_STMT *stmt, int *insert);
DLL_EXPORT int taos_stmt_num_params(TAOS_STMT *stmt, int *nums);
DLL_EXPORT int taos_stmt_get_param(TAOS_STMT *stmt, int idx, int *type, int *bytes);
DLL_EXPORT int taos_stmt_bind_param(TAOS_STMT *stmt, TAOS_BIND_v2 *bind);
DLL_EXPORT int taos_stmt_bind_param_batch(TAOS_STMT *stmt, TAOS_BIND_v2 *bind);
DLL_EXPORT int taos_stmt_bind_single_param_batch(TAOS_STMT *stmt, TAOS_BIND_v2 *bind, int colIdx);
DLL_EXPORT int taos_stmt_add_batch(TAOS_STMT *stmt);
DLL_EXPORT int taos_stmt_execute(TAOS_STMT *stmt);
DLL_EXPORT TAOS_RES *taos_stmt_use_result(TAOS_STMT *stmt);
DLL_EXPORT int taos_stmt_close(TAOS_STMT *stmt);
DLL_EXPORT char *taos_stmt_errstr(TAOS_STMT *stmt);
DLL_EXPORT int taos_stmt_affected_rows(TAOS_STMT *stmt);
DLL_EXPORT int taos_stmt_affected_rows_once(TAOS_STMT *stmt);
DLL_EXPORT TAOS_RES *taos_query(TAOS *taos, const char *sql);
DLL_EXPORT TAOS_RES *taos_query_l(TAOS *taos, const char *sql, int sqlLen);
DLL_EXPORT TAOS_ROW taos_fetch_row(TAOS_RES *res);
DLL_EXPORT int taos_result_precision(TAOS_RES *res); // get the time precision of result
DLL_EXPORT void taos_free_result(TAOS_RES *res);
DLL_EXPORT int taos_field_count(TAOS_RES *res);
DLL_EXPORT int taos_num_fields(TAOS_RES *res);
DLL_EXPORT int taos_affected_rows(TAOS_RES *res);
DLL_EXPORT TAOS_FIELD *taos_fetch_fields(TAOS_RES *res);
DLL_EXPORT int taos_select_db(TAOS *taos, const char *db);
......@@ -234,7 +234,7 @@ DLL_EXPORT tmq_t *tmq_consumer_new(tmq_conf_t *conf, char *errstr, int32_t errst
DLL_EXPORT const char *tmq_err2str(tmq_resp_err_t);
/* ------------------------TMQ CONSUMER INTERFACE------------------------ */
DLL_EXPORT tmq_resp_err_t tmq_subscribe(tmq_t *tmq, tmq_list_t *topic_list);
DLL_EXPORT tmq_resp_err_t tmq_subscribe(tmq_t *tmq, const tmq_list_t *topic_list);
DLL_EXPORT tmq_resp_err_t tmq_unsubscribe(tmq_t *tmq);
DLL_EXPORT tmq_resp_err_t tmq_subscription(tmq_t *tmq, tmq_list_t **topics);
DLL_EXPORT TAOS_RES *tmq_consumer_poll(tmq_t *tmq, int64_t blocking_time);
......
include/common/tglobal.h
浏览文件 @ 842b75cd
......@@ -123,9 +123,9 @@ extern SDiskCfg tsDiskCfg[];
#define NEEDTO_COMPRESSS_MSG(size) (tsCompressMsgSize != -1 && (size) > tsCompressMsgSize)
int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDir, const char *envFile,
const char *apolloUrl, SArray *pArgs, bool tsc);
int32_t taosInitCfg(const char *cfgDir, const char *envFile, const char *apolloUrl, SArray *pArgs, bool tsc);
int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDir, const char **envCmd, const char *envFile,
char *apolloUrl, SArray *pArgs, bool tsc);
int32_t taosInitCfg(const char *cfgDir, const char **envCmd, const char *envFile, char *apolloUrl, SArray *pArgs, bool tsc);
void taosCleanupCfg();
void taosCfgDynamicOptions(const char *option, const char *value);
void taosAddDataDir(int32_t index, char *v1, int32_t level, int32_t primary);
......
include/common/tmsg.h
浏览文件 @ 842b75cd
......@@ -227,8 +227,16 @@ typedef struct {
} SSubmitBlkIter;
typedef struct {
int32_t totalLen;
int32_t len;
int32_t totalLen;
int32_t len;
// head of SSubmitBlk
// int64_t uid; // table unique id
// int64_t suid; // stable id
// int32_t sversion; // data schema version
// int32_t dataLen; // data part length, not including the SSubmitBlk head
// int32_t schemaLen; // schema length, if length is 0, no schema exists
// int16_t numOfRows; // total number of rows in current submit block
// head of SSubmitBlk
const void* pMsg;
} SSubmitMsgIter;
......@@ -237,6 +245,15 @@ int32_t tGetSubmitMsgNext(SSubmitMsgIter* pIter, SSubmitBlk** pPBlock);
int32_t tInitSubmitBlkIter(SSubmitBlk* pBlock, SSubmitBlkIter* pIter);
STSRow* tGetSubmitBlkNext(SSubmitBlkIter* pIter);
// TODO: KEEP one suite of iterator API finally.
// 1) use tInitSubmitMsgIterEx firstly as not decrease the merge conflicts
// 2) replace tInitSubmitMsgIterEx with tInitSubmitMsgIter later
// 3) finally, rename tInitSubmitMsgIterEx to tInitSubmitMsgIter
// int32_t tInitSubmitMsgIterEx(const SSubmitReq* pMsg, SSubmitMsgIter* pIter);
// int32_t tGetSubmitMsgNextEx(SSubmitMsgIter* pIter, SSubmitBlk** pPBlock);
// int32_t tInitSubmitBlkIterEx(SSubmitMsgIter* pMsgIter, SSubmitBlk* pBlock, SSubmitBlkIter* pIter);
// STSRow* tGetSubmitBlkNextEx(SSubmitBlkIter* pIter);
typedef struct {
int32_t index; // index of failed block in submit blocks
int32_t vnode; // vnode index of failed block
......@@ -275,7 +292,6 @@ typedef struct {
char name[TSDB_TABLE_FNAME_LEN];
int8_t igExists;
float xFilesFactor;
int32_t aggregationMethod;
int32_t delay;
int32_t ttl;
int32_t numOfColumns;
......@@ -1316,7 +1332,7 @@ typedef struct {
typedef struct {
int64_t consumerId;
char cgroup[TSDB_CGROUP_LEN];
SArray* topicNames; // SArray<char*>
SArray* topicNames; // SArray<char**>
} SCMSubscribeReq;
static FORCE_INLINE int32_t tSerializeSCMSubscribeReq(void** buf, const SCMSubscribeReq* pReq) {
......@@ -1495,10 +1511,8 @@ typedef struct {
int32_t delay;
int32_t qmsg1Len;
int32_t qmsg2Len;
func_id_t* pFuncIds;
char* qmsg1; // not null: pAst1:qmsg1:SRetention1 => trigger aggr task1
char* qmsg2; // not null: pAst2:qmsg2:SRetention2 => trigger aggr task2
int8_t nFuncIds;
char* qmsg1; // pAst1:qmsg1:SRetention1 => trigger aggr task1
char* qmsg2; // pAst2:qmsg2:SRetention2 => trigger aggr task2
} SRSmaParam;
typedef struct SVCreateTbReq {
......@@ -2052,80 +2066,6 @@ static FORCE_INLINE void* tDecodeSMqSetCVgReq(void* buf, SMqSetCVgReq* pReq) {
return buf;
}
typedef struct {
int64_t leftForVer;
int32_t vgId;
int32_t epoch;
int64_t consumerId;
char topicName[TSDB_TOPIC_FNAME_LEN];
} SMqCancelConnReq;
static FORCE_INLINE int32_t tEncodeSMqCancelConnReq(void** buf, const SMqCancelConnReq* pReq) {
int32_t tlen = 0;
tlen += taosEncodeFixedI64(buf, pReq->leftForVer);
tlen += taosEncodeFixedI32(buf, pReq->vgId);
tlen += taosEncodeFixedI32(buf, pReq->epoch);
tlen += taosEncodeFixedI64(buf, pReq->consumerId);
tlen += taosEncodeString(buf, pReq->topicName);
return tlen;
}
static FORCE_INLINE void* tDecodeSMqCancelConnReq(void* buf, SMqCancelConnReq* pReq) {
buf = taosDecodeFixedI64(buf, &pReq->leftForVer);
buf = taosDecodeFixedI32(buf, &pReq->vgId);
buf = taosDecodeFixedI32(buf, &pReq->epoch);
buf = taosDecodeFixedI64(buf, &pReq->consumerId);
buf = taosDecodeStringTo(buf, pReq->topicName);
return buf;
}
typedef struct {
int8_t reserved;
} SMqCancelConnRsp;
typedef struct {
int64_t leftForVer;
int32_t vgId;
int64_t oldConsumerId;
int64_t newConsumerId;
char* topic;
} SMqMVRebReq;
static FORCE_INLINE int32_t tEncodeSMqMVRebReq(void** buf, const SMqMVRebReq* pReq) {
int32_t tlen = 0;
tlen += taosEncodeFixedI64(buf, pReq->leftForVer);
tlen += taosEncodeFixedI32(buf, pReq->vgId);
tlen += taosEncodeFixedI64(buf, pReq->oldConsumerId);
tlen += taosEncodeFixedI64(buf, pReq->newConsumerId);
tlen += taosEncodeString(buf, pReq->topic);
return tlen;
}
static FORCE_INLINE void* tDecodeSMqMVRebReq(void* buf, SMqMVRebReq* pReq) {
buf = taosDecodeFixedI64(buf, &pReq->leftForVer);
buf = taosDecodeFixedI32(buf, &pReq->vgId);
buf = taosDecodeFixedI64(buf, &pReq->oldConsumerId);
buf = taosDecodeFixedI64(buf, &pReq->newConsumerId);
buf = taosDecodeString(buf, &pReq->topic);
return buf;
}
typedef struct {
SMsgHead header;
int32_t vgId;
int64_t consumerId;
char topicName[TSDB_TOPIC_FNAME_LEN];
char cgroup[TSDB_CGROUP_LEN];
} SMqSetCVgRsp;
typedef struct {
SMsgHead header;
int32_t vgId;
int64_t consumerId;
char topicName[TSDB_TOPIC_FNAME_LEN];
char cgroup[TSDB_CGROUP_LEN];
} SMqMVRebRsp;
typedef struct {
int32_t vgId;
int64_t offset;
......@@ -2152,6 +2092,24 @@ typedef struct {
SSchema* pSchema;
} SSchemaWrapper;
static FORCE_INLINE SSchemaWrapper* tCloneSSchemaWrapper(const SSchemaWrapper* pSchemaWrapper) {
SSchemaWrapper* pSW = (SSchemaWrapper*)taosMemoryMalloc(sizeof(SSchemaWrapper));
if (pSW == NULL) return pSW;
pSW->nCols = pSchemaWrapper->nCols;
pSW->pSchema = (SSchema*)taosMemoryCalloc(pSW->nCols, sizeof(SSchema));
if (pSW->pSchema == NULL) {
taosMemoryFree(pSW);
return NULL;
}
memcpy(pSW->pSchema, pSchemaWrapper->pSchema, pSW->nCols * sizeof(SSchema));
return pSW;
}
static FORCE_INLINE void tDeleteSSchemaWrapper(SSchemaWrapper* pSchemaWrapper) {
taosMemoryFree(pSchemaWrapper->pSchema);
taosMemoryFree(pSchemaWrapper);
}
static FORCE_INLINE int32_t taosEncodeSSchema(void** buf, const SSchema* pSchema) {
int32_t tlen = 0;
tlen += taosEncodeFixedI8(buf, pSchema->type);
......@@ -2162,13 +2120,13 @@ static FORCE_INLINE int32_t taosEncodeSSchema(void** buf, const SSchema* pSchema
return tlen;
}
static FORCE_INLINE void* taosDecodeSSchema(void* buf, SSchema* pSchema) {
static FORCE_INLINE void* taosDecodeSSchema(const void* buf, SSchema* pSchema) {
buf = taosDecodeFixedI8(buf, &pSchema->type);
buf = taosDecodeFixedI8(buf, &pSchema->flags);
buf = taosDecodeFixedI32(buf, &pSchema->bytes);
buf = taosDecodeFixedI16(buf, &pSchema->colId);
buf = taosDecodeStringTo(buf, pSchema->name);
return buf;
return (void*)buf;
}
static FORCE_INLINE int32_t tEncodeSSchema(SCoder* pEncoder, const SSchema* pSchema) {
......@@ -2198,7 +2156,7 @@ static FORCE_INLINE int32_t taosEncodeSSchemaWrapper(void** buf, const SSchemaWr
return tlen;
}
static FORCE_INLINE void* taosDecodeSSchemaWrapper(void* buf, SSchemaWrapper* pSW) {
static FORCE_INLINE void* taosDecodeSSchemaWrapper(const void* buf, SSchemaWrapper* pSW) {
buf = taosDecodeFixedU32(buf, &pSW->nCols);
pSW->pSchema = (SSchema*)taosMemoryCalloc(pSW->nCols, sizeof(SSchema));
if (pSW->pSchema == NULL) {
......@@ -2208,7 +2166,7 @@ static FORCE_INLINE void* taosDecodeSSchemaWrapper(void* buf, SSchemaWrapper* pS
for (int32_t i = 0; i < pSW->nCols; i++) {
buf = taosDecodeSSchema(buf, &pSW->pSchema[i]);
}
return buf;
return (void*)buf;
}
static FORCE_INLINE int32_t tEncodeSSchemaWrapper(SCoder* pEncoder, const SSchemaWrapper* pSW) {
......@@ -2615,6 +2573,10 @@ static FORCE_INLINE int32_t tEncodeSMqDataBlkRsp(void** buf, const SMqDataBlkRsp
void* data = taosArrayGetP(pRsp->blockData, i);
tlen += taosEncodeFixedI32(buf, bLen);
tlen += taosEncodeBinary(buf, data, bLen);
if (pRsp->withSchema) {
SSchemaWrapper* pSW = (SSchemaWrapper*)taosArrayGetP(pRsp->blockSchema, i);
tlen += taosEncodeSSchemaWrapper(buf, pSW);
}
}
}
return tlen;
......@@ -2627,6 +2589,7 @@ static FORCE_INLINE void* tDecodeSMqDataBlkRsp(const void* buf, SMqDataBlkRsp* p
buf = taosDecodeFixedI32(buf, &pRsp->blockNum);
pRsp->blockData = taosArrayInit(pRsp->blockNum, sizeof(void*));
pRsp->blockDataLen = taosArrayInit(pRsp->blockNum, sizeof(void*));
pRsp->blockSchema = taosArrayInit(pRsp->blockNum, sizeof(void*));
if (pRsp->blockNum != 0) {
buf = taosDecodeFixedI8(buf, &pRsp->withTbName);
buf = taosDecodeFixedI8(buf, &pRsp->withSchema);
......@@ -2639,6 +2602,11 @@ static FORCE_INLINE void* tDecodeSMqDataBlkRsp(const void* buf, SMqDataBlkRsp* p
buf = taosDecodeBinary(buf, &data, bLen);
taosArrayPush(pRsp->blockDataLen, &bLen);
taosArrayPush(pRsp->blockData, &data);
if (pRsp->withSchema) {
SSchemaWrapper* pSW = (SSchemaWrapper*)taosMemoryMalloc(sizeof(SSchemaWrapper));
buf = taosDecodeSSchemaWrapper(buf, pSW);
taosArrayPush(pRsp->blockSchema, &pSW);
}
}
}
return (void*)buf;
......
include/common/tmsgdef.h
浏览文件 @ 842b75cd
......@@ -145,7 +145,7 @@ enum {
TD_DEF_MSG_TYPE(TDMT_MND_ALTER_TOPIC, "mnode-alter-topic", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_DROP_TOPIC, "mnode-drop-topic", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_SUBSCRIBE, "mnode-subscribe", SCMSubscribeReq, SCMSubscribeRsp)
TD_DEF_MSG_TYPE(TDMT_MND_GET_SUB_EP, "mnode-get-sub-ep", SMqCMGetSubEpReq, SMqCMGetSubEpRsp)
TD_DEF_MSG_TYPE(TDMT_MND_GET_SUB_EP, "mnode-mq-ask-ep", SMqCMGetSubEpReq, SMqCMGetSubEpRsp)
TD_DEF_MSG_TYPE(TDMT_MND_MQ_TIMER, "mnode-mq-tmr", SMTimerReq, SMTimerReq)
TD_DEF_MSG_TYPE(TDMT_MND_MQ_CONSUMER_LOST, "mnode-mq-consumer-lost", SMTimerReq, SMTimerReq)
TD_DEF_MSG_TYPE(TDMT_MND_MQ_DO_REBALANCE, "mnode-mq-do-rebalance", SMqDoRebalanceMsg, SMqDoRebalanceMsg)
......
include/libs/function/function.h
浏览文件 @ 842b75cd
......@@ -322,6 +322,20 @@ struct SUdfInfo;
void qAddUdfInfo(uint64_t id, struct SUdfInfo* pUdfInfo);
void qRemoveUdfInfo(uint64_t id, struct SUdfInfo* pUdfInfo);
/**
* create udfd proxy, called once in process that call setupUdf/callUdfxxx/teardownUdf
* @return error code
*/
int32_t udfcOpen();
/**
* destroy udfd proxy
* @return error code
*/
int32_t udfcClose();
typedef void *UdfcFuncHandle;
#ifdef __cplusplus
}
#endif
......
include/libs/function/functionMgt.h
浏览文件 @ 842b75cd
......@@ -125,15 +125,15 @@ typedef struct SFmGetFuncInfoParam {
struct SCatalog* pCtg;
void *pRpc;
const SEpSet* pMgmtEps;
char* pErrBuf;
int32_t errBufLen;
} SFmGetFuncInfoParam;
int32_t fmFuncMgtInit();
void fmFuncMgtDestroy();
int32_t fmGetFuncInfo(SFmGetFuncInfoParam* pParam, const char* pFuncName, int32_t* pFuncId, int32_t* pFuncType);
int32_t fmGetFuncResultType(SFunctionNode* pFunc, char* pErrBuf, int32_t len);
int32_t fmGetFuncInfo(SFmGetFuncInfoParam* pParam, SFunctionNode* pFunc);
bool fmIsAggFunc(int32_t funcId);
bool fmIsScalarFunc(int32_t funcId);
......
include/libs/nodes/querynodes.h
浏览文件 @ 842b75cd
......@@ -25,14 +25,16 @@ extern "C" {
#include "tvariant.h"
#define TABLE_TOTAL_COL_NUM(pMeta) ((pMeta)->tableInfo.numOfColumns + (pMeta)->tableInfo.numOfTags)
#define TABLE_META_SIZE(pMeta) (NULL == (pMeta) ? 0 : (sizeof(STableMeta) + TABLE_TOTAL_COL_NUM((pMeta)) * sizeof(SSchema)))
#define VGROUPS_INFO_SIZE(pInfo) (NULL == (pInfo) ? 0 : (sizeof(SVgroupsInfo) + (pInfo)->numOfVgroups * sizeof(SVgroupInfo)))
#define TABLE_META_SIZE(pMeta) \
(NULL == (pMeta) ? 0 : (sizeof(STableMeta) + TABLE_TOTAL_COL_NUM((pMeta)) * sizeof(SSchema)))
#define VGROUPS_INFO_SIZE(pInfo) \
(NULL == (pInfo) ? 0 : (sizeof(SVgroupsInfo) + (pInfo)->numOfVgroups * sizeof(SVgroupInfo)))
typedef struct SRawExprNode {
ENodeType nodeType;
char* p;
uint32_t n;
SNode* pNode;
char* p;
uint32_t n;
SNode* pNode;
} SRawExprNode;
typedef struct SDataType {
......@@ -45,170 +47,155 @@ typedef struct SDataType {
typedef struct SExprNode {
ENodeType type;
SDataType resType;
char aliasName[TSDB_COL_NAME_LEN];
SNodeList* pAssociationList;
char aliasName[TSDB_COL_NAME_LEN];
SArray* pAssociation;
} SExprNode;
typedef enum EColumnType {
COLUMN_TYPE_COLUMN = 1,
COLUMN_TYPE_TAG
} EColumnType;
typedef enum EColumnType { COLUMN_TYPE_COLUMN = 1, COLUMN_TYPE_TAG } EColumnType;
typedef struct SColumnNode {
SExprNode node; // QUERY_NODE_COLUMN
uint64_t tableId;
int8_t tableType;
col_id_t colId;
EColumnType colType; // column or tag
char dbName[TSDB_DB_NAME_LEN];
char tableName[TSDB_TABLE_NAME_LEN];
char tableAlias[TSDB_TABLE_NAME_LEN];
char colName[TSDB_COL_NAME_LEN];
SNode* pProjectRef;
int16_t dataBlockId;
int16_t slotId;
SExprNode node; // QUERY_NODE_COLUMN
uint64_t tableId;
int8_t tableType;
col_id_t colId;
EColumnType colType; // column or tag
char dbName[TSDB_DB_NAME_LEN];
char tableName[TSDB_TABLE_NAME_LEN];
char tableAlias[TSDB_TABLE_NAME_LEN];
char colName[TSDB_COL_NAME_LEN];
SNode* pProjectRef;
int16_t dataBlockId;
int16_t slotId;
} SColumnNode;
typedef struct STargetNode {
ENodeType type;
int16_t dataBlockId;
int16_t slotId;
SNode* pExpr;
int16_t dataBlockId;
int16_t slotId;
SNode* pExpr;
} STargetNode;
typedef struct SValueNode {
SExprNode node; // QUERY_NODE_VALUE
char* literal;
bool isDuration;
bool translate;
bool genByCalc;
int16_t placeholderNo;
SExprNode node; // QUERY_NODE_VALUE
char* literal;
bool isDuration;
bool translate;
int16_t placeholderNo;
union {
bool b;
int64_t i;
bool b;
int64_t i;
uint64_t u;
double d;
char* p;
double d;
char* p;
} datum;
char unit;
} SValueNode;
typedef struct SOperatorNode {
SExprNode node; // QUERY_NODE_OPERATOR
SExprNode node; // QUERY_NODE_OPERATOR
EOperatorType opType;
SNode* pLeft;
SNode* pRight;
SNode* pLeft;
SNode* pRight;
} SOperatorNode;
typedef struct SLogicConditionNode {
SExprNode node; // QUERY_NODE_LOGIC_CONDITION
SExprNode node; // QUERY_NODE_LOGIC_CONDITION
ELogicConditionType condType;
SNodeList* pParameterList;
SNodeList* pParameterList;
} SLogicConditionNode;
typedef struct SNodeListNode {
ENodeType type; // QUERY_NODE_NODE_LIST
SDataType dataType;
ENodeType type; // QUERY_NODE_NODE_LIST
SDataType dataType;
SNodeList* pNodeList;
} SNodeListNode;
typedef struct SFunctionNode {
SExprNode node; // QUERY_NODE_FUNCTION
char functionName[TSDB_FUNC_NAME_LEN];
int32_t funcId;
int32_t funcType;
SExprNode node; // QUERY_NODE_FUNCTION
char functionName[TSDB_FUNC_NAME_LEN];
int32_t funcId;
int32_t funcType;
SNodeList* pParameterList;
int32_t udfBufSize;
} SFunctionNode;
typedef struct STableNode {
SExprNode node;
char dbName[TSDB_DB_NAME_LEN];
char tableName[TSDB_TABLE_NAME_LEN];
char tableAlias[TSDB_TABLE_NAME_LEN];
uint8_t precision;
char dbName[TSDB_DB_NAME_LEN];
char tableName[TSDB_TABLE_NAME_LEN];
char tableAlias[TSDB_TABLE_NAME_LEN];
uint8_t precision;
} STableNode;
struct STableMeta;
typedef struct SRealTableNode {
STableNode table; // QUERY_NODE_REAL_TABLE
STableNode table; // QUERY_NODE_REAL_TABLE
struct STableMeta* pMeta;
SVgroupsInfo* pVgroupList;
char useDbName[TSDB_DB_NAME_LEN];
double ratio;
SVgroupsInfo* pVgroupList;
char qualDbName[TSDB_DB_NAME_LEN]; // SHOW qualDbName.TABLES
double ratio;
} SRealTableNode;
typedef struct STempTableNode {
STableNode table; // QUERY_NODE_TEMP_TABLE
SNode* pSubquery;
STableNode table; // QUERY_NODE_TEMP_TABLE
SNode* pSubquery;
} STempTableNode;
typedef enum EJoinType {
JOIN_TYPE_INNER = 1
} EJoinType;
typedef enum EJoinType { JOIN_TYPE_INNER = 1 } EJoinType;
typedef struct SJoinTableNode {
STableNode table; // QUERY_NODE_JOIN_TABLE
EJoinType joinType;
SNode* pLeft;
SNode* pRight;
SNode* pOnCond;
STableNode table; // QUERY_NODE_JOIN_TABLE
EJoinType joinType;
SNode* pLeft;
SNode* pRight;
SNode* pOnCond;
} SJoinTableNode;
typedef enum EGroupingSetType {
GP_TYPE_NORMAL = 1
} EGroupingSetType;
typedef enum EGroupingSetType { GP_TYPE_NORMAL = 1 } EGroupingSetType;
typedef struct SGroupingSetNode {
ENodeType type; // QUERY_NODE_GROUPING_SET
ENodeType type; // QUERY_NODE_GROUPING_SET
EGroupingSetType groupingSetType;
SNodeList* pParameterList;
SNodeList* pParameterList;
} SGroupingSetNode;
typedef enum EOrder {
ORDER_ASC = 1,
ORDER_DESC
} EOrder;
typedef enum EOrder { ORDER_ASC = 1, ORDER_DESC } EOrder;
typedef enum ENullOrder {
NULL_ORDER_DEFAULT = 1,
NULL_ORDER_FIRST,
NULL_ORDER_LAST
} ENullOrder;
typedef enum ENullOrder { NULL_ORDER_DEFAULT = 1, NULL_ORDER_FIRST, NULL_ORDER_LAST } ENullOrder;
typedef struct SOrderByExprNode {
ENodeType type; // QUERY_NODE_ORDER_BY_EXPR
SNode* pExpr;
EOrder order;
ENodeType type; // QUERY_NODE_ORDER_BY_EXPR
SNode* pExpr;
EOrder order;
ENullOrder nullOrder;
} SOrderByExprNode;
typedef struct SLimitNode {
ENodeType type; // QUERY_NODE_LIMIT
int64_t limit;
int64_t offset;
ENodeType type; // QUERY_NODE_LIMIT
int64_t limit;
int64_t offset;
} SLimitNode;
typedef struct SStateWindowNode {
ENodeType type; // QUERY_NODE_STATE_WINDOW
SNode* pCol; // timestamp primary key
SNode* pExpr;
ENodeType type; // QUERY_NODE_STATE_WINDOW
SNode* pCol; // timestamp primary key
SNode* pExpr;
} SStateWindowNode;
typedef struct SSessionWindowNode {
ENodeType type; // QUERY_NODE_SESSION_WINDOW
SColumnNode* pCol; // timestamp primary key
SValueNode* pGap; // gap between two session window(in microseconds)
ENodeType type; // QUERY_NODE_SESSION_WINDOW
SColumnNode* pCol; // timestamp primary key
SValueNode* pGap; // gap between two session window(in microseconds)
} SSessionWindowNode;
typedef struct SIntervalWindowNode {
ENodeType type; // QUERY_NODE_INTERVAL_WINDOW
SNode* pCol; // timestamp primary key
SNode* pInterval; // SValueNode
SNode* pOffset; // SValueNode
SNode* pSliding; // SValueNode
SNode* pFill;
ENodeType type; // QUERY_NODE_INTERVAL_WINDOW
SNode* pCol; // timestamp primary key
SNode* pInterval; // SValueNode
SNode* pOffset; // SValueNode
SNode* pSliding; // SValueNode
SNode* pFill;
} SIntervalWindowNode;
typedef enum EFillMode {
......@@ -221,42 +208,40 @@ typedef enum EFillMode {
} EFillMode;
typedef struct SFillNode {
ENodeType type; // QUERY_NODE_FILL
ENodeType type; // QUERY_NODE_FILL
EFillMode mode;
SNode* pValues; // SNodeListNode
SNode* pValues; // SNodeListNode
} SFillNode;
typedef struct SSelectStmt {
ENodeType type; // QUERY_NODE_SELECT_STMT
bool isDistinct;
SNodeList* pProjectionList;
SNode* pFromTable;
SNode* pWhere;
SNodeList* pPartitionByList;
SNode* pWindow;
SNodeList* pGroupByList; // SGroupingSetNode
SNode* pHaving;
SNodeList* pOrderByList; // SOrderByExprNode
ENodeType type; // QUERY_NODE_SELECT_STMT
bool isDistinct;
SNodeList* pProjectionList;
SNode* pFromTable;
SNode* pWhere;
SNodeList* pPartitionByList;
SNode* pWindow;
SNodeList* pGroupByList; // SGroupingSetNode
SNode* pHaving;
SNodeList* pOrderByList; // SOrderByExprNode
SLimitNode* pLimit;
SLimitNode* pSlimit;
char stmtName[TSDB_TABLE_NAME_LEN];
uint8_t precision;
bool isEmptyResult;
char stmtName[TSDB_TABLE_NAME_LEN];
uint8_t precision;
bool isEmptyResult;
bool hasAggFuncs;
} SSelectStmt;
typedef enum ESetOperatorType {
SET_OP_TYPE_UNION_ALL = 1,
SET_OP_TYPE_UNION
} ESetOperatorType;
typedef enum ESetOperatorType { SET_OP_TYPE_UNION_ALL = 1, SET_OP_TYPE_UNION } ESetOperatorType;
typedef struct SSetOperator {
ENodeType type; // QUERY_NODE_SET_OPERATOR
ENodeType type; // QUERY_NODE_SET_OPERATOR
ESetOperatorType opType;
SNodeList* pProjectionList;
SNode* pLeft;
SNode* pRight;
SNodeList* pOrderByList; // SOrderByExprNode
SNode* pLimit;
SNodeList* pProjectionList;
SNode* pLeft;
SNode* pRight;
SNodeList* pOrderByList; // SOrderByExprNode
SNode* pLimit;
} SSetOperator;
typedef enum ESqlClause {
......@@ -271,7 +256,6 @@ typedef enum ESqlClause {
SQL_CLAUSE_ORDER_BY
} ESqlClause;
typedef enum {
PAYLOAD_TYPE_KV = 0,
PAYLOAD_TYPE_RAW = 1,
......@@ -281,29 +265,29 @@ typedef struct SVgDataBlocks {
SVgroupInfo vg;
int32_t numOfTables; // number of tables in current submit block
uint32_t size;
char *pData; // SMsgDesc + SSubmitReq + SSubmitBlk + ...
char* pData; // SMsgDesc + SSubmitReq + SSubmitBlk + ...
} SVgDataBlocks;
typedef struct SVnodeModifOpStmt {
ENodeType nodeType;
ENodeType sqlNodeType;
SArray* pDataBlocks; // data block for each vgroup, SArray<SVgDataBlocks*>.
uint8_t payloadType; // EPayloadType. 0: K-V payload for non-prepare insert, 1: rawPayload for prepare insert
uint32_t insertType; // insert data from [file|sql statement| bound statement]
const char* sql; // current sql statement position
ENodeType nodeType;
ENodeType sqlNodeType;
SArray* pDataBlocks; // data block for each vgroup, SArray<SVgDataBlocks*>.
uint8_t payloadType; // EPayloadType. 0: K-V payload for non-prepare insert, 1: rawPayload for prepare insert
uint32_t insertType; // insert data from [file|sql statement| bound statement]
const char* sql; // current sql statement position
} SVnodeModifOpStmt;
typedef struct SExplainOptions {
ENodeType type;
bool verbose;
double ratio;
bool verbose;
double ratio;
} SExplainOptions;
typedef struct SExplainStmt {
ENodeType type;
bool analyze;
ENodeType type;
bool analyze;
SExplainOptions* pOptions;
SNode* pQuery;
SNode* pQuery;
} SExplainStmt;
void nodesWalkSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeWalker walker, void* pContext);
......@@ -324,10 +308,10 @@ bool nodesIsJsonOp(const SOperatorNode* pOp);
bool nodesIsTimeorderQuery(const SNode* pQuery);
bool nodesIsTimelineQuery(const SNode* pQuery);
void* nodesGetValueFromNode(SValueNode *pNode);
char* nodesGetStrValueFromNode(SValueNode *pNode);
char *getFillModeString(EFillMode mode);
void valueNodeToVariant(const SValueNode* pNode, SVariant* pVal);
void* nodesGetValueFromNode(SValueNode* pNode);
char* nodesGetStrValueFromNode(SValueNode* pNode);
char* getFillModeString(EFillMode mode);
void valueNodeToVariant(const SValueNode* pNode, SVariant* pVal);
#ifdef __cplusplus
}
......
include/util/taoserror.h
浏览文件 @ 842b75cd
......@@ -573,7 +573,6 @@ int32_t* taosGetErrno();
#define TSDB_CODE_PAR_TABLE_NOT_EXIST TAOS_DEF_ERROR_CODE(0, 0x2602)
#define TSDB_CODE_PAR_AMBIGUOUS_COLUMN TAOS_DEF_ERROR_CODE(0, 0x2603)
#define TSDB_CODE_PAR_WRONG_VALUE_TYPE TAOS_DEF_ERROR_CODE(0, 0x2604)
#define TSDB_CODE_PAR_INVALID_FUNTION TAOS_DEF_ERROR_CODE(0, 0x2605)
#define TSDB_CODE_PAR_ILLEGAL_USE_AGG_FUNCTION TAOS_DEF_ERROR_CODE(0, 0x2608)
#define TSDB_CODE_PAR_WRONG_NUMBER_OF_SELECT TAOS_DEF_ERROR_CODE(0, 0x2609)
#define TSDB_CODE_PAR_GROUPBY_LACK_EXPRESSION TAOS_DEF_ERROR_CODE(0, 0x260A)
......@@ -633,6 +632,7 @@ int32_t* taosGetErrno();
#define TSDB_CODE_FUNC_FUNTION_PARA_NUM TAOS_DEF_ERROR_CODE(0, 0x2801)
#define TSDB_CODE_FUNC_FUNTION_PARA_TYPE TAOS_DEF_ERROR_CODE(0, 0x2802)
#define TSDB_CODE_FUNC_FUNTION_PARA_VALUE TAOS_DEF_ERROR_CODE(0, 0x2803)
#define TSDB_CODE_FUNC_INVALID_FUNTION TAOS_DEF_ERROR_CODE(0, 0x2604)
#ifdef __cplusplus
}
......
include/util/tconfig.h
浏览文件 @ 842b75cd
......@@ -30,6 +30,7 @@ typedef enum {
CFG_STYPE_CFG_FILE,
CFG_STYPE_ENV_FILE,
CFG_STYPE_ENV_VAR,
CFG_STYPE_ENV_CMD,
CFG_STYPE_APOLLO_URL,
CFG_STYPE_ARG_LIST,
CFG_STYPE_TAOS_OPTIONS
......@@ -82,7 +83,7 @@ typedef struct SConfig {
} SConfig;
SConfig *cfgInit();
int32_t cfgLoad(SConfig *pCfg, ECfgSrcType cfgType, const char *sourceStr);
int32_t cfgLoad(SConfig *pCfg, ECfgSrcType cfgType, const void *sourceStr);
int32_t cfgLoadFromArray(SConfig *pCfg, SArray *pArgs); // SConfigPair
void cfgCleanup(SConfig *pCfg);
......@@ -105,6 +106,8 @@ const char *cfgDtypeStr(ECfgDataType type);
void cfgDumpCfg(SConfig *pCfg, bool tsc, bool dump);
int32_t cfgGetApollUrl(const char **envCmd, const char *envFile, char* apolloUrl);
#ifdef __cplusplus
}
#endif
......
include/util/tenv.h 0 → 100644
浏览文件 @ 842b75cd
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_ENV_H_
#define _TD_ENV_H_
#include "os.h"
#ifdef __cplusplus
extern "C" {
#endif
int32_t taosEnvNameToCfgName(const char *envNameStr, char *cfgNameStr, int32_t cfgNameMaxLen);
int32_t taosEnvToCfg(const char *envStr, char *cfgStr);
#ifdef __cplusplus
}
#endif
#endif /*_TD_ENV_H_*/
include/util/tjson.h
浏览文件 @ 842b75cd
......@@ -49,6 +49,8 @@ int32_t tjsonAddItemToObject(SJson* pJson, const char* pName, SJson* pItem);
int32_t tjsonAddItemToArray(SJson* pJson, SJson* pItem);
SJson* tjsonGetObjectItem(const SJson* pJson, const char* pName);
int32_t tjsonGetObjectName(const SJson* pJson, char** pName);
int32_t tjsonGetObjectValueString(const SJson* pJson, char** pStringValue);
int32_t tjsonGetStringValue(const SJson* pJson, const char* pName, char* pVal);
int32_t tjsonDupStringValue(const SJson* pJson, const char* pName, char** pVal);
int32_t tjsonGetBigIntValue(const SJson* pJson, const char* pName, int64_t* pVal);
......@@ -81,6 +83,7 @@ char* tjsonToUnformattedString(const SJson* pJson);
SJson* tjsonParse(const char* pStr);
bool tjsonValidateJson(const char* pJson);
const char* tjsonGetError();
#ifdef __cplusplus
}
......
include/util/ttimer.h
浏览文件 @ 842b75cd
......@@ -41,10 +41,6 @@ bool taosTmrReset(TAOS_TMR_CALLBACK fp, int32_t mseconds, void *param, void *han
void taosTmrCleanUp(void *handle);
int32_t taosInitTimer(void (*callback)(int32_t), int32_t ms);
void taosUninitTimer();
#ifdef __cplusplus
}
#endif
......
source/client/inc/clientInt.h
浏览文件 @ 842b75cd
......@@ -231,6 +231,10 @@ static FORCE_INLINE SReqResultInfo* tmqGetNextResInfo(TAOS_RES* res, bool conver
msg->resIter++;
if (msg->resIter < msg->rsp.blockNum) {
SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)taosArrayGetP(msg->rsp.blockData, msg->resIter);
if (msg->rsp.withSchema) {
SSchemaWrapper* pSW = (SSchemaWrapper*)taosArrayGetP(msg->rsp.blockSchema, msg->resIter);
setResSchemaInfo(&msg->resInfo, pSW->pSchema, pSW->nCols);
}
setQueryResultFromRsp(&msg->resInfo, pRetrieve, convertUcs4);
return &msg->resInfo;
}
......
source/client/src/clientEnv.c
浏览文件 @ 842b75cd
......@@ -254,12 +254,12 @@ void taos_init_imp(void) {
deltaToUtcInitOnce();
if (taosCreateLog("taoslog", 10, configDir, NULL, NULL, NULL, 1) != 0) {
if (taosCreateLog("taoslog", 10, configDir, NULL, NULL, NULL, NULL, 1) != 0) {
tscInitRes = -1;
return;
}
if (taosInitCfg(configDir, NULL, NULL, NULL, 1) != 0) {
if (taosInitCfg(configDir, NULL, NULL, NULL, NULL, 1) != 0) {
tscInitRes = -1;
return;
}
......
source/client/src/clientHb.c
浏览文件 @ 842b75cd
......@@ -14,9 +14,9 @@
*/
#include "catalog.h"
#include "scheduler.h"
#include "clientInt.h"
#include "clientLog.h"
#include "scheduler.h"
#include "trpc.h"
static SClientHbMgr clientHbMgr = {0};
......@@ -110,7 +110,8 @@ static int32_t hbProcessStbInfoRsp(void *value, int32_t valueLen, struct SCatalo
static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) {
SHbConnInfo *info = taosHashGet(pAppHbMgr->connInfo, &pRsp->connKey, sizeof(SClientHbKey));
if (NULL == info) {
tscWarn("fail to get connInfo, may be dropped, refId:%" PRIx64 ", type:%d", pRsp->connKey.tscRid, pRsp->connKey.connType);
tscWarn("fail to get connInfo, may be dropped, refId:%" PRIx64 ", type:%d", pRsp->connKey.tscRid,
pRsp->connKey.connType);
return TSDB_CODE_SUCCESS;
}
......@@ -121,7 +122,7 @@ static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) {
} else {
updateEpSet_s(&pTscObj->pAppInfo->mgmtEp, &pRsp->query->epSet);
pTscObj->connId = pRsp->query->connId;
if (pRsp->query->killRid) {
SRequestObj *pRequest = acquireRequest(pRsp->query->killRid);
if (NULL == pRequest) {
......@@ -131,7 +132,7 @@ static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) {
releaseRequest(pRsp->query->killRid);
}
}
if (pRsp->query->killConnection) {
taos_close(pTscObj);
}
......@@ -139,7 +140,7 @@ static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) {
releaseTscObj(pRsp->connKey.tscRid);
}
}
int32_t kvNum = pRsp->info ? taosArrayGetSize(pRsp->info) : 0;
tscDebug("hb got %d rsp kv", kvNum);
......@@ -236,24 +237,24 @@ static int32_t hbAsyncCallBack(void *param, const SDataBuf *pMsg, int32_t code)
}
int32_t hbBuildQueryDesc(SQueryHbReqBasic *hbBasic, STscObj *pObj) {
int64_t now = taosGetTimestampUs();
int64_t now = taosGetTimestampUs();
SQueryDesc desc = {0};
int32_t code = 0;
int32_t code = 0;
void *pIter = taosHashIterate(pObj->pRequests, NULL);
void *pIter = taosHashIterate(pObj->pRequests, NULL);
while (pIter != NULL) {
int64_t *rid = pIter;
int64_t *rid = pIter;
SRequestObj *pRequest = acquireRequest(*rid);
if (NULL == pRequest) {
continue;
}
tstrncpy(desc.sql, pRequest->sqlstr, sizeof(desc.sql));
desc.stime = pRequest->metric.start;
desc.queryId = pRequest->requestId;
desc.stime = pRequest->metric.start;
desc.queryId = pRequest->requestId;
desc.useconds = now - pRequest->metric.start;
desc.reqRid = pRequest->self;
desc.pid = hbBasic->pid;
desc.reqRid = pRequest->self;
desc.pid = hbBasic->pid;
taosGetFqdn(desc.fqdn);
desc.subPlanNum = pRequest->body.pDag ? pRequest->body.pDag->numOfSubplans : 0;
......@@ -271,9 +272,9 @@ int32_t hbBuildQueryDesc(SQueryHbReqBasic *hbBasic, STscObj *pObj) {
}
}
releaseRequest(*rid);
releaseRequest(*rid);
taosArrayPush(hbBasic->queryDesc, &desc);
pIter = taosHashIterate(pObj->pRequests, pIter);
}
......@@ -286,14 +287,14 @@ int32_t hbGetQueryBasicInfo(SClientHbKey *connKey, SClientHbReq *req) {
tscWarn("tscObj rid %" PRIx64 " not exist", connKey->tscRid);
return TSDB_CODE_QRY_APP_ERROR;
}
int32_t numOfQueries = pTscObj->pRequests ? taosHashGetSize(pTscObj->pRequests) : 0;
if (numOfQueries <= 0) {
releaseTscObj(connKey->tscRid);
tscDebug("no queries on connection");
return TSDB_CODE_QRY_APP_ERROR;
}
SQueryHbReqBasic *hbBasic = (SQueryHbReqBasic *)taosMemoryCalloc(1, sizeof(SQueryHbReqBasic));
if (NULL == hbBasic) {
tscError("calloc %d failed", (int32_t)sizeof(SQueryHbReqBasic));
......@@ -308,7 +309,7 @@ int32_t hbGetQueryBasicInfo(SClientHbKey *connKey, SClientHbReq *req) {
taosMemoryFree(hbBasic);
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
hbBasic->connId = pTscObj->connId;
hbBasic->pid = taosGetPId();
taosGetAppName(hbBasic->app, NULL);
......@@ -405,7 +406,7 @@ int32_t hbQueryHbReqHandle(SClientHbKey *connKey, void *param, SClientHbReq *req
}
hbGetQueryBasicInfo(connKey, req);
code = hbGetExpiredDBInfo(connKey, pCatalog, req);
if (TSDB_CODE_SUCCESS != code) {
return code;
......@@ -471,10 +472,10 @@ SClientHbBatchReq *hbGatherAllInfo(SAppHbMgr *pAppHbMgr) {
pIter = taosHashIterate(pAppHbMgr->activeInfo, pIter);
}
// if (code) {
// taosArrayDestroyEx(pBatchReq->reqs, hbFreeReq);
// taosMemoryFreeClear(pBatchReq);
// }
// if (code) {
// taosArrayDestroyEx(pBatchReq->reqs, hbFreeReq);
// taosMemoryFreeClear(pBatchReq);
// }
return pBatchReq;
}
......@@ -630,24 +631,23 @@ void appHbMgrCleanup(void) {
int sz = taosArrayGetSize(clientHbMgr.appHbMgrs);
for (int i = 0; i < sz; i++) {
SAppHbMgr *pTarget = taosArrayGetP(clientHbMgr.appHbMgrs, i);
void *pIter = taosHashIterate(pTarget->activeInfo, NULL);
void *pIter = taosHashIterate(pTarget->activeInfo, NULL);
while (pIter != NULL) {
SClientHbReq *pOneReq = pIter;
hbFreeReq(pOneReq);
taosHashCleanup(pOneReq->info);
pIter = taosHashIterate(pTarget->activeInfo, pIter);
}
}
taosHashCleanup(pTarget->activeInfo);
pTarget->activeInfo = NULL;
pIter = taosHashIterate(pTarget->connInfo, NULL);
while (pIter != NULL) {
SHbConnInfo *info = pIter;
taosMemoryFree(info->param);
pIter = taosHashIterate(pTarget->connInfo, pIter);
}
}
taosHashCleanup(pTarget->connInfo);
pTarget->connInfo = NULL;
......@@ -668,13 +668,13 @@ int hbMgrInit() {
hbMgrInitHandle();
// init backgroud thread
//hbCreateThread();
/*hbCreateThread();*/
return 0;
}
void hbMgrCleanUp() {
//hbStopThread();
// hbStopThread();
// destroy all appHbMgr
int8_t old = atomic_val_compare_exchange_8(&clientHbMgr.inited, 1, 0);
......@@ -747,11 +747,11 @@ void hbDeregisterConn(SAppHbMgr *pAppHbMgr, SClientHbKey connKey) {
taosMemoryFree(info->param);
taosHashRemove(pAppHbMgr->connInfo, &connKey, sizeof(SClientHbKey));
}
if (NULL == pReq || NULL == info) {
return;
}
atomic_sub_fetch_32(&pAppHbMgr->connKeyCnt, 1);
}
......
source/client/src/clientImpl.c
浏览文件 @ 842b75cd
......@@ -226,17 +226,15 @@ int32_t execDdlQuery(SRequestObj* pRequest, SQuery* pQuery) {
int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArray* pNodeList) {
pRequest->type = pQuery->msgType;
SPlanContext cxt = {
.queryId = pRequest->requestId,
.acctId = pRequest->pTscObj->acctId,
.mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp),
.pAstRoot = pQuery->pRoot,
.showRewrite = pQuery->showRewrite,
.pTransporter = pRequest->pTscObj->pAppInfo->pTransporter,
.pMsg = pRequest->msgBuf,
.msgLen = ERROR_MSG_BUF_DEFAULT_SIZE
};
int32_t code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &cxt.pCatalog);
SPlanContext cxt = {.queryId = pRequest->requestId,
.acctId = pRequest->pTscObj->acctId,
.mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp),
.pAstRoot = pQuery->pRoot,
.showRewrite = pQuery->showRewrite,
.pTransporter = pRequest->pTscObj->pAppInfo->pTransporter,
.pMsg = pRequest->msgBuf,
.msgLen = ERROR_MSG_BUF_DEFAULT_SIZE};
int32_t code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &cxt.pCatalog);
if (TSDB_CODE_SUCCESS == code) {
code = qCreateQueryPlan(&cxt, pPlan, pNodeList);
}
......@@ -247,6 +245,7 @@ void setResSchemaInfo(SReqResultInfo* pResInfo, const SSchema* pSchema, int32_t
ASSERT(pSchema != NULL && numOfCols > 0);
pResInfo->numOfCols = numOfCols;
// TODO handle memory leak
pResInfo->fields = taosMemoryCalloc(numOfCols, sizeof(TAOS_FIELD));
pResInfo->userFields = taosMemoryCalloc(numOfCols, sizeof(TAOS_FIELD));
......@@ -282,7 +281,7 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
SQueryResult res = {.code = 0, .numOfRows = 0, .msgSize = ERROR_MSG_BUF_DEFAULT_SIZE, .msg = pRequest->msgBuf};
int32_t code = schedulerExecJob(pTransporter, pNodeList, pDag, &pRequest->body.queryJob, pRequest->sqlstr,
pRequest->metric.start, &res);
pRequest->metric.start, &res);
if (code != TSDB_CODE_SUCCESS) {
if (pRequest->body.queryJob != 0) {
schedulerFreeJob(pRequest->body.queryJob);
......@@ -840,12 +839,12 @@ int32_t setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32
return code;
}
char* p = (char*) pResultInfo->pData;
char* p = (char*)pResultInfo->pData;
int32_t dataLen = *(int32_t*) p;
int32_t dataLen = *(int32_t*)p;
p += sizeof(int32_t);
uint64_t groupId = *(uint64_t*) p;
uint64_t groupId = *(uint64_t*)p;
p += sizeof(uint64_t);
int32_t* colLength = (int32_t*)p;
......
source/client/src/clientMain.c
浏览文件 @ 842b75cd
......@@ -120,7 +120,7 @@ const char *taos_errstr(TAOS_RES *res) {
return (const char *)tstrerror(terrno);
}
if (strlen(pRequest->msgBuf) > 0 || pRequest->code == TSDB_CODE_RPC_FQDN_ERROR) {
if (NULL != pRequest->msgBuf && (strlen(pRequest->msgBuf) > 0 || pRequest->code == TSDB_CODE_RPC_FQDN_ERROR)) {
return pRequest->msgBuf;
} else {
return (const char *)tstrerror(pRequest->code);
......
source/client/src/tmq.c
浏览文件 @ 842b75cd
......@@ -23,6 +23,9 @@
#include "tmsgtype.h"
#include "tqueue.h"
#include "tref.h"
#include "ttimer.h"
int32_t tmqAskEp(tmq_t* tmq, bool sync);
typedef struct {
int8_t tmqRspType;
......@@ -61,29 +64,40 @@ struct tmq_conf_t {
tmq_commit_cb* commit_cb;
};
typedef struct {
int8_t inited;
tmr_h timer;
} SMqMgmt;
static SMqMgmt tmqMgmt = {0};
struct tmq_t {
// conf
char groupId[TSDB_CGROUP_LEN];
char clientId[256];
int8_t autoCommit;
/*int8_t inWaiting;*/
char groupId[TSDB_CGROUP_LEN];
char clientId[256];
int8_t autoCommit;
int64_t consumerId;
int32_t epoch;
int32_t resetOffsetCfg;
int64_t status;
STscObj* pTscObj;
tmq_commit_cb* commit_cb;
/*int32_t nextTopicIdx;*/
// status
int8_t status;
int8_t epStatus;
int32_t epoch;
int32_t epSkipCnt;
/*int32_t waitingRequest;*/
/*int32_t readyRequest;*/
int64_t pollCnt;
// connection
STscObj* pTscObj;
// container
SArray* clientTopics; // SArray<SMqClientTopic>
STaosQueue* mqueue; // queue of tmq_message_t
STaosQueue* mqueue; // queue of rsp
STaosQall* qall;
tsem_t rspSem;
// stat
int64_t pollCnt;
STaosQueue* delayedTask; // delayed task queue for heartbeat and auto commit
// ctl
tsem_t rspSem;
};
enum {
......@@ -93,6 +107,7 @@ enum {
enum {
TMQ_CONSUMER_STATUS__INIT = 0,
TMQ_CONSUMER_STATUS__SUBSCRIBED,
TMQ_CONSUMER_STATUS__READY,
};
......@@ -110,13 +125,11 @@ typedef struct {
typedef struct {
// subscribe info
int32_t sqlLen;
char* sql;
char* topicName;
int64_t topicId;
SArray* vgs; // SArray<SMqClientVg>
char* topicName;
SArray* vgs; // SArray<SMqClientVg>
int8_t isSchemaAdaptive;
int32_t numOfFields;
SSchemaWrapper schema;
} SMqClientTopic;
......@@ -156,7 +169,6 @@ typedef struct {
int32_t async;
tsem_t rspSem;
tmq_resp_err_t rspErr;
/*SMqClientVg* pVg;*/
} SMqCommitCbParam;
tmq_conf_t* tmq_conf_new() {
......@@ -251,13 +263,7 @@ int32_t tmq_list_append(tmq_list_t* list, const char* src) {
void tmq_list_destroy(tmq_list_t* list) {
SArray* container = &list->container;
/*taosArrayDestroy(container);*/
int32_t sz = taosArrayGetSize(container);
for (int32_t i = 0; i < sz; i++) {
char* str = taosArrayGetP(container, i);
taosMemoryFree(str);
}
taosArrayDestroy(container);
taosArrayDestroyP(container, taosMemoryFree);
}
int32_t tmq_list_get_size(const tmq_list_t* list) {
......@@ -298,6 +304,8 @@ void tmqClearUnhandleMsg(tmq_t* tmq) {
int32_t tmqSubscribeCb(void* param, const SDataBuf* pMsg, int32_t code) {
SMqSubscribeCbParam* pParam = (SMqSubscribeCbParam*)param;
pParam->rspErr = code;
tmq_t* tmq = pParam->tmq;
atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__SUBSCRIBED);
tsem_post(&pParam->rspSem);
return 0;
}
......@@ -335,12 +343,9 @@ tmq_t* tmq_consumer_new(void* conn, tmq_conf_t* conf, char* errstr, int32_t errs
return NULL;
}
pTmq->pTscObj = (STscObj*)conn;
/*pTmq->inWaiting = 0;*/
pTmq->status = 0;
pTmq->pollCnt = 0;
pTmq->epoch = 0;
/*pTmq->waitingRequest = 0;*/
/*pTmq->readyRequest = 0;*/
pTmq->epStatus = 0;
pTmq->epSkipCnt = 0;
// set conf
......@@ -367,26 +372,45 @@ tmq_t* tmq_consumer_new(void* conn, tmq_conf_t* conf, char* errstr, int32_t errs
#endif
tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
// init timer
int8_t inited = atomic_val_compare_exchange_8(&tmqMgmt.inited, 0, 1);
if (inited == 0) {
tmqMgmt.timer = taosTmrInit(1000, 100, 360000, "TMQ");
if (tmqMgmt.timer == NULL) {
atomic_store_8(&tmqMgmt.inited, 0);
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
}
tmq_t* pTmq = taosMemoryCalloc(1, sizeof(tmq_t));
if (pTmq == NULL) {
return NULL;
}
const char* user = conf->user == NULL ? TSDB_DEFAULT_USER : conf->user;
const char* pass = conf->pass == NULL ? TSDB_DEFAULT_PASS : conf->pass;
ASSERT(user);
ASSERT(pass);
ASSERT(conf->db);
ASSERT(conf->groupId[0]);
pTmq->pTscObj = taos_connect_internal(conf->ip, user, pass, NULL, conf->db, conf->port, CONN_TYPE__TMQ);
if (pTmq->pTscObj == NULL) return NULL;
pTmq->clientTopics = taosArrayInit(0, sizeof(SMqClientTopic));
pTmq->mqueue = taosOpenQueue();
pTmq->qall = taosAllocateQall();
pTmq->delayedTask = taosOpenQueue();
if (pTmq->clientTopics == NULL || pTmq->mqueue == NULL || pTmq->qall == NULL || pTmq->delayedTask == NULL) {
goto FAIL;
}
pTmq->status = 0;
// init status
pTmq->status = TMQ_CONSUMER_STATUS__INIT;
pTmq->pollCnt = 0;
pTmq->epoch = 0;
pTmq->epStatus = 0;
pTmq->epSkipCnt = 0;
// set conf
strcpy(pTmq->clientId, conf->clientId);
strcpy(pTmq->groupId, conf->groupId);
......@@ -394,19 +418,30 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
pTmq->commit_cb = conf->commit_cb;
pTmq->resetOffsetCfg = conf->resetOffset;
// assign consumerId
pTmq->consumerId = tGenIdPI64();
pTmq->clientTopics = taosArrayInit(0, sizeof(SMqClientTopic));
if (pTmq->clientTopics == NULL) {
taosMemoryFree(pTmq);
return NULL;
}
pTmq->mqueue = taosOpenQueue();
pTmq->qall = taosAllocateQall();
// init semaphore
if (tsem_init(&pTmq->rspSem, 0, 0) != 0) {
goto FAIL;
}
tsem_init(&pTmq->rspSem, 0, 0);
// init connection
pTmq->pTscObj = taos_connect_internal(conf->ip, user, pass, NULL, NULL, conf->port, CONN_TYPE__TMQ);
if (pTmq->pTscObj == NULL) {
tsem_destroy(&pTmq->rspSem);
goto FAIL;
}
return pTmq;
FAIL:
if (pTmq->clientTopics) taosArrayDestroy(pTmq->clientTopics);
if (pTmq->mqueue) taosCloseQueue(pTmq->mqueue);
if (pTmq->delayedTask) taosCloseQueue(pTmq->delayedTask);
if (pTmq->qall) taosFreeQall(pTmq->qall);
taosMemoryFree(pTmq);
return NULL;
}
tmq_resp_err_t tmq_commit(tmq_t* tmq, const tmq_topic_vgroup_list_t* offsets, int32_t async) {
......@@ -497,86 +532,64 @@ tmq_resp_err_t tmq_commit(tmq_t* tmq, const tmq_topic_vgroup_list_t* offsets, in
return resp;
}
tmq_resp_err_t tmq_subscribe(tmq_t* tmq, tmq_list_t* topic_list) {
SRequestObj* pRequest = NULL;
SArray* container = &topic_list->container;
int32_t sz = taosArrayGetSize(container);
// destroy ex
taosArrayDestroy(tmq->clientTopics);
tmq->clientTopics = taosArrayInit(sz, sizeof(SMqClientTopic));
tmq_resp_err_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) {
const SArray* container = &topic_list->container;
int32_t sz = taosArrayGetSize(container);
void* buf = NULL;
SCMSubscribeReq req = {0};
int32_t code = -1;
SCMSubscribeReq req;
req.consumerId = tmq->consumerId;
strcpy(req.cgroup, tmq->groupId);
tstrncpy(req.cgroup, tmq->groupId, TSDB_CGROUP_LEN);
req.topicNames = taosArrayInit(sz, sizeof(void*));
if (req.topicNames == NULL) goto FAIL;
for (int i = 0; i < sz; i++) {
/*char* topicName = topic_list->elems[i];*/
char* topicName = taosArrayGetP(container, i);
for (int32_t i = 0; i < sz; i++) {
char* topic = taosArrayGetP(container, i);
SName name = {0};
#if 0
char* dbName = getDbOfConnection(tmq->pTscObj);
if (dbName == NULL) {
return TMQ_RESP_ERR__FAIL;
}
#endif
tNameSetDbName(&name, tmq->pTscObj->acctId, topicName, strlen(topicName));
#if 0
tNameFromString(&name, topicName, T_NAME_TABLE);
#endif
tNameSetDbName(&name, tmq->pTscObj->acctId, topic, strlen(topic));
char* topicFname = taosMemoryCalloc(1, TSDB_TOPIC_FNAME_LEN);
if (topicFname == NULL) {
goto _return;
char* topicFName = taosMemoryCalloc(1, TSDB_TOPIC_FNAME_LEN);
if (topicFName == NULL) {
goto FAIL;
}
tNameExtractFullName(&name, topicFname);
tscDebug("subscribe topic: %s", topicFname);
SMqClientTopic topic = {
.sql = NULL,
.sqlLen = 0,
.topicId = 0,
.topicName = topicFname,
.vgs = NULL,
};
topic.vgs = taosArrayInit(0, sizeof(SMqClientVg));
taosArrayPush(tmq->clientTopics, &topic);
taosArrayPush(req.topicNames, &topicFname);
#if 0
taosMemoryFree(dbName);
#endif
}
tNameExtractFullName(&name, topicFName);
int tlen = tSerializeSCMSubscribeReq(NULL, &req);
void* buf = taosMemoryMalloc(tlen);
if (buf == NULL) {
goto _return;
tscDebug("subscribe topic: %s", topicFName);
taosArrayPush(req.topicNames, &topicFName);
}
int32_t tlen = tSerializeSCMSubscribeReq(NULL, &req);
buf = taosMemoryMalloc(tlen);
if (buf == NULL) goto FAIL;
void* abuf = buf;
tSerializeSCMSubscribeReq(&abuf, &req);
/*printf("formatted: %s\n", dagStr);*/
pRequest = createRequest(tmq->pTscObj, NULL, NULL, TDMT_MND_SUBSCRIBE);
if (pRequest == NULL) {
tscError("failed to malloc request");
}
SMsgSendInfo* sendInfo = taosMemoryMalloc(sizeof(SMsgSendInfo));
if (sendInfo == NULL) goto FAIL;
SMqSubscribeCbParam param = {
.rspErr = TMQ_RESP_ERR__SUCCESS,
.tmq = tmq,
};
tsem_init(&param.rspSem, 0, 0);
pRequest->body.requestMsg = (SDataBuf){
if (tsem_init(&param.rspSem, 0, 0) != 0) goto FAIL;
sendInfo->msgInfo = (SDataBuf){
.pData = buf,
.len = tlen,
.handle = NULL,
};
SMsgSendInfo* sendInfo = buildMsgInfoImpl(pRequest);
sendInfo->requestId = generateRequestId();
sendInfo->requestObjRefId = 0;
sendInfo->param = &param;
sendInfo->fp = tmqSubscribeCb;
sendInfo->msgType = TDMT_MND_SUBSCRIBE;
SEpSet epSet = getEpSet_s(&tmq->pTscObj->pAppInfo->mgmtEp);
int64_t transporterId = 0;
......@@ -585,15 +598,28 @@ tmq_resp_err_t tmq_subscribe(tmq_t* tmq, tmq_list_t* topic_list) {
tsem_wait(&param.rspSem);
tsem_destroy(&param.rspSem);
_return:
/*if (sendInfo != NULL) {*/
/*destroySendMsgInfo(sendInfo);*/
/*}*/
code = param.rspErr;
if (code != 0) goto FAIL;
// TODO: add max retry cnt
while (TSDB_CODE_MND_CONSUMER_NOT_READY == tmqAskEp(tmq, true)) {
tscDebug("not ready, retry\n");
taosMsleep(500);
}
return param.rspErr;
code = 0;
FAIL:
if (req.topicNames != NULL) taosArrayDestroyP(req.topicNames, taosMemoryFree);
if (code != 0) {
taosMemoryFree(buf);
}
return code;
}
void tmq_conf_set_offset_commit_cb(tmq_conf_t* conf, tmq_commit_cb* cb) { conf->commit_cb = cb; }
void tmq_conf_set_offset_commit_cb(tmq_conf_t* conf, tmq_commit_cb* cb) {
//
conf->commit_cb = cb;
}
TAOS_RES* tmq_create_stream(TAOS* taos, const char* streamName, const char* tbName, const char* sql) {
STscObj* pTscObj = (STscObj*)taos;
......@@ -627,9 +653,6 @@ TAOS_RES* tmq_create_stream(TAOS* taos, const char* streamName, const char* tbNa
int32_t code = 0;
CHECK_CODE_GOTO(buildRequest(pTscObj, sql, sqlLen, &pRequest), _return);
CHECK_CODE_GOTO(parseSql(pRequest, false, &pQueryNode, NULL), _return);
// todo check for invalid sql statement and return with error code
CHECK_CODE_GOTO(nodesNodeToString(pQueryNode->pRoot, false, &astStr, NULL), _return);
/*printf("%s\n", pStr);*/
......@@ -653,7 +676,6 @@ TAOS_RES* tmq_create_stream(TAOS* taos, const char* streamName, const char* tbNa
}
tSerializeSCMCreateStreamReq(buf, tlen, &req);
/*printf("formatted: %s\n", dagStr);*/
pRequest->body.requestMsg = (SDataBuf){
.pData = buf,
......@@ -684,94 +706,6 @@ _return:
return pRequest;
}
#if 0
TAOS_RES* tmq_create_topic(TAOS* taos, const char* topicName, const char* sql, int sqlLen) {
STscObj* pTscObj = (STscObj*)taos;
SRequestObj* pRequest = NULL;
SQuery* pQueryNode = NULL;
char* astStr = NULL;
terrno = TSDB_CODE_SUCCESS;
if (taos == NULL || topicName == NULL || sql == NULL) {
tscError("invalid parameters for creating topic, connObj:%p, topic name:%s, sql:%s", taos, topicName, sql);
terrno = TSDB_CODE_TSC_INVALID_INPUT;
goto _return;
}
if (strlen(topicName) >= TSDB_TOPIC_NAME_LEN) {
tscError("topic name too long, max length:%d", TSDB_TOPIC_NAME_LEN - 1);
terrno = TSDB_CODE_TSC_INVALID_INPUT;
goto _return;
}
if (sqlLen > TSDB_MAX_ALLOWED_SQL_LEN) {
tscError("sql string exceeds max length:%d", TSDB_MAX_ALLOWED_SQL_LEN);
terrno = TSDB_CODE_TSC_EXCEED_SQL_LIMIT;
goto _return;
}
tscDebug("start to create topic: %s", topicName);
int32_t code = TSDB_CODE_SUCCESS;
CHECK_CODE_GOTO(buildRequest(pTscObj, sql, sqlLen, &pRequest), _return);
CHECK_CODE_GOTO(parseSql(pRequest, true, &pQueryNode), _return);
// todo check for invalid sql statement and return with error code
CHECK_CODE_GOTO(nodesNodeToString(pQueryNode->pRoot, false, &astStr, NULL), _return);
/*printf("%s\n", pStr);*/
SName name = {.acctId = pTscObj->acctId, .type = TSDB_TABLE_NAME_T};
strcpy(name.dbname, pRequest->pDb);
strcpy(name.tname, topicName);
SCMCreateTopicReq req = {
.igExists = 1,
.ast = astStr,
.sql = (char*)sql,
};
tNameExtractFullName(&name, req.name);
int tlen = tSerializeSCMCreateTopicReq(NULL, 0, &req);
void* buf = taosMemoryMalloc(tlen);
if (buf == NULL) {
goto _return;
}
tSerializeSCMCreateTopicReq(buf, tlen, &req);
/*printf("formatted: %s\n", dagStr);*/
pRequest->body.requestMsg = (SDataBuf){
.pData = buf,
.len = tlen,
.handle = NULL,
};
pRequest->type = TDMT_MND_CREATE_TOPIC;
SMsgSendInfo* sendInfo = buildMsgInfoImpl(pRequest);
SEpSet epSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
int64_t transporterId = 0;
asyncSendMsgToServer(pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, sendInfo);
tsem_wait(&pRequest->body.rspSem);
_return:
taosMemoryFreeClear(astStr);
qDestroyQuery(pQueryNode);
/*if (sendInfo != NULL) {*/
/*destroySendMsgInfo(sendInfo);*/
/*}*/
if (pRequest != NULL && terrno != TSDB_CODE_SUCCESS) {
pRequest->code = terrno;
}
return pRequest;
}
#endif
#if 0
int32_t tmqGetSkipLogNum(tmq_message_t* tmq_message) {
if (tmq_message == NULL) return 0;
......@@ -954,7 +888,7 @@ int32_t tmqAskEpCb(void* param, const SDataBuf* pMsg, int32_t code) {
/*printf("rsp epoch %ld sz %ld\n", rsp.epoch, rsp.topics->size);*/
/*printf("tmq epoch %ld sz %ld\n", tmq->epoch, tmq->clientTopics->size);*/
if (tmqUpdateEp(tmq, head->epoch, &rsp)) {
atomic_store_64(&tmq->status, TMQ_CONSUMER_STATUS__READY);
atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__READY);
}
tDeleteSMqCMGetSubEpRsp(&rsp);
} else {
......@@ -1118,7 +1052,9 @@ SMqRspObj* tmqBuildRspFromWrapper(SMqPollRspWrapper* pWrapper) {
pRspObj->resInfo.totalRows = 0;
pRspObj->resInfo.precision = TSDB_TIME_PRECISION_MILLI;
setResSchemaInfo(&pRspObj->resInfo, pWrapper->topicHandle->schema.pSchema, pWrapper->topicHandle->schema.nCols);
if (!pWrapper->msg.withSchema) {
setResSchemaInfo(&pRspObj->resInfo, pWrapper->topicHandle->schema.pSchema, pWrapper->topicHandle->schema.nCols);
}
taosFreeQitem(pWrapper);
return pRspObj;
......@@ -1187,7 +1123,6 @@ int32_t tmqPollImpl(tmq_t* tmq, int64_t blockingTime) {
int64_t transporterId = 0;
/*printf("send poll\n");*/
/*atomic_add_fetch_32(&tmq->waitingRequest, 1);*/
tscDebug("consumer %ld send poll to %s : vg %d, epoch %d, req offset %ld, reqId %lu", tmq->consumerId,
pTopic->topicName, pVg->vgId, tmq->epoch, pVg->currentOffset, pReq->reqId);
/*printf("send vg %d %ld\n", pVg->vgId, pVg->currentOffset);*/
......@@ -1266,12 +1201,14 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t blocking_time) {
SMqRspObj* rspObj;
int64_t startTime = taosGetTimestampMs();
// TODO: put into another thread or delayed queue
int64_t status = atomic_load_64(&tmq->status);
while (0 != tmqAskEp(tmq, status == TMQ_CONSUMER_STATUS__INIT)) {
// TODO: put into delayed queue
#if 0
int8_t status = atomic_load_8(&tmq->status);
while (0 != tmqAskEp(tmq, status != TMQ_CONSUMER_STATUS__READY)) {
tscDebug("not ready, retry\n");
taosSsleep(1);
}
#endif
rspObj = tmqHandleAllRsp(tmq, blocking_time, false);
if (rspObj) {
......@@ -1279,7 +1216,6 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t blocking_time) {
}
while (1) {
/*printf("cycle\n");*/
tmqAskEp(tmq, false);
tmqPollImpl(tmq, blocking_time);
......
source/common/src/tglobal.c
浏览文件 @ 842b75cd
......@@ -220,7 +220,7 @@ struct SConfig *taosGetCfg() {
return tsCfg;
}
static int32_t taosLoadCfg(SConfig *pCfg, const char *inputCfgDir, const char *envFile, const char *apolloUrl) {
static int32_t taosLoadCfg(SConfig *pCfg, const char **envCmd, const char *inputCfgDir, const char *envFile, char *apolloUrl) {
char cfgDir[PATH_MAX] = {0};
char cfgFile[PATH_MAX + 100] = {0};
......@@ -231,6 +231,8 @@ static int32_t taosLoadCfg(SConfig *pCfg, const char *inputCfgDir, const char *e
tstrncpy(cfgFile, cfgDir, sizeof(cfgDir));
}
if (apolloUrl == NULL || apolloUrl[0] == '\0') cfgGetApollUrl(envCmd, envFile, apolloUrl);
if (cfgLoad(pCfg, CFG_STYPE_APOLLO_URL, apolloUrl) != 0) {
uError("failed to load from apollo url:%s since %s", apolloUrl, terrstr());
return -1;
......@@ -251,6 +253,11 @@ static int32_t taosLoadCfg(SConfig *pCfg, const char *inputCfgDir, const char *e
return -1;
}
if (cfgLoad(pCfg, CFG_STYPE_ENV_CMD, envCmd) != 0) {
uError("failed to load from cmd env variables since %s", terrstr());
return -1;
}
return 0;
}
......@@ -325,8 +332,8 @@ static int32_t taosAddSystemCfg(SConfig *pCfg) {
if (cfgAddFloat(pCfg, "numOfCores", tsNumOfCores, 0, 100000, 1) != 0) return -1;
if (cfgAddInt64(pCfg, "openMax", tsOpenMax, 0, INT64_MAX, 1) != 0) return -1;
if (cfgAddInt64(pCfg, "streamMax", tsStreamMax, 0, INT64_MAX, 1) != 0) return -1;
if (cfgAddInt32(pCfg, "pageSize(KB)", tsPageSizeKB, 0, INT64_MAX, 1) != 0) return -1;
if (cfgAddInt64(pCfg, "totalMemory(KB)", tsTotalMemoryKB, 0, INT64_MAX, 1) != 0) return -1;
if (cfgAddInt32(pCfg, "pageSizeKB", tsPageSizeKB, 0, INT64_MAX, 1) != 0) return -1;
if (cfgAddInt64(pCfg, "totalMemoryKB", tsTotalMemoryKB, 0, INT64_MAX, 1) != 0) return -1;
if (cfgAddString(pCfg, "os sysname", info.sysname, 1) != 0) return -1;
if (cfgAddString(pCfg, "os nodename", info.nodename, 1) != 0) return -1;
if (cfgAddString(pCfg, "os release", info.release, 1) != 0) return -1;
......@@ -572,8 +579,8 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
return 0;
}
int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDir, const char *envFile,
const char *apolloUrl, SArray *pArgs, bool tsc) {
int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDir, const char **envCmd, const char *envFile,
char *apolloUrl, SArray *pArgs, bool tsc) {
osDefaultInit();
SConfig *pCfg = cfgInit();
......@@ -588,7 +595,7 @@ int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDi
if (taosAddServerLogCfg(pCfg) != 0) return -1;
}
if (taosLoadCfg(pCfg, cfgDir, envFile, apolloUrl) != 0) {
if (taosLoadCfg(pCfg, envCmd, cfgDir, envFile, apolloUrl) != 0) {
uError("failed to load cfg since %s", terrstr());
cfgCleanup(pCfg);
return -1;
......@@ -625,7 +632,7 @@ int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDi
return 0;
}
int32_t taosInitCfg(const char *cfgDir, const char *envFile, const char *apolloUrl, SArray *pArgs, bool tsc) {
int32_t taosInitCfg(const char *cfgDir, const char **envCmd, const char *envFile, char *apolloUrl, SArray *pArgs, bool tsc) {
if (tsCfg != NULL) return 0;
tsCfg = cfgInit();
......@@ -640,7 +647,7 @@ int32_t taosInitCfg(const char *cfgDir, const char *envFile, const char *apolloU
}
taosAddSystemCfg(tsCfg);
if (taosLoadCfg(tsCfg, cfgDir, envFile, apolloUrl) != 0) {
if (taosLoadCfg(tsCfg, envCmd, cfgDir, envFile, apolloUrl) != 0) {
uError("failed to load cfg since %s", terrstr());
cfgCleanup(tsCfg);
tsCfg = NULL;
......
source/common/src/tmsg.c
浏览文件 @ 842b75cd
......@@ -93,7 +93,87 @@ STSRow *tGetSubmitBlkNext(SSubmitBlkIter *pIter) {
return row;
}
}
#if 0
// TODO: KEEP one suite of iterator API finally.
// 1) use tInitSubmitMsgIterEx firstly as not decrease the merge conflicts
// 2) replace tInitSubmitMsgIterEx with tInitSubmitMsgIter later
// 3) finally, rename tInitSubmitMsgIterEx to tInitSubmitMsgIter
int32_t tInitSubmitMsgIterEx(const SSubmitReq *pMsg, SSubmitMsgIter *pIter) {
if (pMsg == NULL) {
terrno = TSDB_CODE_TDB_SUBMIT_MSG_MSSED_UP;
return -1;
}
pIter->totalLen = htonl(pMsg->length);
ASSERT(pIter->totalLen > 0);
pIter->len = 0;
pIter->pMsg = pMsg;
if (pIter->totalLen <= sizeof(SSubmitReq)) {
terrno = TSDB_CODE_TDB_SUBMIT_MSG_MSSED_UP;
return -1;
}
return 0;
}
int32_t tGetSubmitMsgNextEx(SSubmitMsgIter *pIter, SSubmitBlk **pPBlock) {
ASSERT(pIter->len >= 0);
if (pIter->len == 0) {
pIter->len += sizeof(SSubmitReq);
} else {
if (pIter->len >= pIter->totalLen) {
ASSERT(0);
}
SSubmitBlk *pSubmitBlk = (SSubmitBlk *)POINTER_SHIFT(pIter->pMsg, pIter->len);
pIter->len += (sizeof(SSubmitBlk) + pIter->dataLen + pIter->schemaLen);
ASSERT(pIter->len > 0);
}
if (pIter->len > pIter->totalLen) {
terrno = TSDB_CODE_TDB_SUBMIT_MSG_MSSED_UP;
*pPBlock = NULL;
return -1;
}
if (pIter->len == pIter->totalLen) {
*pPBlock = NULL;
} else {
*pPBlock = (SSubmitBlk *)POINTER_SHIFT(pIter->pMsg, pIter->len);
pIter->uid = htobe64((*pPBlock)->uid);
pIter->suid = htobe64((*pPBlock)->suid);
pIter->sversion = htonl((*pPBlock)->sversion);
pIter->dataLen = htonl((*pPBlock)->dataLen);
pIter->schemaLen = htonl((*pPBlock)->schemaLen);
pIter->numOfRows = htons((*pPBlock)->numOfRows);
}
return 0;
}
int32_t tInitSubmitBlkIterEx(SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlock, SSubmitBlkIter *pIter) {
if (pMsgIter->dataLen <= 0) return -1;
pIter->totalLen = pMsgIter->dataLen;
pIter->len = 0;
pIter->row = (STSRow *)(pBlock->data + pMsgIter->schemaLen);
return 0;
}
STSRow *tGetSubmitBlkNextEx(SSubmitBlkIter *pIter) {
STSRow *row = pIter->row;
if (pIter->len >= pIter->totalLen) {
return NULL;
} else {
pIter->len += TD_ROW_LEN(row);
if (pIter->len < pIter->totalLen) {
pIter->row = POINTER_SHIFT(row, TD_ROW_LEN(row));
}
return row;
}
}
#endif
int32_t tEncodeSEpSet(SCoder *pEncoder, const SEpSet *pEp) {
if (tEncodeI8(pEncoder, pEp->inUse) < 0) return -1;
if (tEncodeI8(pEncoder, pEp->numOfEps) < 0) return -1;
......@@ -442,10 +522,6 @@ int32_t tSerializeSVCreateTbReq(void **buf, SVCreateTbReq *pReq) {
SRSmaParam *param = pReq->stbCfg.pRSmaParam;
tlen += taosEncodeBinary(buf, (const void *)&param->xFilesFactor, sizeof(param->xFilesFactor));
tlen += taosEncodeFixedI32(buf, param->delay);
tlen += taosEncodeFixedI8(buf, param->nFuncIds);
for (int8_t i = 0; i < param->nFuncIds; ++i) {
tlen += taosEncodeFixedI32(buf, param->pFuncIds[i]);
}
tlen += taosEncodeFixedI32(buf, param->qmsg1Len);
if (param->qmsg1Len > 0) {
tlen += taosEncodeString(buf, param->qmsg1);
......@@ -475,10 +551,6 @@ int32_t tSerializeSVCreateTbReq(void **buf, SVCreateTbReq *pReq) {
SRSmaParam *param = pReq->ntbCfg.pRSmaParam;
tlen += taosEncodeBinary(buf, (const void *)&param->xFilesFactor, sizeof(param->xFilesFactor));
tlen += taosEncodeFixedI32(buf, param->delay);
tlen += taosEncodeFixedI8(buf, param->nFuncIds);
for (int8_t i = 0; i < param->nFuncIds; ++i) {
tlen += taosEncodeFixedI32(buf, param->pFuncIds[i]);
}
}
break;
default:
......@@ -521,13 +593,6 @@ void *tDeserializeSVCreateTbReq(void *buf, SVCreateTbReq *pReq) {
SRSmaParam *param = pReq->stbCfg.pRSmaParam;
buf = taosDecodeBinaryTo(buf, (void *)&param->xFilesFactor, sizeof(param->xFilesFactor));
buf = taosDecodeFixedI32(buf, &param->delay);
buf = taosDecodeFixedI8(buf, &param->nFuncIds);
if (param->nFuncIds > 0) {
param->pFuncIds = (func_id_t *)taosMemoryCalloc(param->nFuncIds, sizeof(func_id_t));
for (int8_t i = 0; i < param->nFuncIds; ++i) {
buf = taosDecodeFixedI32(buf, param->pFuncIds + i);
}
}
buf = taosDecodeFixedI32(buf, &param->qmsg1Len);
if (param->qmsg1Len > 0) {
buf = taosDecodeString(buf, &param->qmsg1);
......@@ -561,15 +626,6 @@ void *tDeserializeSVCreateTbReq(void *buf, SVCreateTbReq *pReq) {
SRSmaParam *param = pReq->ntbCfg.pRSmaParam;
buf = taosDecodeBinaryTo(buf, (void *)&param->xFilesFactor, sizeof(param->xFilesFactor));
buf = taosDecodeFixedI32(buf, &param->delay);
buf = taosDecodeFixedI8(buf, &param->nFuncIds);
if (param->nFuncIds > 0) {
param->pFuncIds = (func_id_t *)taosMemoryMalloc(param->nFuncIds * sizeof(func_id_t));
for (int8_t i = 0; i < param->nFuncIds; ++i) {
buf = taosDecodeFixedI32(buf, param->pFuncIds + i);
}
} else {
param->pFuncIds = NULL;
}
} else {
pReq->ntbCfg.pRSmaParam = NULL;
}
......@@ -632,7 +688,6 @@ int32_t tSerializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pReq
if (tEncodeCStr(&encoder, pReq->name) < 0) return -1;
if (tEncodeI8(&encoder, pReq->igExists) < 0) return -1;
if (tEncodeFloat(&encoder, pReq->xFilesFactor) < 0) return -1;
if (tEncodeI32(&encoder, pReq->aggregationMethod) < 0) return -1;
if (tEncodeI32(&encoder, pReq->delay) < 0) return -1;
if (tEncodeI32(&encoder, pReq->ttl) < 0) return -1;
if (tEncodeI32(&encoder, pReq->numOfColumns) < 0) return -1;
......@@ -687,7 +742,6 @@ int32_t tDeserializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pR
if (tDecodeCStrTo(&decoder, pReq->name) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->igExists) < 0) return -1;
if (tDecodeFloat(&decoder, &pReq->xFilesFactor) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->aggregationMethod) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->delay) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->ttl) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->numOfColumns) < 0) return -1;
......
source/dnode/mgmt/exe/dmMain.c
浏览文件 @ 842b75cd
......@@ -24,6 +24,7 @@ static struct {
bool printVersion;
char envFile[PATH_MAX];
char apolloUrl[PATH_MAX];
const char **envCmd;
SArray *pArgs; // SConfigPair
SDnode *pDnode;
EDndNodeType ntype;
......@@ -56,6 +57,9 @@ static void dmSetSignalHandle() {
}
static int32_t dmParseArgs(int32_t argc, char const *argv[]) {
int32_t cmdEnvIndex = 0;
global.envCmd = taosMemoryMalloc(argc-1);
memset(global.envCmd, 0, argc-1);
for (int32_t i = 1; i < argc; ++i) {
if (strcmp(argv[i], "-c") == 0) {
if (i < argc - 1) {
......@@ -70,7 +74,7 @@ static int32_t dmParseArgs(int32_t argc, char const *argv[]) {
}
} else if (strcmp(argv[i], "-a") == 0) {
tstrncpy(global.apolloUrl, argv[++i], PATH_MAX);
} else if (strcmp(argv[i], "-e") == 0) {
} else if (strcmp(argv[i], "-E") == 0) {
tstrncpy(global.envFile, argv[++i], PATH_MAX);
} else if (strcmp(argv[i], "-n") == 0) {
global.ntype = atoi(argv[++i]);
......@@ -84,6 +88,9 @@ static int32_t dmParseArgs(int32_t argc, char const *argv[]) {
global.dumpConfig = true;
} else if (strcmp(argv[i], "-V") == 0) {
global.printVersion = true;
} else if (strcmp(argv[i], "-e") == 0) {
global.envCmd[cmdEnvIndex] = argv[++i];
cmdEnvIndex++;
} else {
}
}
......@@ -129,7 +136,7 @@ static SDnodeOpt dmGetOpt() {
static int32_t dmInitLog() {
char logName[12] = {0};
snprintf(logName, sizeof(logName), "%slog", dmLogName(global.ntype));
return taosCreateLog(logName, 1, configDir, global.envFile, global.apolloUrl, global.pArgs, 0);
return taosCreateLog(logName, 1, configDir, global.envCmd, global.envFile, global.apolloUrl, global.pArgs, 0);
}
static void dmSetProcInfo(int32_t argc, char **argv) {
......@@ -168,6 +175,10 @@ static int32_t dmRunDnode() {
return code;
}
static void taosCleanupArgs() {
if (global.envCmd != NULL) taosMemoryFree(global.envCmd);
}
int main(int argc, char const *argv[]) {
if (!taosCheckSystemIsSmallEnd()) {
printf("failed to start since on non-small-end machines\n");
......@@ -176,26 +187,31 @@ int main(int argc, char const *argv[]) {
if (dmParseArgs(argc, argv) != 0) {
printf("failed to start since parse args error\n");
taosCleanupArgs();
return -1;
}
if (global.generateGrant) {
dmGenerateGrant();
taosCleanupArgs();
return 0;
}
if (global.printVersion) {
dmPrintVersion();
taosCleanupArgs();
return 0;
}
if (dmInitLog() != 0) {
dError("failed to start since init log error");
printf("failed to start since init log error");
taosCleanupArgs();
return -1;
}
if (taosInitCfg(configDir, global.envFile, global.apolloUrl, global.pArgs, 0) != 0) {
if (taosInitCfg(configDir, global.envCmd, global.envFile, global.apolloUrl, global.pArgs, 0) != 0) {
dError("failed to start since read config error");
taosCleanupArgs();
return -1;
}
......@@ -203,9 +219,11 @@ int main(int argc, char const *argv[]) {
dmDumpCfg();
taosCleanupCfg();
taosCloseLog();
taosCleanupArgs();
return 0;
}
dmSetProcInfo(argc, (char **)argv);
taosCleanupArgs();
return dmRunDnode();
}
source/dnode/mgmt/interface/src/dmFile.c
浏览文件 @ 842b75cd
......@@ -124,7 +124,7 @@ TdFilePtr dmCheckRunning(const char *dataDir) {
TdFilePtr pFile = taosOpenFile(filepath, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
dError("failed to lock file:%s since %s", filepath, terrstr());
dError("failed to open file:%s since %s", filepath, terrstr());
return NULL;
}
......
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
浏览文件 @ 842b75cd
......@@ -202,6 +202,17 @@ int32_t vmProcessCreateVnodeReq(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
return code;
}
code = vnodeStart(pImpl);
if (code != 0) {
tFreeSCreateVnodeReq(&createReq);
dError("vgId:%d, failed to start sync since %s", createReq.vgId, terrstr());
vnodeClose(pImpl);
vnodeDestroy(path, pMgmt->pTfs);
terrno = code;
return code;
}
code = vmWriteVnodesToFile(pMgmt);
if (code != 0) {
tFreeSCreateVnodeReq(&createReq);
......
source/dnode/mgmt/mgmt_vnode/src/vmInt.c
浏览文件 @ 842b75cd
......@@ -15,6 +15,7 @@
#define _DEFAULT_SOURCE
#include "vmInt.h"
#include "libs/function/function.h"
SVnodeObj *vmAcquireVnode(SVnodesMgmt *pMgmt, int32_t vgId) {
SVnodeObj *pVnode = NULL;
......@@ -73,12 +74,6 @@ int32_t vmOpenVnode(SVnodesMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl) {
return -1;
}
// sync integration
vnodeSyncSetQ(pImpl, NULL);
vnodeSyncSetRpc(pImpl, NULL);
int32_t ret = vnodeSyncStart(pImpl);
assert(ret == 0);
taosWLockLatch(&pMgmt->latch);
int32_t code = taosHashPut(pMgmt->hash, &pVnode->vgId, sizeof(int32_t), &pVnode, sizeof(SVnodeObj *));
taosWUnLockLatch(&pMgmt->latch);
......@@ -152,6 +147,7 @@ static void *vmOpenVnodeFunc(void *param) {
pThread->failed++;
} else {
vmOpenVnode(pMgmt, pCfg, pImpl);
//vnodeStart(pImpl);
dDebug("vgId:%d, is opened by thread:%d", pCfg->vgId, pThread->threadIndex);
pThread->opened++;
}
......@@ -275,7 +271,7 @@ static void vmCleanup(SMgmtWrapper *pWrapper) {
pWrapper->pMgmt = NULL;
// syncCleanUp();
udfcClose();
dInfo("vnode-mgmt is cleaned up");
}
......@@ -339,6 +335,10 @@ static int32_t vmInit(SMgmtWrapper *pWrapper) {
}
dmReportStartup(pDnode, "vnode-vnodes", "initialized");
if (udfcOpen() != 0) {
dError("failed to open udfc in dnode");
}
code = 0;
_OVER:
......@@ -359,10 +359,52 @@ static int32_t vmRequire(SMgmtWrapper *pWrapper, bool *required) {
return 0;
}
static int32_t vmStart(SMgmtWrapper *pWrapper) {
dDebug("vnode-mgmt start to run");
SVnodesMgmt *pMgmt = pWrapper->pMgmt;
taosRLockLatch(&pMgmt->latch);
void *pIter = taosHashIterate(pMgmt->hash, NULL);
while (pIter) {
SVnodeObj **ppVnode = pIter;
if (ppVnode == NULL || *ppVnode == NULL) continue;
SVnodeObj *pVnode = *ppVnode;
vnodeStart(pVnode->pImpl);
pIter = taosHashIterate(pMgmt->hash, pIter);
}
taosRUnLockLatch(&pMgmt->latch);
return 0;
}
static void vmStop(SMgmtWrapper *pWrapper) {
#if 0
dDebug("vnode-mgmt start to stop");
SVnodesMgmt *pMgmt = pWrapper->pMgmt;
taosRLockLatch(&pMgmt->latch);
void *pIter = taosHashIterate(pMgmt->hash, NULL);
while (pIter) {
SVnodeObj **ppVnode = pIter;
if (ppVnode == NULL || *ppVnode == NULL) continue;
SVnodeObj *pVnode = *ppVnode;
vnodeStop(pVnode->pImpl);
pIter = taosHashIterate(pMgmt->hash, pIter);
}
taosRUnLockLatch(&pMgmt->latch);
#endif
}
void vmSetMgmtFp(SMgmtWrapper *pWrapper) {
SMgmtFp mgmtFp = {0};
mgmtFp.openFp = vmInit;
mgmtFp.closeFp = vmCleanup;
mgmtFp.startFp = vmStart;
mgmtFp.stopFp = vmStop;
mgmtFp.requiredFp = vmRequire;
vmInitMsgHandle(pWrapper);
......@@ -391,4 +433,4 @@ void vmGetVnodeLoads(SMgmtWrapper *pWrapper, SMonVloadInfo *pInfo) {
}
taosRUnLockLatch(&pMgmt->latch);
}
\ No newline at end of file
}
source/dnode/mnode/impl/inc/mndDef.h
浏览文件 @ 842b75cd
......@@ -350,7 +350,6 @@ typedef struct {
int32_t version;
int32_t nextColId;
float xFilesFactor;
int32_t aggregationMethod;
int32_t delay;
int32_t ttl;
int32_t numOfColumns;
......
source/dnode/mnode/impl/src/mndScheduler.c
浏览文件 @ 842b75cd
......@@ -476,32 +476,36 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscribeObj* pSub) {
SSdb* pSdb = pMnode->pSdb;
SVgObj* pVgroup = NULL;
SQueryPlan* pPlan = qStringToQueryPlan(pTopic->physicalPlan);
if (pPlan == NULL) {
terrno = TSDB_CODE_QRY_INVALID_INPUT;
return -1;
}
SQueryPlan* pPlan = NULL;
SSubplan* plan = NULL;
if (pTopic->subType == TOPIC_SUB_TYPE__TABLE) {
pPlan = qStringToQueryPlan(pTopic->physicalPlan);
if (pPlan == NULL) {
terrno = TSDB_CODE_QRY_INVALID_INPUT;
return -1;
}
ASSERT(pSub->vgNum == -1);
ASSERT(pSub->vgNum == -1);
pSub->vgNum = 0;
pSub->vgNum = 0;
int32_t levelNum = LIST_LENGTH(pPlan->pSubplans);
if (levelNum != 1) {
qDestroyQueryPlan(pPlan);
terrno = TSDB_CODE_MND_UNSUPPORTED_TOPIC;
return -1;
}
int32_t levelNum = LIST_LENGTH(pPlan->pSubplans);
if (levelNum != 1) {
qDestroyQueryPlan(pPlan);
terrno = TSDB_CODE_MND_UNSUPPORTED_TOPIC;
return -1;
}
SNodeListNode* inner = nodesListGetNode(pPlan->pSubplans, 0);
SNodeListNode* inner = nodesListGetNode(pPlan->pSubplans, 0);
int32_t opNum = LIST_LENGTH(inner->pNodeList);
if (opNum != 1) {
qDestroyQueryPlan(pPlan);
terrno = TSDB_CODE_MND_UNSUPPORTED_TOPIC;
return -1;
int32_t opNum = LIST_LENGTH(inner->pNodeList);
if (opNum != 1) {
qDestroyQueryPlan(pPlan);
terrno = TSDB_CODE_MND_UNSUPPORTED_TOPIC;
return -1;
}
plan = nodesListGetNode(inner->pNodeList, 0);
}
SSubplan* plan = nodesListGetNode(inner->pNodeList, 0);
int64_t unexistKey = -1;
SMqConsumerEpInSub* pEpInSub = taosHashGet(pSub->consumerHash, &unexistKey, sizeof(int64_t));
......@@ -519,38 +523,35 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
}
pSub->vgNum++;
plan->execNode.nodeId = pVgroup->vgId;
plan->execNode.epSet = mndGetVgroupEpset(pMnode, pVgroup);
SMqVgEp* pVgEp = taosMemoryMalloc(sizeof(SMqVgEp));
pVgEp->epSet = plan->execNode.epSet;
pVgEp->vgId = plan->execNode.nodeId;
#if 0
SMqConsumerEp consumerEp = {0};
consumerEp.status = 0;
consumerEp.consumerId = -1;
consumerEp.epSet = plan->execNode.epSet;
consumerEp.vgId = plan->execNode.nodeId;
#endif
pVgEp->epSet = mndGetVgroupEpset(pMnode, pVgroup);
pVgEp->vgId = pVgroup->vgId;
taosArrayPush(pEpInSub->vgs, &pVgEp);
mDebug("init subscribption %s, assign vg: %d", pSub->key, pVgEp->vgId);
int32_t msgLen;
if (qSubPlanToString(plan, &pVgEp->qmsg, &msgLen) < 0) {
sdbRelease(pSdb, pVgroup);
qDestroyQueryPlan(pPlan);
terrno = TSDB_CODE_QRY_INVALID_INPUT;
return -1;
if (pTopic->subType == TOPIC_SUB_TYPE__TABLE) {
int32_t msgLen;
plan->execNode.epSet = pVgEp->epSet;
plan->execNode.nodeId = pVgEp->vgId;
if (qSubPlanToString(plan, &pVgEp->qmsg, &msgLen) < 0) {
sdbRelease(pSdb, pVgroup);
qDestroyQueryPlan(pPlan);
terrno = TSDB_CODE_QRY_INVALID_INPUT;
return -1;
}
} else {
pVgEp->qmsg = strdup("");
}
taosArrayPush(pEpInSub->vgs, &pVgEp);
ASSERT(taosHashGetSize(pSub->consumerHash) == 1);
/*taosArrayPush(pSub->unassignedVg, &consumerEp);*/
}
ASSERT(pEpInSub->vgs->size > 0);
pEpInSub = taosHashGet(pSub->consumerHash, &unexistKey, sizeof(int64_t));
ASSERT(pEpInSub->vgs->size > 0);
......
source/dnode/mnode/impl/src/mndStb.c
浏览文件 @ 842b75cd
......@@ -87,7 +87,6 @@ SSdbRaw *mndStbActionEncode(SStbObj *pStb) {
SDB_SET_INT32(pRaw, dataPos, pStb->version, _OVER)
SDB_SET_INT32(pRaw, dataPos, pStb->nextColId, _OVER)
SDB_SET_INT32(pRaw, dataPos, (int32_t)(pStb->xFilesFactor * 10000), _OVER)
SDB_SET_INT32(pRaw, dataPos, pStb->aggregationMethod, _OVER)
SDB_SET_INT32(pRaw, dataPos, pStb->delay, _OVER)
SDB_SET_INT32(pRaw, dataPos, pStb->ttl, _OVER)
SDB_SET_INT32(pRaw, dataPos, pStb->numOfColumns, _OVER)
......@@ -175,7 +174,6 @@ static SSdbRow *mndStbActionDecode(SSdbRaw *pRaw) {
int32_t xFilesFactor = 0;
SDB_GET_INT32(pRaw, dataPos, &xFilesFactor, _OVER)
pStb->xFilesFactor = xFilesFactor / 10000.0f;
SDB_GET_INT32(pRaw, dataPos, &pStb->aggregationMethod, _OVER)
SDB_GET_INT32(pRaw, dataPos, &pStb->delay, _OVER)
SDB_GET_INT32(pRaw, dataPos, &pStb->ttl, _OVER)
SDB_GET_INT32(pRaw, dataPos, &pStb->numOfColumns, _OVER)
......@@ -404,7 +402,7 @@ static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pSt
req.name = (char *)tNameGetTableName(&name);
req.ttl = 0;
req.keep = 0;
req.rollup = pStb->aggregationMethod > -1 ? 1 : 0;
req.rollup = pStb->pAst1 > 0 ? 1 : 0;
req.type = TD_SUPER_TABLE;
req.stbCfg.suid = pStb->uid;
req.stbCfg.nCols = pStb->numOfColumns;
......@@ -433,29 +431,15 @@ static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pSt
pRSmaParam->xFilesFactor = pStb->xFilesFactor;
pRSmaParam->delay = pStb->delay;
pRSmaParam->nFuncIds = 1; // only 1 aggregation method supported currently
pRSmaParam->pFuncIds = (func_id_t *)taosMemoryCalloc(pRSmaParam->nFuncIds, sizeof(func_id_t));
if (pRSmaParam->pFuncIds == NULL) {
taosMemoryFreeClear(req.stbCfg.pRSmaParam);
taosMemoryFreeClear(req.stbCfg.pSchema);
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
for (int32_t f = 0; f < pRSmaParam->nFuncIds; ++f) {
*(pRSmaParam->pFuncIds + f) = pStb->aggregationMethod;
}
if (pStb->ast1Len > 0) {
if (mndConvertRSmaTask(pStb->pAst1, 0, 0, &pRSmaParam->qmsg1, &pRSmaParam->qmsg1Len) != TSDB_CODE_SUCCESS) {
taosMemoryFreeClear(pRSmaParam->pFuncIds);
taosMemoryFreeClear(req.stbCfg.pRSmaParam);
taosMemoryFreeClear(req.stbCfg.pSchema);
return NULL;
}
}
if (pStb->ast2Len > 0) {
int32_t qmsgLen2 = 0;
if (mndConvertRSmaTask(pStb->pAst2, 0, 0, &pRSmaParam->qmsg2, &pRSmaParam->qmsg2Len) != TSDB_CODE_SUCCESS) {
taosMemoryFreeClear(pRSmaParam->pFuncIds);
taosMemoryFreeClear(pRSmaParam->qmsg1);
taosMemoryFreeClear(req.stbCfg.pRSmaParam);
taosMemoryFreeClear(req.stbCfg.pSchema);
......@@ -470,7 +454,6 @@ static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pSt
SMsgHead *pHead = taosMemoryMalloc(contLen);
if (pHead == NULL) {
if (pRSmaParam) {
taosMemoryFreeClear(pRSmaParam->pFuncIds);
taosMemoryFreeClear(pRSmaParam->qmsg1);
taosMemoryFreeClear(pRSmaParam->qmsg2);
taosMemoryFreeClear(pRSmaParam);
......@@ -488,7 +471,6 @@ static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pSt
*pContLen = contLen;
if (pRSmaParam) {
taosMemoryFreeClear(pRSmaParam->pFuncIds);
taosMemoryFreeClear(pRSmaParam->qmsg1);
taosMemoryFreeClear(pRSmaParam->qmsg2);
taosMemoryFreeClear(pRSmaParam);
......@@ -706,7 +688,6 @@ static int32_t mndCreateStb(SMnode *pMnode, SNodeMsg *pReq, SMCreateStbReq *pCre
stbObj.version = 1;
stbObj.nextColId = 1;
stbObj.xFilesFactor = pCreate->xFilesFactor;
stbObj.aggregationMethod = pCreate->aggregationMethod;
stbObj.delay = pCreate->delay;
stbObj.ttl = pCreate->ttl;
stbObj.numOfColumns = pCreate->numOfColumns;
......
source/dnode/mnode/impl/src/mndTopic.c
浏览文件 @ 842b75cd
......@@ -282,10 +282,10 @@ static int32_t mndCreateTopic(SMnode *pMnode, SNodeMsg *pReq, SCMCreateTopicReq
topicObj.version = 1;
topicObj.sql = strdup(pCreate->sql);
topicObj.sqlLen = strlen(pCreate->sql) + 1;
topicObj.ast = strdup(pCreate->ast);
topicObj.astLen = strlen(pCreate->ast) + 1;
if (pCreate->ast && pCreate->ast[0]) {
topicObj.ast = strdup(pCreate->ast);
topicObj.astLen = strlen(pCreate->ast) + 1;
topicObj.subType = TOPIC_SUB_TYPE__TABLE;
topicObj.withTbName = 0;
topicObj.withSchema = 0;
......@@ -314,6 +314,9 @@ static int32_t mndCreateTopic(SMnode *pMnode, SNodeMsg *pReq, SCMCreateTopicReq
return -1;
}
} else {
topicObj.ast = strdup("");
topicObj.astLen = 1;
topicObj.physicalPlan = strdup("");
topicObj.subType = TOPIC_SUB_TYPE__DB;
topicObj.withTbName = 1;
topicObj.withSchema = 1;
......
source/dnode/vnode/inc/vnode.h
浏览文件 @ 842b75cd
......@@ -61,6 +61,9 @@ int32_t vnodeSync(SVnode *pVnode);
int32_t vnodeGetLoad(SVnode *pVnode, SVnodeLoad *pLoad);
int vnodeValidateTableHash(SVnodeCfg *pVnodeOptions, char *tableFName);
int32_t vnodeStart(SVnode *pVnode);
void vnodeStop(SVnode *pVnode);
int64_t vnodeGetSyncHandle(SVnode *pVnode);
void vnodeGetSnapshot(SVnode *pVnode, SSnapshot *pSnapshot);
......@@ -162,11 +165,6 @@ typedef struct {
uint64_t uid;
} STableKeyInfo;
// sync integration
void vnodeSyncSetQ(SVnode *pVnode, void *qHandle);
void vnodeSyncSetRpc(SVnode *pVnode, void *rpcHandle);
int32_t vnodeSyncStart(SVnode *pVnode);
#ifdef __cplusplus
}
#endif
......
source/dnode/vnode/src/inc/meta.h
浏览文件 @ 842b75cd
......@@ -51,7 +51,7 @@ static FORCE_INLINE tb_uid_t metaGenerateUid(SMeta* pMeta) { return tGenIdPI64()
#define META_CHILD_TABLE TD_CHILD_TABLE
#define META_NORMAL_TABLE TD_NORMAL_TABLE
int metaCreateTable(SMeta* pMeta, STbCfg* pTbCfg);
int metaCreateTable(SMeta* pMeta, STbCfg* pTbCfg, STbDdlH* pHandle);
int metaDropTable(SMeta* pMeta, tb_uid_t uid);
int metaCommit(SMeta* pMeta);
int32_t metaCreateTSma(SMeta* pMeta, SSmaCfg* pCfg);
......@@ -74,7 +74,7 @@ tb_uid_t metaCtbCursorNext(SMCtbCursor* pCtbCur);
// SMetaDB
int metaOpenDB(SMeta* pMeta);
void metaCloseDB(SMeta* pMeta);
int metaSaveTableToDB(SMeta* pMeta, STbCfg* pTbCfg);
int metaSaveTableToDB(SMeta* pMeta, STbCfg* pTbCfg, STbDdlH* pHandle);
int metaRemoveTableFromDb(SMeta* pMeta, tb_uid_t uid);
int metaSaveSmaToDB(SMeta* pMeta, STSma* pTbCfg);
int metaRemoveSmaFromDb(SMeta* pMeta, int64_t indexUid);
......
source/dnode/vnode/src/inc/tsdb.h
浏览文件 @ 842b75cd
......@@ -57,7 +57,6 @@ int32_t tsdbDropTSmaData(STsdb *pTsdb, int64_t indexUid);
int32_t tsdbInsertRSmaData(STsdb *pTsdb, char *msg);
void tsdbCleanupReadHandle(tsdbReaderT queryHandle);
int32_t tdScanAndConvertSubmitMsg(SSubmitReq *pMsg);
typedef enum {
TSDB_FILE_HEAD = 0, // .head
TSDB_FILE_DATA, // .data
......
source/dnode/vnode/src/inc/tsdbSma.h 0 → 100644
浏览文件 @ 842b75cd
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_VNODE_TSDB_SMA_H_
#define _TD_VNODE_TSDB_SMA_H_
#include "os.h"
#include "thash.h"
#include "tmsg.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef int32_t (*__tb_ddl_fn_t)(void *ahandle, void **result, void *p1, void *p2);
struct STbDdlH {
void *ahandle;
void *result;
__tb_ddl_fn_t fp;
};
typedef struct {
tb_uid_t suid;
SArray *tbUids;
SHashObj *uidHash;
} STbUidStore;
static FORCE_INLINE int32_t tsdbUidStoreInit(STbUidStore **pStore) {
ASSERT(*pStore == NULL);
*pStore = taosMemoryCalloc(1, sizeof(STbUidStore));
if (*pStore == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return TSDB_CODE_FAILED;
}
return TSDB_CODE_SUCCESS;
}
int32_t tsdbUidStorePut(STbUidStore *pStore, tb_uid_t suid, tb_uid_t *uid);
void tsdbUidStoreDestory(STbUidStore *pStore);
void *tsdbUidStoreFree(STbUidStore *pStore);
int32_t tsdbRegisterRSma(STsdb *pTsdb, SMeta *pMeta, SVCreateTbReq *pReq);
int32_t tsdbFetchTbUidList(void *pTsdb, void **result, void *suid, void *uid);
int32_t tsdbUpdateTbUidList(STsdb *pTsdb, STbUidStore *pUidStore);
int32_t tsdbTriggerRSma(STsdb *pTsdb, SMeta *pMeta, void *pMsg, int32_t inputType);
#ifdef __cplusplus
}
#endif
#endif /*_TD_VNODE_TSDB_SMA_H_*/
\ No newline at end of file
source/dnode/vnode/src/inc/vnodeInt.h
浏览文件 @ 842b75cd
......@@ -103,6 +103,8 @@ struct SVnode {
#define TD_VID(PVNODE) (PVNODE)->config.vgId
typedef struct STbDdlH STbDdlH;
// sma
void smaHandleRes(void *pVnode, int64_t smaId, const SArray *data);
......@@ -116,6 +118,8 @@ void smaHandleRes(void *pVnode, int64_t smaId, const SArray *data);
#include "vnodeSync.h"
#include "tsdbSma.h"
#ifdef __cplusplus
}
#endif
......
source/dnode/vnode/src/meta/metaTDBImpl.c
浏览文件 @ 842b75cd
......@@ -250,7 +250,7 @@ void metaCloseDB(SMeta *pMeta) {
}
}
int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg) {
int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg, STbDdlH *pHandle) {
tb_uid_t uid;
SMetaDB *pMetaDb;
void *pKey;
......@@ -349,6 +349,12 @@ int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg) {
if (ret < 0) {
return -1;
}
// child table handle for rsma
if (pHandle && pHandle->fp) {
if (((*pHandle->fp)(pHandle->ahandle, &pHandle->result, &ctbIdxKey.suid, &uid)) < 0) {
return -1;
};
}
} else if (pTbCfg->type == META_NORMAL_TABLE) {
pKey = &uid;
kLen = sizeof(uid);
......
source/dnode/vnode/src/meta/metaTable.c
浏览文件 @ 842b75cd
......@@ -15,7 +15,7 @@
#include "vnodeInt.h"
int metaCreateTable(SMeta *pMeta, STbCfg *pTbCfg) {
int metaCreateTable(SMeta *pMeta, STbCfg *pTbCfg, STbDdlH *pHandle) {
// Validate the tbOptions
// if (metaValidateTbCfg(pMeta, pTbCfg) < 0) {
// // TODO: handle error
......@@ -24,7 +24,7 @@ int metaCreateTable(SMeta *pMeta, STbCfg *pTbCfg) {
// TODO: add atomicity
if (metaSaveTableToDB(pMeta, pTbCfg) < 0) {
if (metaSaveTableToDB(pMeta, pTbCfg, pHandle) < 0) {
// TODO: handle error
return -1;
}
......
source/dnode/vnode/src/tq/tq.c
浏览文件 @ 842b75cd
......@@ -411,8 +411,10 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
SMqDataBlkRsp rsp = {0};
rsp.reqOffset = pReq->currentOffset;
rsp.withSchema = pExec->withSchema;
rsp.blockData = taosArrayInit(0, sizeof(void*));
rsp.blockDataLen = taosArrayInit(0, sizeof(int32_t));
rsp.blockSchema = taosArrayInit(0, sizeof(void*));
while (1) {
consumerEpoch = atomic_load_32(&pExec->epoch);
......@@ -512,6 +514,12 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
ASSERT(actualLen <= dataStrLen);
taosArrayPush(rsp.blockDataLen, &actualLen);
taosArrayPush(rsp.blockData, &buf);
if (pExec->withSchema) {
SSchemaWrapper* pSW = tCloneSSchemaWrapper(pExec->pExecReader[workerId]->pSchemaWrapper);
taosArrayPush(rsp.blockSchema, &pSW);
}
rsp.blockNum++;
}
// db subscribe
......@@ -540,6 +548,10 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
ASSERT(actualLen <= dataStrLen);
taosArrayPush(rsp.blockDataLen, &actualLen);
taosArrayPush(rsp.blockData, &buf);
SSchemaWrapper* pSW = tCloneSSchemaWrapper(pExec->pExecReader[workerId]->pSchemaWrapper);
taosArrayPush(rsp.blockSchema, &pSW);
rsp.blockNum++;
}
} else {
......@@ -586,6 +598,8 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
// TODO destroy
taosArrayDestroy(rsp.blockData);
taosArrayDestroy(rsp.blockDataLen);
taosArrayDestroyP(rsp.blockSchema, (FDelete)tDeleteSSchemaWrapper);
return 0;
}
......@@ -827,12 +841,16 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
pExec->pWalReader = walOpenReadHandle(pTq->pVnode->pWal);
for (int32_t i = 0; i < 5; i++) {
pExec->pExecReader[i] = tqInitSubmitMsgScanner(pTq->pVnode->pMeta);
SReadHandle handle = {
.reader = pExec->pExecReader[i],
.meta = pTq->pVnode->pMeta,
};
pExec->task[i] = qCreateStreamExecTaskInfo(pExec->qmsg, &handle);
ASSERT(pExec->task[i]);
if (pExec->subType == TOPIC_SUB_TYPE__TABLE) {
SReadHandle handle = {
.reader = pExec->pExecReader[i],
.meta = pTq->pVnode->pMeta,
};
pExec->task[i] = qCreateStreamExecTaskInfo(pExec->qmsg, &handle);
ASSERT(pExec->task[i]);
} else {
pExec->task[i] = NULL;
}
}
taosHashPut(pTq->execs, req.subKey, strlen(req.subKey), pExec, sizeof(STqExec));
return 0;
......
source/dnode/vnode/src/tq/tqRead.c
浏览文件 @ 842b75cd
......@@ -65,7 +65,9 @@ bool tqNextDataBlock(STqReadHandle* pHandle) {
/*pHandle->pBlock->uid = htobe64(pHandle->pBlock->uid);*/
/*if (pHandle->tbUid == pHandle->pBlock->uid) {*/
ASSERT(pHandle->tbIdHash);
if (pHandle->tbIdHash == NULL) {
return true;
}
void* ret = taosHashGet(pHandle->tbIdHash, &pHandle->pBlock->uid, sizeof(int64_t));
if (ret != NULL) {
/*printf("retrieve one tb %ld\n", pHandle->pBlock->uid);*/
......@@ -107,26 +109,15 @@ int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* p
*pNumOfRows = pHandle->pBlock->numOfRows;
int32_t colNumNeed = taosArrayGetSize(pHandle->pColIdList);
if (colNumNeed > pSchemaWrapper->nCols) {
colNumNeed = pSchemaWrapper->nCols;
}
*ppCols = taosArrayInit(colNumNeed, sizeof(SColumnInfoData));
if (*ppCols == NULL) {
return -1;
}
if (colNumNeed == 0) {
*ppCols = taosArrayInit(pSchemaWrapper->nCols, sizeof(SColumnInfoData));
if (*ppCols == NULL) {
return -1;
}
int32_t colMeta = 0;
int32_t colNeed = 0;
while (colMeta < pSchemaWrapper->nCols && colNeed < colNumNeed) {
SSchema* pColSchema = &pSchemaWrapper->pSchema[colMeta];
col_id_t colIdSchema = pColSchema->colId;
col_id_t colIdNeed = *(col_id_t*)taosArrayGet(pHandle->pColIdList, colNeed);
if (colIdSchema < colIdNeed) {
colMeta++;
} else if (colIdSchema > colIdNeed) {
colNeed++;
} else {
int32_t colMeta = 0;
while (colMeta < pSchemaWrapper->nCols) {
SSchema* pColSchema = &pSchemaWrapper->pSchema[colMeta];
SColumnInfoData colInfo = {0};
colInfo.info.bytes = pColSchema->bytes;
colInfo.info.colId = pColSchema->colId;
......@@ -137,7 +128,40 @@ int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* p
}
taosArrayPush(*ppCols, &colInfo);
colMeta++;
colNeed++;
}
} else {
if (colNumNeed > pSchemaWrapper->nCols) {
colNumNeed = pSchemaWrapper->nCols;
}
*ppCols = taosArrayInit(colNumNeed, sizeof(SColumnInfoData));
if (*ppCols == NULL) {
return -1;
}
int32_t colMeta = 0;
int32_t colNeed = 0;
while (colMeta < pSchemaWrapper->nCols && colNeed < colNumNeed) {
SSchema* pColSchema = &pSchemaWrapper->pSchema[colMeta];
col_id_t colIdSchema = pColSchema->colId;
col_id_t colIdNeed = *(col_id_t*)taosArrayGet(pHandle->pColIdList, colNeed);
if (colIdSchema < colIdNeed) {
colMeta++;
} else if (colIdSchema > colIdNeed) {
colNeed++;
} else {
SColumnInfoData colInfo = {0};
colInfo.info.bytes = pColSchema->bytes;
colInfo.info.colId = pColSchema->colId;
colInfo.info.type = pColSchema->type;
if (colInfoDataEnsureCapacity(&colInfo, 0, *pNumOfRows) < 0) {
goto FAIL;
}
taosArrayPush(*ppCols, &colInfo);
colMeta++;
colNeed++;
}
}
}
......
source/dnode/vnode/src/tsdb/tsdbMain.c
浏览文件 @ 842b75cd
......@@ -81,8 +81,8 @@ static STsdb *tsdbNew(const char *path, SVnode *pVnode, const STsdbCfg *pTsdbCfg
static void tsdbFree(STsdb *pTsdb) {
if (pTsdb) {
// tsdbFreeSmaEnv(REPO_TSMA_ENV(pTsdb));
// tsdbFreeSmaEnv(REPO_RSMA_ENV(pTsdb));
tsdbFreeSmaEnv(REPO_TSMA_ENV(pTsdb));
tsdbFreeSmaEnv(REPO_RSMA_ENV(pTsdb));
tsdbFreeFS(pTsdb->fs);
taosMemoryFreeClear(pTsdb->path);
taosMemoryFree(pTsdb);
......
source/dnode/vnode/src/tsdb/tsdbRead.c
浏览文件 @ 842b75cd
......@@ -3618,6 +3618,8 @@ int32_t tsdbQuerySTableByTagCond(void* pMeta, uint64_t uid, TSKEY skey, const ch
tsdbError("%p failed to get stable, uid:%" PRIu64 ", TID:0x%" PRIx64 " QID:0x%" PRIx64, pMeta, uid, taskId, reqId);
terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
goto _error;
} else {
tsdbDebug("%p succeed to get stable, uid:%" PRIu64 ", TID:0x%" PRIx64 " QID:0x%" PRIx64, pMeta, uid, taskId, reqId);
}
if (pTbCfg->type != META_SUPER_TABLE) {
......
source/dnode/vnode/src/vnd/vnodeOpen.c
浏览文件 @ 842b75cd
......@@ -14,6 +14,7 @@
*/
#include "vnodeInt.h"
#include "vnodeSync.h"
int vnodeCreate(const char *path, SVnodeCfg *pCfg, STfs *pTfs) {
SVnodeInfo info = {0};
......@@ -171,6 +172,16 @@ void vnodeClose(SVnode *pVnode) {
}
}
// start the sync timer after the queue is ready
int32_t vnodeStart(SVnode *pVnode) {
vnodeSyncSetQ(pVnode, NULL);
vnodeSyncSetRpc(pVnode, NULL);
vnodeSyncStart(pVnode);
return 0;
}
void vnodeStop(SVnode *pVnode) {}
int64_t vnodeGetSyncHandle(SVnode *pVnode) { return pVnode->sync; }
void vnodeGetSnapshot(SVnode *pVnode, SSnapshot *pSnapshot) { pSnapshot->lastApplyIndex = pVnode->state.committed; }
\ No newline at end of file
source/dnode/vnode/src/vnd/vnodeSvr.c
浏览文件 @ 842b75cd
......@@ -83,6 +83,7 @@ int vnodeProcessWriteReq(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRpcMsg
case TDMT_VND_SUBMIT:
pRsp->msgType = TDMT_VND_SUBMIT_RSP;
vnodeProcessSubmitReq(pVnode, ptr, pRsp);
tsdbTriggerRSma(pVnode->pTsdb, pVnode->pMeta, ptr, STREAM_DATA_TYPE_SUBMIT_BLOCK);
break;
case TDMT_VND_MQ_VG_CHANGE:
if (tqProcessVgChangeReq(pVnode->pTq, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
......@@ -101,7 +102,6 @@ int vnodeProcessWriteReq(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRpcMsg
}
} break;
case TDMT_VND_CREATE_SMA: { // timeRangeSMA
if (tsdbCreateTSma(pVnode->pTsdb, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead))) < 0) {
// TODO
}
......@@ -277,24 +277,16 @@ int vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
static int vnodeProcessCreateStbReq(SVnode *pVnode, void *pReq) {
SVCreateTbReq vCreateTbReq = {0};
tDeserializeSVCreateTbReq(pReq, &vCreateTbReq);
if (metaCreateTable(pVnode->pMeta, &(vCreateTbReq)) < 0) {
if (metaCreateTable(pVnode->pMeta, &(vCreateTbReq), NULL) < 0) {
// TODO
return -1;
}
// TODO: remove the debug log
SRSmaParam *param = vCreateTbReq.stbCfg.pRSmaParam;
if (param) {
printf("qmsg1 len = %d, body = %s\n", param->qmsg1 ? (int32_t)strlen(param->qmsg1) : 0,
param->qmsg1 ? param->qmsg1 : "");
printf("qmsg1 len = %d, body = %s\n", param->qmsg2 ? (int32_t)strlen(param->qmsg2) : 0,
param->qmsg2 ? param->qmsg2 : "");
}
tsdbRegisterRSma(pVnode->pTsdb, pVnode->pMeta, &vCreateTbReq);
taosMemoryFree(vCreateTbReq.stbCfg.pSchema);
taosMemoryFree(vCreateTbReq.stbCfg.pTagSchema);
if (vCreateTbReq.stbCfg.pRSmaParam) {
taosMemoryFree(vCreateTbReq.stbCfg.pRSmaParam->pFuncIds);
taosMemoryFree(vCreateTbReq.stbCfg.pRSmaParam->qmsg1);
taosMemoryFree(vCreateTbReq.stbCfg.pRSmaParam->qmsg2);
taosMemoryFree(vCreateTbReq.stbCfg.pRSmaParam);
......@@ -309,6 +301,13 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, SRpcMsg *pMsg, void *pReq, SR
SVCreateTbBatchRsp vCreateTbBatchRsp = {0};
tDeserializeSVCreateTbBatchReq(pReq, &vCreateTbBatchReq);
int reqNum = taosArrayGetSize(vCreateTbBatchReq.pArray);
STbDdlH ddlHandle = {
.ahandle = pVnode->pTsdb,
.result = NULL,
.fp = tsdbFetchTbUidList,
};
for (int i = 0; i < reqNum; i++) {
SVCreateTbReq *pCreateTbReq = taosArrayGet(vCreateTbBatchReq.pArray, i);
......@@ -324,7 +323,7 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, SRpcMsg *pMsg, void *pReq, SR
taosArrayPush(vCreateTbBatchRsp.rspList, &rsp);
}
if (metaCreateTable(pVnode->pMeta, pCreateTbReq) < 0) {
if (metaCreateTable(pVnode->pMeta, pCreateTbReq, &ddlHandle) < 0) {
// TODO: handle error
vError("vgId:%d, failed to create table: %s", TD_VID(pVnode), pCreateTbReq->name);
}
......@@ -334,7 +333,6 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, SRpcMsg *pMsg, void *pReq, SR
taosMemoryFree(pCreateTbReq->stbCfg.pSchema);
taosMemoryFree(pCreateTbReq->stbCfg.pTagSchema);
if (pCreateTbReq->stbCfg.pRSmaParam) {
taosMemoryFree(pCreateTbReq->stbCfg.pRSmaParam->pFuncIds);
taosMemoryFree(pCreateTbReq->stbCfg.pRSmaParam);
}
} else if (pCreateTbReq->type == TD_CHILD_TABLE) {
......@@ -342,12 +340,13 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, SRpcMsg *pMsg, void *pReq, SR
} else {
taosMemoryFree(pCreateTbReq->ntbCfg.pSchema);
if (pCreateTbReq->ntbCfg.pRSmaParam) {
taosMemoryFree(pCreateTbReq->ntbCfg.pRSmaParam->pFuncIds);
taosMemoryFree(pCreateTbReq->ntbCfg.pRSmaParam);
}
}
}
tsdbUpdateTbUidList(pVnode->pTsdb, ddlHandle.result);
vTrace("vgId:%d process create %" PRIzu " tables", TD_VID(pVnode), taosArrayGetSize(vCreateTbBatchReq.pArray));
taosArrayDestroy(vCreateTbBatchReq.pArray);
if (vCreateTbBatchRsp.rspList) {
......@@ -371,7 +370,6 @@ static int vnodeProcessAlterStbReq(SVnode *pVnode, void *pReq) {
taosMemoryFree(vAlterTbReq.stbCfg.pSchema);
taosMemoryFree(vAlterTbReq.stbCfg.pTagSchema);
if (vAlterTbReq.stbCfg.pRSmaParam) {
taosMemoryFree(vAlterTbReq.stbCfg.pRSmaParam->pFuncIds);
taosMemoryFree(vAlterTbReq.stbCfg.pRSmaParam);
}
taosMemoryFree(vAlterTbReq.name);
......
source/dnode/vnode/test/tsdbSmaTest.cpp
浏览文件 @ 842b75cd
......@@ -407,7 +407,7 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
}
}
EXPECT_EQ(tdScanAndConvertSubmitMsg(pMsg), TSDB_CODE_SUCCESS);
// EXPECT_EQ(tdScanAndConvertSubmitMsg(pMsg), TSDB_CODE_SUCCESS);
EXPECT_EQ(tsdbUpdateSmaWindow(pTsdb, pMsg, 0), 0);
......
source/libs/executor/src/executorimpl.c
浏览文件 @ 842b75cd
......@@ -1065,6 +1065,7 @@ static int32_t doSetInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCt
SInputColumnInfoData* pInput = &pCtx[i].input;
pInput->uid = pBlock->info.uid;
pInput->colDataAggIsSet = false;
SExprInfo* pOneExpr = &pOperator->pExpr[i];
for (int32_t j = 0; j < pOneExpr->base.numOfParams; ++j) {
......@@ -3749,6 +3750,9 @@ int32_t setSDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadI
}
memcpy(pColInfoData->pData, pStart, colLen[i]);
//TODO setting this flag to true temporarily so aggregate function on stable will
//examine NULL value for non-primary key column
pColInfoData->hasNull = true;
pStart += colLen[i];
}
} else { // extract data according to pColList
......
source/libs/function/inc/functionMgtInt.h
浏览文件 @ 842b75cd
......@@ -41,7 +41,9 @@ extern "C" {
#define FUNC_MGT_TEST_MASK(val, mask) (((val) & (mask)) != 0)
#define FUNC_UDF_ID_START_OFFSET_VAL 5000
#define FUNC_UDF_ID_START 5000
#define FUNC_AGGREGATE_UDF_ID 5001
#define FUNC_SCALAR_UDF_ID 5002
extern const int funcMgtUdfNum;
......
source/libs/function/inc/tudf.h
浏览文件 @ 842b75cd
......@@ -42,19 +42,7 @@ enum {
UDFC_CODE_INVALID_STATE = -5
};
typedef void *UdfcFuncHandle;
/**
* create udfd proxy, called once in process that call setupUdf/callUdfxxx/teardownUdf
* @return error code
*/
int32_t udfcOpen();
/**
* destroy udfd proxy
* @return error code
*/
int32_t udfcClose();
/**
......
source/libs/function/src/builtins.c
浏览文件 @ 842b75cd
此差异已折叠。
source/libs/function/src/udfd.c
浏览文件 @ 842b75cd
......@@ -548,7 +548,7 @@ static int32_t udfdParseArgs(int32_t argc, char *argv[]) {
static int32_t udfdInitLog() {
char logName[12] = {0};
snprintf(logName, sizeof(logName), "%slog", "udfd");
return taosCreateLog(logName, 1, configDir, NULL, NULL, NULL, 0);
return taosCreateLog(logName, 1, configDir, NULL, NULL, NULL, NULL, 0);
}
void udfdCtrlAllocBufCb(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf) {
......@@ -656,7 +656,7 @@ int main(int argc, char *argv[]) {
return -1;
}
if (taosInitCfg(configDir, NULL, NULL, NULL, 0) != 0) {
if (taosInitCfg(configDir, NULL, NULL, NULL, NULL, 0) != 0) {
fnError("failed to start since read config error");
return -1;
}
......
source/libs/function/test/runUdf.c
浏览文件 @ 842b75cd
此差异已折叠。
source/libs/nodes/inc/nodesUtil.h
浏览文件 @ 842b75cd
此差异已折叠。
source/libs/parser/inc/parAst.h
浏览文件 @ 842b75cd
此差异已折叠。
source/libs/parser/inc/parInt.h
浏览文件 @ 842b75cd
此差异已折叠。
source/libs/parser/inc/parToken.h
浏览文件 @ 842b75cd
此差异已折叠。
source/libs/parser/inc/parUtil.h
浏览文件 @ 842b75cd
此差异已折叠。
source/libs/parser/inc/sql.y
浏览文件 @ 842b75cd
此差异已折叠。
source/libs/parser/src/parInsert.c
浏览文件 @ 842b75cd
此差异已折叠。
source/libs/parser/src/parUtil.c
浏览文件 @ 842b75cd
此差异已折叠。
source/libs/parser/src/parser.c
浏览文件 @ 842b75cd
此差异已折叠。
source/libs/parser/src/sql.c
浏览文件 @ 842b75cd
此差异已折叠。
source/libs/planner/inc/planInt.h
浏览文件 @ 842b75cd
此差异已折叠。
source/libs/planner/src/planner.c
浏览文件 @ 842b75cd
此差异已折叠。
source/libs/planner/test/planSTableTest.cpp 0 → 100644
浏览文件 @ 842b75cd
此差异已折叠。
source/libs/scalar/src/scalar.c
浏览文件 @ 842b75cd
此差异已折叠。
source/libs/scalar/src/sclfunc.c
浏览文件 @ 842b75cd
此差异已折叠。
source/libs/sync/CMakeLists.txt
浏览文件 @ 842b75cd
此差异已折叠。
source/libs/sync/src/syncMain.c
浏览文件 @ 842b75cd
此差异已折叠。
source/libs/sync/test/test.sh 0 → 100644
浏览文件 @ 842b75cd
此差异已折叠。
source/util/src/tconfig.c
浏览文件 @ 842b75cd
此差异已折叠。
source/util/src/tenv.c 0 → 100644
浏览文件 @ 842b75cd
此差异已折叠。
source/util/src/terror.c
浏览文件 @ 842b75cd
此差异已折叠。
source/util/src/tjson.c
浏览文件 @ 842b75cd
此差异已折叠。
source/util/test/CMakeLists.txt
浏览文件 @ 842b75cd
此差异已折叠。
source/util/test/cfgTest.cpp
浏览文件 @ 842b75cd
此差异已折叠。
tests/system-test/2-query/Now.py 0 → 100644
浏览文件 @ 842b75cd
此差异已折叠。
tests/system-test/2-query/To_iso8061.py 0 → 100644
浏览文件 @ 842b75cd
此差异已折叠。
tests/system-test/2-query/Today.py 0 → 100644
浏览文件 @ 842b75cd
此差异已折叠。
tests/system-test/2-query/timezone.py 0 → 100644
浏览文件 @ 842b75cd
此差异已折叠。
tests/system-test/fulltest.sh
浏览文件 @ 842b75cd
此差异已折叠。
tests/tsim/src/simSystem.c
浏览文件 @ 842b75cd
此差异已折叠。
tools/shell/CMakeLists.txt
浏览文件 @ 842b75cd
此差异已折叠。
tools/shell/src/shellCommand.c
浏览文件 @ 842b75cd
此差异已折叠。
tools/shell/src/shellEngine.c
浏览文件 @ 842b75cd
此差异已折叠。
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或