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

Merge branch '3.0' into feature/TD-11274-3.0

This commit is contained in:
Cary Xu 2022-06-01 15:07:00 +08:00
parent 1e47de0523 a5c0802bb4
commit 4d73981e22
63 changed files with 2678 additions and 3813 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2452
docs-cn/12-taos-sql/07-function.md
View File

File diff suppressed because it is too large Load Diff

3
docs-cn/12-taos-sql/12-keywords.md
View File

@ -93,10 +93,13 @@ title: TDengine 参数限制与保留关键字
`TBNAME` 可以视为超级表中一个特殊的标签，代表子表的表名。
获取一个超级表所有的子表名及相关的标签信息：
```mysql
SELECT TBNAME, location FROM meters;
```
统计超级表下辖子表数量：
```mysql
SELECT COUNT(TBNAME) FROM meters;
```

2451
docs-en/12-taos-sql/07-function.md
View File

File diff suppressed because it is too large Load Diff

1
docs-en/12-taos-sql/12-keywords.md
View File

@ -56,6 +56,7 @@ There are about 200 keywords reserved by TDengine, they can't be used as the nam
Get the table name and tag values of all subtables in a STable.
```mysql
SELECT TBNAME, location FROM meters;
```
Count the number of subtables in a STable.
```mysql

45
example/src/tmq.c
View File

@ -165,7 +165,6 @@ 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, "msg.with.table.name", "true");
tmq_conf_set(conf, "enable.auto.commit", "false");
tmq_conf_set_auto_commit_cb(conf, tmq_commit_cb_print, NULL);
@ -191,20 +190,18 @@ void basic_consume_loop(tmq_t* tmq, tmq_list_t* topics) {
return;
}
int32_t cnt = 0;
/*clock_t startTime = clock();*/
while (running) {
TAOS_RES* tmqmessage = tmq_consumer_poll(tmq, 0);
if (tmqmessage) {
cnt++;
msg_process(tmqmessage);
if (cnt >= 2) break;
/*printf("get data\n");*/
/*msg_process(tmqmessage);*/
taos_free_result(tmqmessage);
/*} else {*/
/*break;*/
}
}
/*clock_t endTime = clock();*/
/*printf("log cnt: %d %f s\n", cnt, (double)(endTime - startTime) / CLOCKS_PER_SEC);*/
err = tmq_consumer_close(tmq);
if (err)
@ -253,39 +250,6 @@ void sync_consume_loop(tmq_t* tmq, tmq_list_t* topics) {
fprintf(stderr, "%% Consumer closed\n");
}
void perf_loop(tmq_t* tmq, tmq_list_t* topics) {
tmq_resp_err_t err;
if ((err = tmq_subscribe(tmq, topics))) {
fprintf(stderr, "%% Failed to start consuming topics: %s\n", tmq_err2str(err));
printf("subscribe err\n");
return;
}
int32_t batchCnt = 0;
int32_t skipLogNum = 0;
clock_t startTime = clock();
while (running) {
TAOS_RES* tmqmessage = tmq_consumer_poll(tmq, 500);
if (tmqmessage) {
batchCnt++;
/*skipLogNum += tmqGetSkipLogNum(tmqmessage);*/
/*msg_process(tmqmessage);*/
taos_free_result(tmqmessage);
} else {
break;
}
}
clock_t endTime = clock();
printf("log batch cnt: %d, skip log cnt: %d, time used:%f s\n", batchCnt, skipLogNum,
(double)(endTime - startTime) / CLOCKS_PER_SEC);
err = tmq_consumer_close(tmq);
if (err)
fprintf(stderr, "%% Failed to close consumer: %s\n", tmq_err2str(err));
else
fprintf(stderr, "%% Consumer closed\n");
}
int main(int argc, char* argv[]) {
if (argc > 1) {
printf("env init\n");
@ -296,7 +260,6 @@ int main(int argc, char* argv[]) {
}
tmq_t* tmq = build_consumer();
tmq_list_t* topic_list = build_topic_list();
/*perf_loop(tmq, topic_list);*/
/*basic_consume_loop(tmq, topic_list);*/
sync_consume_loop(tmq, topic_list);
basic_consume_loop(tmq, topic_list);
/*sync_consume_loop(tmq, topic_list);*/
}

13
include/client/taos.h
View File

@ -85,6 +85,14 @@ typedef struct taosField {
int32_t bytes;
} TAOS_FIELD;
typedef struct TAOS_FIELD_E {
char name[65];
int8_t type;
uint8_t precision;
uint8_t scale;
int32_t bytes;
} TAOS_FIELD_E;
#ifdef WINDOWS
#define DLL_EXPORT __declspec(dllexport)
#else
@ -134,7 +142,10 @@ 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_MULTI_BIND *tags);
DLL_EXPORT int taos_stmt_set_tbname(TAOS_STMT *stmt, const char *name);
DLL_EXPORT int taos_stmt_set_tags(TAOS_STMT *stmt, TAOS_MULTI_BIND *tags);
DLL_EXPORT int taos_stmt_set_sub_tbname(TAOS_STMT *stmt, const char *name);
DLL_EXPORT int taos_stmt_get_tag_fields(TAOS_STMT *stmt, int* fieldNum, TAOS_FIELD_E** fields);
DLL_EXPORT int taos_stmt_get_col_fields(TAOS_STMT *stmt, int* fieldNum, TAOS_FIELD_E** fields);
DLL_EXPORT int taos_stmt_is_insert(TAOS_STMT *stmt, int *insert);
DLL_EXPORT int taos_stmt_num_params(TAOS_STMT *stmt, int *nums);
@ -230,7 +241,7 @@ DLL_EXPORT const char *tmq_err2str(tmq_resp_err_t);
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 wait_time);
DLL_EXPORT TAOS_RES *tmq_consumer_poll(tmq_t *tmq, int64_t timeout);
DLL_EXPORT tmq_resp_err_t tmq_consumer_close(tmq_t *tmq);
DLL_EXPORT tmq_resp_err_t tmq_commit_sync(tmq_t *tmq, const tmq_topic_vgroup_list_t *offsets);
DLL_EXPORT void tmq_commit_async(tmq_t *tmq, const tmq_topic_vgroup_list_t *offsets, tmq_commit_cb *cb, void *param);

2
include/common/tmsg.h
View File

@ -2438,7 +2438,7 @@ typedef struct {
int32_t epoch;
uint64_t reqId;
int64_t consumerId;
int64_t waitTime;
int64_t timeout;
int64_t currentOffset;
} SMqPollReq;

4
include/libs/parser/parser.h
View File

@ -77,8 +77,8 @@ int32_t qStmtParseQuerySql(SParseContext* pCxt, SQuery* pQuery);
int32_t qBindStmtColsValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen);
int32_t qBindStmtSingleColValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen, int32_t colIdx,
int32_t rowNum);
int32_t qBuildStmtColFields(void* pDataBlock, int32_t* fieldNum, TAOS_FIELD** fields);
int32_t qBuildStmtTagFields(void* pBlock, void* boundTags, int32_t* fieldNum, TAOS_FIELD** fields);
int32_t qBuildStmtColFields(void* pDataBlock, int32_t* fieldNum, TAOS_FIELD_E** fields);
int32_t qBuildStmtTagFields(void* pBlock, void* boundTags, int32_t* fieldNum, TAOS_FIELD_E** fields);
int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, char* tName, TAOS_MULTI_BIND* bind,
char* msgBuf, int32_t msgBufLen);
void destroyBoundColumnInfo(void* pBoundInfo);

11
include/os/osDir.h
View File

@ -33,8 +33,19 @@ extern "C" {
#ifdef WINDOWS
#define TD_TMP_DIR_PATH "C:\\Windows\\Temp\\"
#define TD_CFG_DIR_PATH "C:\\TDengine\\cfg\\"
#define TD_DATA_DIR_PATH "C:\\TDengine\\data\\"
#define TD_LOG_DIR_PATH "C:\\TDengine\\log\\"
#elif defined(_TD_DARWIN_64)
#define TD_TMP_DIR_PATH "/tmp/taosd/"
#define TD_CFG_DIR_PATH "/usr/local/etc/taos/"
#define TD_DATA_DIR_PATH "/usr/local/var/lib/taos/"
#define TD_LOG_DIR_PATH "/usr/local/var/log/taos/"
#else
#define TD_TMP_DIR_PATH "/tmp/"
#define TD_CFG_DIR_PATH "/etc/taos/"
#define TD_DATA_DIR_PATH "/var/lib/taos/"
#define TD_LOG_DIR_PATH "/var/log/taos/"
#endif
typedef struct TdDir *TdDirPtr;

2
include/util/taoserror.h
View File

@ -182,7 +182,7 @@ int32_t* taosGetErrno();
#define TSDB_CODE_MND_BNODE_ALREADY_EXIST TAOS_DEF_ERROR_CODE(0, 0x0356)
#define TSDB_CODE_MND_BNODE_NOT_EXIST TAOS_DEF_ERROR_CODE(0, 0x0357)
#define TSDB_CODE_MND_TOO_FEW_MNODES TAOS_DEF_ERROR_CODE(0, 0x0358)
#define TSDB_CODE_MND_MNODE_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0359)
#define TSDB_CODE_MND_TOO_MANY_MNODES TAOS_DEF_ERROR_CODE(0, 0x0359)
#define TSDB_CODE_MND_CANT_DROP_MASTER TAOS_DEF_ERROR_CODE(0, 0x035A)
// mnode-acct

3
include/util/tdef.h
View File

@ -253,8 +253,7 @@ typedef enum ELogicConditionType {
#define TSDB_TRANS_STAGE_LEN 12
#define TSDB_TRANS_TYPE_LEN 16
#define TSDB_TRANS_ERROR_LEN 64
#define TSDB_TRANS_DESC_LEN 128
#define TSDB_TRANS_ERROR_LEN 512
#define TSDB_STEP_NAME_LEN 32
#define TSDB_STEP_DESC_LEN 128

3
source/client/inc/clientStmt.h
View File

@ -116,8 +116,11 @@ int stmtAffectedRowsOnce(TAOS_STMT *stmt);
int stmtPrepare(TAOS_STMT *stmt, const char *sql, unsigned long length);
int stmtSetTbName(TAOS_STMT *stmt, const char *tbName);
int stmtSetTbTags(TAOS_STMT *stmt, TAOS_MULTI_BIND *tags);
int stmtGetTagFields(TAOS_STMT* stmt, int* nums, TAOS_FIELD_E** fields);
int stmtGetColFields(TAOS_STMT* stmt, int* nums, TAOS_FIELD_E** fields);
int stmtIsInsert(TAOS_STMT *stmt, int *insert);
int stmtGetParamNum(TAOS_STMT *stmt, int *nums);
int stmtGetParam(TAOS_STMT *stmt, int idx, int *type, int *bytes);
int stmtAddBatch(TAOS_STMT *stmt);
TAOS_RES *stmtUseResult(TAOS_STMT *stmt);
int stmtBindBatch(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind, int32_t colIdx);

41
source/client/src/clientMain.c
View File

@ -666,8 +666,39 @@ int taos_stmt_set_tbname(TAOS_STMT *stmt, const char *name) {
return stmtSetTbName(stmt, name);
}
int taos_stmt_set_tags(TAOS_STMT *stmt, TAOS_MULTI_BIND *tags) {
if (stmt == NULL || tags == NULL) {
tscError("NULL parameter for %s", __FUNCTION__);
terrno = TSDB_CODE_INVALID_PARA;
return terrno;
}
return stmtSetTbTags(stmt, tags);
}
int taos_stmt_set_sub_tbname(TAOS_STMT *stmt, const char *name) { return taos_stmt_set_tbname(stmt, name); }
int taos_stmt_get_tag_fields(TAOS_STMT *stmt, int* fieldNum, TAOS_FIELD_E** fields) {
if (stmt == NULL || NULL == fieldNum) {
tscError("NULL parameter for %s", __FUNCTION__);
terrno = TSDB_CODE_INVALID_PARA;
return terrno;
}
return stmtGetTagFields(stmt, fieldNum, fields);
}
int taos_stmt_get_col_fields(TAOS_STMT *stmt, int* fieldNum, TAOS_FIELD_E** fields) {
if (stmt == NULL || NULL == fieldNum) {
tscError("NULL parameter for %s", __FUNCTION__);
terrno = TSDB_CODE_INVALID_PARA;
return terrno;
}
return stmtGetColFields(stmt, fieldNum, fields);
}
int taos_stmt_bind_param(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind) {
if (stmt == NULL || bind == NULL) {
tscError("NULL parameter for %s", __FUNCTION__);
@ -772,6 +803,16 @@ int taos_stmt_num_params(TAOS_STMT *stmt, int *nums) {
return stmtGetParamNum(stmt, nums);
}
int taos_stmt_get_param(TAOS_STMT *stmt, int idx, int *type, int *bytes) {
if (stmt == NULL || type == NULL || NULL == bytes || idx < 0) {
tscError("invalid parameter for %s", __FUNCTION__);
terrno = TSDB_CODE_INVALID_PARA;
return terrno;
}
return stmtGetParam(stmt, idx, type, bytes);
}
TAOS_RES *taos_stmt_use_result(TAOS_STMT *stmt) {
if (stmt == NULL) {
tscError("NULL parameter for %s", __FUNCTION__);

121
source/client/src/clientStmt.c
View File

@ -17,7 +17,7 @@ int32_t stmtSwitchStatus(STscStmt* pStmt, STMT_STATUS newStatus) {
}
break;
case STMT_SETTAGS:
if (STMT_STATUS_NE(SETTBNAME)) {
if (STMT_STATUS_NE(SETTBNAME) && STMT_STATUS_NE(FETCH_FIELDS)) {
code = TSDB_CODE_TSC_STMT_API_ERROR;
}
break;
@ -540,6 +540,8 @@ int stmtSetTbName(TAOS_STMT* stmt, const char* tbName) {
if (pStmt->bInfo.needParse) {
strncpy(pStmt->bInfo.tbName, tbName, sizeof(pStmt->bInfo.tbName) - 1);
pStmt->bInfo.tbName[sizeof(pStmt->bInfo.tbName) - 1] = 0;
STMT_ERR_RET(stmtParseSql(pStmt));
}
return TSDB_CODE_SUCCESS;
@ -550,10 +552,6 @@ int stmtSetTbTags(TAOS_STMT* stmt, TAOS_MULTI_BIND* tags) {
STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
if (pStmt->bInfo.needParse) {
STMT_ERR_RET(stmtParseSql(pStmt));
}
if (pStmt->bInfo.inExecCache) {
return TSDB_CODE_SUCCESS;
}
@ -571,7 +569,7 @@ int stmtSetTbTags(TAOS_STMT* stmt, TAOS_MULTI_BIND* tags) {
return TSDB_CODE_SUCCESS;
}
int32_t stmtFetchTagFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD** fields) {
int stmtFetchTagFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields) {
if (STMT_TYPE_QUERY == pStmt->sql.type) {
tscError("invalid operation to get query tag fileds");
STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
@ -589,7 +587,7 @@ int32_t stmtFetchTagFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD** fiel
return TSDB_CODE_SUCCESS;
}
int32_t stmtFetchColFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD** fields) {
int stmtFetchColFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields) {
if (STMT_TYPE_QUERY == pStmt->sql.type) {
tscError("invalid operation to get query column fileds");
STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
@ -852,6 +850,71 @@ int stmtIsInsert(TAOS_STMT* stmt, int* insert) {
return TSDB_CODE_SUCCESS;
}
int stmtGetTagFields(TAOS_STMT* stmt, int* nums, TAOS_FIELD_E** fields) {
STscStmt* pStmt = (STscStmt*)stmt;
if (STMT_TYPE_QUERY == pStmt->sql.type) {
STMT_RET(TSDB_CODE_TSC_STMT_API_ERROR);
}
STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
pStmt->bInfo.needParse = false;
}
if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
taos_free_result(pStmt->exec.pRequest);
pStmt->exec.pRequest = NULL;
}
if (NULL == pStmt->exec.pRequest) {
STMT_ERR_RET(buildRequest(pStmt->taos, pStmt->sql.sqlStr, pStmt->sql.sqlLen, &pStmt->exec.pRequest));
}
if (pStmt->bInfo.needParse) {
STMT_ERR_RET(stmtParseSql(pStmt));
}
STMT_ERR_RET(stmtFetchTagFields(stmt, nums, fields));
return TSDB_CODE_SUCCESS;
}
int stmtGetColFields(TAOS_STMT* stmt, int* nums, TAOS_FIELD_E** fields) {
STscStmt* pStmt = (STscStmt*)stmt;
if (STMT_TYPE_QUERY == pStmt->sql.type) {
STMT_RET(TSDB_CODE_TSC_STMT_API_ERROR);
}
STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
pStmt->bInfo.needParse = false;
}
if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
taos_free_result(pStmt->exec.pRequest);
pStmt->exec.pRequest = NULL;
}
if (NULL == pStmt->exec.pRequest) {
STMT_ERR_RET(buildRequest(pStmt->taos, pStmt->sql.sqlStr, pStmt->sql.sqlLen, &pStmt->exec.pRequest));
}
if (pStmt->bInfo.needParse) {
STMT_ERR_RET(stmtParseSql(pStmt));
}
STMT_ERR_RET(stmtFetchColFields(stmt, nums, fields));
return TSDB_CODE_SUCCESS;
}
int stmtGetParamNum(TAOS_STMT* stmt, int* nums) {
STscStmt* pStmt = (STscStmt*)stmt;
@ -884,6 +947,50 @@ int stmtGetParamNum(TAOS_STMT* stmt, int* nums) {
return TSDB_CODE_SUCCESS;
}
int stmtGetParam(TAOS_STMT *stmt, int idx, int *type, int *bytes) {
STscStmt* pStmt = (STscStmt*)stmt;
if (STMT_TYPE_QUERY == pStmt->sql.type) {
STMT_RET(TSDB_CODE_TSC_STMT_API_ERROR);
}
STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
pStmt->bInfo.needParse = false;
}
if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
taos_free_result(pStmt->exec.pRequest);
pStmt->exec.pRequest = NULL;
}
if (NULL == pStmt->exec.pRequest) {
STMT_ERR_RET(buildRequest(pStmt->taos, pStmt->sql.sqlStr, pStmt->sql.sqlLen, &pStmt->exec.pRequest));
}
if (pStmt->bInfo.needParse) {
STMT_ERR_RET(stmtParseSql(pStmt));
}
int32_t nums = 0;
TAOS_FIELD_E *pField = NULL;
STMT_ERR_RET(stmtFetchColFields(stmt, &nums, &pField));
if (idx >= nums) {
tscError("idx %d is too big", idx);
taosMemoryFree(pField);
STMT_ERR_RET(TSDB_CODE_INVALID_PARA);
}
*type = pField[idx].type;
*bytes = pField[idx].bytes;
taosMemoryFree(pField);
return TSDB_CODE_SUCCESS;
}
TAOS_RES* stmtUseResult(TAOS_STMT* stmt) {
STscStmt* pStmt = (STscStmt*)stmt;

34
source/client/src/tmq.c
View File

@ -1243,7 +1243,7 @@ tmq_resp_err_t tmq_seek(tmq_t* tmq, const tmq_topic_vgroup_t* offset) {
return TMQ_RESP_ERR__FAIL;
}
SMqPollReq* tmqBuildConsumeReqImpl(tmq_t* tmq, int64_t waitTime, SMqClientTopic* pTopic, SMqClientVg* pVg) {
SMqPollReq* tmqBuildConsumeReqImpl(tmq_t* tmq, int64_t timeout, SMqClientTopic* pTopic, SMqClientVg* pVg) {
int64_t reqOffset;
if (pVg->currentOffset >= 0) {
reqOffset = pVg->currentOffset;
@ -1269,7 +1269,7 @@ SMqPollReq* tmqBuildConsumeReqImpl(tmq_t* tmq, int64_t waitTime, SMqClientTopic*
strcpy(pReq->subKey + tlen + 1, pTopic->topicName);
pReq->withTbName = tmq->withTbName;
pReq->waitTime = waitTime;
pReq->timeout = timeout;
pReq->consumerId = tmq->consumerId;
pReq->epoch = tmq->epoch;
pReq->currentOffset = reqOffset;
@ -1297,7 +1297,7 @@ SMqRspObj* tmqBuildRspFromWrapper(SMqPollRspWrapper* pWrapper) {
return pRspObj;
}
int32_t tmqPollImpl(tmq_t* tmq, int64_t waitTime) {
int32_t tmqPollImpl(tmq_t* tmq, int64_t timeout) {
/*printf("call poll\n");*/
for (int i = 0; i < taosArrayGetSize(tmq->clientTopics); i++) {
SMqClientTopic* pTopic = taosArrayGet(tmq->clientTopics, i);
@ -1318,7 +1318,7 @@ int32_t tmqPollImpl(tmq_t* tmq, int64_t waitTime) {
#endif
}
atomic_store_32(&pVg->vgSkipCnt, 0);
SMqPollReq* pReq = tmqBuildConsumeReqImpl(tmq, waitTime, pTopic, pVg);
SMqPollReq* pReq = tmqBuildConsumeReqImpl(tmq, timeout, pTopic, pVg);
if (pReq == NULL) {
atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);
tsem_post(&tmq->rspSem);
@ -1388,7 +1388,7 @@ int32_t tmqHandleNoPollRsp(tmq_t* tmq, SMqRspWrapper* rspWrapper, bool* pReset)
return 0;
}
SMqRspObj* tmqHandleAllRsp(tmq_t* tmq, int64_t waitTime, bool pollIfReset) {
SMqRspObj* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
while (1) {
SMqRspWrapper* rspWrapper = NULL;
taosGetQitem(tmq->qall, (void**)&rspWrapper);
@ -1428,17 +1428,17 @@ SMqRspObj* tmqHandleAllRsp(tmq_t* tmq, int64_t waitTime, bool pollIfReset) {
taosFreeQitem(rspWrapper);
if (pollIfReset && reset) {
tscDebug("consumer %ld reset and repoll", tmq->consumerId);
tmqPollImpl(tmq, waitTime);
tmqPollImpl(tmq, timeout);
}
}
}
}
TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t wait_time) {
TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t timeout) {
SMqRspObj* rspObj;
int64_t startTime = taosGetTimestampMs();
rspObj = tmqHandleAllRsp(tmq, wait_time, false);
rspObj = tmqHandleAllRsp(tmq, timeout, false);
if (rspObj) {
return (TAOS_RES*)rspObj;
}
@ -1450,16 +1450,16 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t wait_time) {
while (1) {
tmqHandleAllDelayedTask(tmq);
if (tmqPollImpl(tmq, wait_time) < 0) return NULL;
if (tmqPollImpl(tmq, timeout) < 0) return NULL;
rspObj = tmqHandleAllRsp(tmq, wait_time, false);
rspObj = tmqHandleAllRsp(tmq, timeout, false);
if (rspObj) {
return (TAOS_RES*)rspObj;
}
if (wait_time != 0) {
if (timeout != 0) {
int64_t endTime = taosGetTimestampMs();
int64_t leftTime = endTime - startTime;
if (leftTime > wait_time) {
if (leftTime > timeout) {
tscDebug("consumer %ld (epoch %d) timeout, no rsp", tmq->consumerId, tmq->epoch);
return NULL;
}
@ -1474,10 +1474,7 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t wait_time) {
tmq_resp_err_t tmq_consumer_close(tmq_t* tmq) {
if (tmq->status == TMQ_CONSUMER_STATUS__READY) {
tmq_resp_err_t rsp = tmq_commit_sync(tmq, NULL);
if (rsp == TMQ_RESP_ERR__SUCCESS) {
// TODO: free resources
return TMQ_RESP_ERR__SUCCESS;
} else {
if (rsp == TMQ_RESP_ERR__FAIL) {
return TMQ_RESP_ERR__FAIL;
}
@ -1485,10 +1482,7 @@ tmq_resp_err_t tmq_consumer_close(tmq_t* tmq) {
rsp = tmq_subscribe(tmq, lst);
tmq_list_destroy(lst);
if (rsp == TMQ_RESP_ERR__SUCCESS) {
// TODO: free resources
return TMQ_RESP_ERR__SUCCESS;
} else {
if (rsp == TMQ_RESP_ERR__FAIL) {
return TMQ_RESP_ERR__FAIL;
}
}

1
source/common/src/systable.c
View File

@ -215,7 +215,6 @@ static const SSysDbTableSchema transSchema[] = {
{.name = "create_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
{.name = "stage", .bytes = TSDB_TRANS_STAGE_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
{.name = "db", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR},
{.name = "type", .bytes = TSDB_TRANS_TYPE_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
{.name = "failed_times", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
{.name = "last_exec_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
{.name = "last_error", .bytes = (TSDB_TRANS_ERROR_LEN - 1) + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},

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

@ -54,9 +54,11 @@ typedef enum {
} EAuthOp;
typedef enum {
TRN_STEP_LOG = 1,
TRN_STEP_ACTION = 2,
} ETrnStep;
TRN_CONFLICT_NOTHING = 0,
TRN_CONFLICT_GLOBAL = 1,
TRN_CONFLICT_DB = 2,
TRN_CONFLICT_DB_INSIDE = 3,
} ETrnConflct;
typedef enum {
TRN_STAGE_PREPARE = 0,
@ -68,69 +70,15 @@ typedef enum {
TRN_STAGE_FINISHED = 6
} ETrnStage;
typedef enum {
TRN_TYPE_BASIC_SCOPE = 1000,
TRN_TYPE_CREATE_ACCT = 1001,
TRN_TYPE_CREATE_CLUSTER = 1002,
TRN_TYPE_CREATE_USER = 1003,
TRN_TYPE_ALTER_USER = 1004,
TRN_TYPE_DROP_USER = 1005,
TRN_TYPE_CREATE_FUNC = 1006,
TRN_TYPE_DROP_FUNC = 1007,
TRN_TYPE_CREATE_SNODE = 1010,
TRN_TYPE_DROP_SNODE = 1011,
TRN_TYPE_CREATE_QNODE = 1012,
TRN_TYPE_DROP_QNODE = 10013,
TRN_TYPE_CREATE_BNODE = 1014,
TRN_TYPE_DROP_BNODE = 1015,
TRN_TYPE_CREATE_MNODE = 1016,
TRN_TYPE_DROP_MNODE = 1017,
TRN_TYPE_CREATE_TOPIC = 1020,
TRN_TYPE_DROP_TOPIC = 1021,
TRN_TYPE_SUBSCRIBE = 1022,
TRN_TYPE_REBALANCE = 1023,
TRN_TYPE_COMMIT_OFFSET = 1024,
TRN_TYPE_CREATE_STREAM = 1025,
TRN_TYPE_DROP_STREAM = 1026,
TRN_TYPE_ALTER_STREAM = 1027,
TRN_TYPE_CONSUMER_LOST = 1028,
TRN_TYPE_CONSUMER_RECOVER = 1029,
TRN_TYPE_DROP_CGROUP = 1030,
TRN_TYPE_BASIC_SCOPE_END,
TRN_TYPE_GLOBAL_SCOPE = 2000,
TRN_TYPE_CREATE_DNODE = 2001,
TRN_TYPE_DROP_DNODE = 2002,
TRN_TYPE_GLOBAL_SCOPE_END,
TRN_TYPE_DB_SCOPE = 3000,
TRN_TYPE_CREATE_DB = 3001,
TRN_TYPE_ALTER_DB = 3002,
TRN_TYPE_DROP_DB = 3003,
TRN_TYPE_SPLIT_VGROUP = 3004,
TRN_TYPE_MERGE_VGROUP = 3015,
TRN_TYPE_DB_SCOPE_END,
TRN_TYPE_STB_SCOPE = 4000,
TRN_TYPE_CREATE_STB = 4001,
TRN_TYPE_ALTER_STB = 4002,
TRN_TYPE_DROP_STB = 4003,
TRN_TYPE_CREATE_SMA = 4004,
TRN_TYPE_DROP_SMA = 4005,
TRN_TYPE_STB_SCOPE_END,
} ETrnType;
typedef enum {
TRN_POLICY_ROLLBACK = 0,
TRN_POLICY_RETRY = 1,
} ETrnPolicy;
typedef enum {
TRN_EXEC_PARALLEL = 0,
TRN_EXEC_NO_PARALLEL = 1,
} ETrnExecType;
TRN_EXEC_PRARLLEL = 0,
TRN_EXEC_SERIAL = 1,
} ETrnExec;
typedef enum {
DND_REASON_ONLINE = 0,
@ -159,8 +107,8 @@ typedef struct {
int32_t id;
ETrnStage stage;
ETrnPolicy policy;
ETrnType type;
ETrnExecType parallel;
ETrnConflct conflict;
ETrnExec exec;
int32_t code;
int32_t failedTimes;
SRpcHandleInfo rpcInfo;
@ -172,10 +120,11 @@ typedef struct {
SArray* commitActions;
int64_t createdTime;
int64_t lastExecTime;
int64_t dbUid;
int32_t lastErrorAction;
int32_t lastErrorNo;
tmsg_t lastErrorMsgType;
SEpSet lastErrorEpset;
char dbname[TSDB_DB_FNAME_LEN];
char lastError[TSDB_TRANS_ERROR_LEN];
char desc[TSDB_TRANS_DESC_LEN];
int32_t startFunc;
int32_t stopFunc;
int32_t paramLen;

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

@ -34,7 +34,7 @@ typedef struct {
int32_t errCode;
int32_t acceptableCode;
int8_t stage;
int8_t isRaw;
int8_t actionType; // 0-msg, 1-raw
int8_t rawWritten;
int8_t msgSent;
int8_t msgReceived;
@ -52,7 +52,7 @@ void mndCleanupTrans(SMnode *pMnode);
STrans *mndAcquireTrans(SMnode *pMnode, int32_t transId);
void mndReleaseTrans(SMnode *pMnode, STrans *pTrans);
STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, ETrnType type, const SRpcMsg *pReq);
STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, ETrnConflct conflict, const SRpcMsg *pReq);
void mndTransDrop(STrans *pTrans);
int32_t mndTransAppendRedolog(STrans *pTrans, SSdbRaw *pRaw);
int32_t mndTransAppendUndolog(STrans *pTrans, SSdbRaw *pRaw);
@ -62,7 +62,7 @@ int32_t mndTransAppendUndoAction(STrans *pTrans, STransAction *pAction);
void mndTransSetRpcRsp(STrans *pTrans, void *pCont, int32_t contLen);
void mndTransSetCb(STrans *pTrans, ETrnFunc startFunc, ETrnFunc stopFunc, void *param, int32_t paramLen);
void mndTransSetDbInfo(STrans *pTrans, SDbObj *pDb);
void mndTransSetNoParallel(STrans *pTrans);
void mndTransSetSerial(STrans *pTrans);
int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans);
void mndTransProcessRsp(SRpcMsg *pRsp);

2
source/dnode/mnode/impl/src/mndAcct.c
View File

@ -80,7 +80,7 @@ static int32_t mndCreateDefaultAcct(SMnode *pMnode) {
mDebug("acct:%s, will be created when deploying, raw:%p", acctObj.acct, pRaw);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_CREATE_ACCT, NULL);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_NOTHING, NULL);
if (pTrans == NULL) {
mError("acct:%s, failed to create since %s", acctObj.acct, terrstr());
return -1;

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

@ -246,7 +246,7 @@ static int32_t mndCreateBnode(SMnode *pMnode, SRpcMsg *pReq, SDnodeObj *pDnode,
bnodeObj.createdTime = taosGetTimestampMs();
bnodeObj.updateTime = bnodeObj.createdTime;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_BNODE, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pReq);
if (pTrans == NULL) goto _OVER;
mDebug("trans:%d, used to create bnode:%d", pTrans->id, pCreate->dnodeId);
@ -363,7 +363,7 @@ static int32_t mndSetDropBnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, SBn
static int32_t mndDropBnode(SMnode *pMnode, SRpcMsg *pReq, SBnodeObj *pObj) {
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_DROP_BNODE, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_NOTHING, pReq);
if (pTrans == NULL) goto _OVER;
mDebug("trans:%d, used to drop bnode:%d", pTrans->id, pObj->id);

7
source/dnode/mnode/impl/src/mndCluster.c
View File

@ -179,10 +179,8 @@ static int32_t mndCreateDefaultCluster(SMnode *pMnode) {
sdbSetRawStatus(pRaw, SDB_STATUS_READY);
mDebug("cluster:%" PRId64 ", will be created when deploying, raw:%p", clusterObj.id, pRaw);
#if 0
return sdbWrite(pMnode->pSdb, pRaw);
#else
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_CREATE_CLUSTER, NULL);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_NOTHING, NULL);
if (pTrans == NULL) {
mError("cluster:%" PRId64 ", failed to create since %s", clusterObj.id, terrstr());
return -1;
@ -204,7 +202,6 @@ static int32_t mndCreateDefaultCluster(SMnode *pMnode) {
mndTransDrop(pTrans);
return 0;
#endif
}
static int32_t mndRetrieveClusters(SRpcMsg *pMsg, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {

6
source/dnode/mnode/impl/src/mndConsumer.c
View File

@ -97,7 +97,7 @@ static int32_t mndProcessConsumerLostMsg(SRpcMsg *pMsg) {
mndReleaseConsumer(pMnode, pConsumer);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_CONSUMER_LOST, pMsg);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_NOTHING, pMsg);
if (pTrans == NULL) goto FAIL;
if (mndSetConsumerCommitLogs(pMnode, pTrans, pConsumerNew) != 0) goto FAIL;
if (mndTransPrepare(pMnode, pTrans) != 0) goto FAIL;
@ -121,7 +121,7 @@ static int32_t mndProcessConsumerRecoverMsg(SRpcMsg *pMsg) {
mndReleaseConsumer(pMnode, pConsumer);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_CONSUMER_RECOVER, pMsg);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_NOTHING, pMsg);
if (pTrans == NULL) goto FAIL;
if (mndSetConsumerCommitLogs(pMnode, pTrans, pConsumerNew) != 0) goto FAIL;
if (mndTransPrepare(pMnode, pTrans) != 0) goto FAIL;
@ -403,7 +403,7 @@ static int32_t mndProcessSubscribeReq(SRpcMsg *pMsg) {
int32_t newTopicNum = taosArrayGetSize(newSub);
// check topic existance
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_SUBSCRIBE, pMsg);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_NOTHING, pMsg);
if (pTrans == NULL) goto SUBSCRIBE_OVER;
for (int32_t i = 0; i < newTopicNum; i++) {

6
source/dnode/mnode/impl/src/mndDb.c
View File

@ -545,7 +545,7 @@ static int32_t mndCreateDb(SMnode *pMnode, SRpcMsg *pReq, SCreateDbReq *pCreate,
}
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_DB, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB, pReq);
if (pTrans == NULL) goto _OVER;
mDebug("trans:%d, used to create db:%s", pTrans->id, pCreate->db);
@ -775,7 +775,7 @@ static int32_t mndSetAlterDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *
static int32_t mndAlterDb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pOld, SDbObj *pNew) {
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_ALTER_DB, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB, pReq);
if (pTrans == NULL) goto _OVER;
mDebug("trans:%d, used to alter db:%s", pTrans->id, pOld->name);
@ -1036,7 +1036,7 @@ static int32_t mndBuildDropDbRsp(SDbObj *pDb, int32_t *pRspLen, void **ppRsp, bo
static int32_t mndDropDb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb) {
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_DROP_DB, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB, pReq);
if (pTrans == NULL) goto _OVER;
mDebug("trans:%d, used to drop db:%s", pTrans->id, pDb->name);

14
source/dnode/mnode/impl/src/mndDnode.c
View File

@ -101,10 +101,7 @@ static int32_t mndCreateDefaultDnode(SMnode *pMnode) {
mDebug("dnode:%d, will be created when deploying, raw:%p", dnodeObj.id, pRaw);
#if 0
return sdbWrite(pMnode->pSdb, pRaw);
#else
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_CREATE_DNODE, NULL);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, NULL);
if (pTrans == NULL) {
mError("dnode:%s, failed to create since %s", dnodeObj.ep, terrstr());
return -1;
@ -126,7 +123,6 @@ static int32_t mndCreateDefaultDnode(SMnode *pMnode) {
mndTransDrop(pTrans);
return 0;
#endif
}
static SSdbRaw *mndDnodeActionEncode(SDnodeObj *pDnode) {
@ -260,7 +256,7 @@ int32_t mndGetDnodeSize(SMnode *pMnode) {
bool mndIsDnodeOnline(SMnode *pMnode, SDnodeObj *pDnode, int64_t curMs) {
int64_t interval = TABS(pDnode->lastAccessTime - curMs);
if (interval > 30000 * tsStatusInterval) {
if (interval > 5000 * tsStatusInterval) {
if (pDnode->rebootTime > 0) {
pDnode->offlineReason = DND_REASON_STATUS_MSG_TIMEOUT;
}
@ -488,7 +484,7 @@ static int32_t mndCreateDnode(SMnode *pMnode, SRpcMsg *pReq, SCreateDnodeReq *pC
memcpy(dnodeObj.fqdn, pCreate->fqdn, TSDB_FQDN_LEN);
snprintf(dnodeObj.ep, TSDB_EP_LEN, "%s:%u", dnodeObj.fqdn, dnodeObj.port);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_DNODE, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_GLOBAL, pReq);
if (pTrans == NULL) {
mError("dnode:%s, failed to create since %s", dnodeObj.ep, terrstr());
return -1;
@ -564,7 +560,7 @@ CREATE_DNODE_OVER:
}
static int32_t mndDropDnode(SMnode *pMnode, SRpcMsg *pReq, SDnodeObj *pDnode) {
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_DROP_DNODE, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_GLOBAL, pReq);
if (pTrans == NULL) {
mError("dnode:%d, failed to drop since %s", pDnode->id, terrstr());
return -1;
@ -617,7 +613,7 @@ static int32_t mndProcessDropDnodeReq(SRpcMsg *pReq) {
pMObj = mndAcquireMnode(pMnode, dropReq.dnodeId);
if (pMObj != NULL) {
terrno = TSDB_CODE_MND_MNODE_DEPLOYED;
terrno = TSDB_CODE_MND_MNODE_NOT_EXIST;
goto DROP_DNODE_OVER;
}

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

@ -215,7 +215,7 @@ static int32_t mndCreateFunc(SMnode *pMnode, SRpcMsg *pReq, SCreateFuncReq *pCre
}
memcpy(func.pCode, pCreate->pCode, func.codeSize);
pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_FUNC, pReq);
pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pReq);
if (pTrans == NULL) goto _OVER;
mDebug("trans:%d, used to create func:%s", pTrans->id, pCreate->name);
@ -245,7 +245,7 @@ _OVER:
static int32_t mndDropFunc(SMnode *pMnode, SRpcMsg *pReq, SFuncObj *pFunc) {
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_DROP_FUNC, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pReq);
if (pTrans == NULL) goto _OVER;
mDebug("trans:%d, used to drop user:%s", pTrans->id, pFunc->name);

6
source/dnode/mnode/impl/src/mndMain.c
View File

@ -369,7 +369,7 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
mError("failed to process sync msg:%p type:%s since %s", pMsg, TMSG_INFO(pMsg->msgType), terrstr());
return TAOS_SYNC_PROPOSE_OTHER_ERROR;
}
char logBuf[512] = {0};
char *syncNodeStr = sync2SimpleStr(pMgmt->sync);
snprintf(logBuf, sizeof(logBuf), "==vnodeProcessSyncReq== msgType:%d, syncNode: %s", pMsg->msgType, syncNodeStr);
@ -472,7 +472,7 @@ int32_t mndProcessRpcMsg(SRpcMsg *pMsg) {
} else if (code == 0) {
mTrace("msg:%p, successfully processed and response", pMsg);
} else {
mDebug("msg:%p, failed to process since %s, app:%p type:%s", pMsg, terrstr(), pMsg->info.ahandle,
mError("msg:%p, failed to process since %s, app:%p type:%s", pMsg, terrstr(), pMsg->info.ahandle,
TMSG_INFO(pMsg->msgType));
}
@ -686,4 +686,4 @@ void mndReleaseSyncRef(SMnode *pMnode) {
int32_t ref = atomic_sub_fetch_32(&pMnode->syncRef, 1);
mTrace("mnode sync is released, ref:%d", ref);
taosThreadRwlockUnlock(&pMnode->lock);
}
}

29
source/dnode/mnode/impl/src/mndMnode.c
View File

@ -18,9 +18,9 @@
#include "mndAuth.h"
#include "mndDnode.h"
#include "mndShow.h"
#include "mndSync.h"
#include "mndTrans.h"
#include "mndUser.h"
#include "mndSync.h"
#define MNODE_VER_NUMBER 1
#define MNODE_RESERVE_SIZE 64
@ -92,10 +92,7 @@ static int32_t mndCreateDefaultMnode(SMnode *pMnode) {
mDebug("mnode:%d, will be created when deploying, raw:%p", mnodeObj.id, pRaw);
#if 0
return sdbWrite(pMnode->pSdb, pRaw);
#else
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_CREATE_DNODE, NULL);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, NULL);
if (pTrans == NULL) {
mError("mnode:%d, failed to create since %s", mnodeObj.id, terrstr());
return -1;
@ -117,7 +114,6 @@ static int32_t mndCreateDefaultMnode(SMnode *pMnode) {
mndTransDrop(pTrans);
return 0;
#endif
}
static SSdbRaw *mndMnodeActionEncode(SMnodeObj *pObj) {
@ -363,11 +359,11 @@ static int32_t mndCreateMnode(SMnode *pMnode, SRpcMsg *pReq, SDnodeObj *pDnode,
mnodeObj.createdTime = taosGetTimestampMs();
mnodeObj.updateTime = mnodeObj.createdTime;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_CREATE_MNODE, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, pReq);
if (pTrans == NULL) goto _OVER;
mDebug("trans:%d, used to create mnode:%d", pTrans->id, pCreate->dnodeId);
mndTransSetNoParallel(pTrans);
mndTransSetSerial(pTrans);
if (mndSetCreateMnodeRedoLogs(pMnode, pTrans, &mnodeObj) != 0) goto _OVER;
if (mndSetCreateMnodeCommitLogs(pMnode, pTrans, &mnodeObj) != 0) goto _OVER;
if (mndSetCreateMnodeRedoActions(pMnode, pTrans, pDnode, &mnodeObj) != 0) goto _OVER;
@ -396,6 +392,11 @@ static int32_t mndProcessCreateMnodeReq(SRpcMsg *pReq) {
mDebug("mnode:%d, start to create", createReq.dnodeId);
if (sdbGetSize(pMnode->pSdb, SDB_MNODE) >= 3) {
terrno = TSDB_CODE_MND_TOO_MANY_MNODES;
goto _OVER;
}
pObj = mndAcquireMnode(pMnode, createReq.dnodeId);
if (pObj != NULL) {
terrno = TSDB_CODE_MND_MNODE_ALREADY_EXIST;
@ -535,11 +536,11 @@ static int32_t mndSetDropMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnode
static int32_t mndDropMnode(SMnode *pMnode, SRpcMsg *pReq, SMnodeObj *pObj) {
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_DROP_MNODE, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, pReq);
if (pTrans == NULL) goto _OVER;
mDebug("trans:%d, used to drop mnode:%d", pTrans->id, pObj->id);
mndTransSetNoParallel(pTrans);
mndTransSetSerial(pTrans);
if (mndSetDropMnodeRedoLogs(pMnode, pTrans, pObj) != 0) goto _OVER;
if (mndSetDropMnodeCommitLogs(pMnode, pTrans, pObj) != 0) goto _OVER;
if (mndSetDropMnodeRedoActions(pMnode, pTrans, pObj->pDnode, pObj) != 0) goto _OVER;
@ -632,6 +633,7 @@ static int32_t mndRetrieveMnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
int32_t cols = 0;
SMnodeObj *pObj = NULL;
char *pWrite;
int64_t curMs = taosGetTimestampMs();
while (numOfRows < rows) {
pShow->pIter = sdbFetch(pSdb, SDB_MNODE, pShow->pIter, (void **)&pObj);
@ -647,11 +649,16 @@ static int32_t mndRetrieveMnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, b1, false);
bool online = mndIsDnodeOnline(pMnode, pObj->pDnode, curMs);
const char *roles = NULL;
if (pObj->id == pMnode->selfDnodeId) {
roles = syncStr(TAOS_SYNC_STATE_LEADER);
} else {
roles = syncStr(pObj->state);
if (!online) {
roles = "OFFLINE";
} else {
roles = syncStr(pObj->state);
}
}
char *b2 = taosMemoryCalloc(1, 12 + VARSTR_HEADER_SIZE);
STR_WITH_MAXSIZE_TO_VARSTR(b2, roles, pShow->pMeta->pSchemas[cols].bytes);

2
source/dnode/mnode/impl/src/mndOffset.c
View File

@ -179,7 +179,7 @@ static int32_t mndProcessCommitOffsetReq(SRpcMsg *pMsg) {
tDecodeSMqCMCommitOffsetReq(&decoder, &commitOffsetReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_COMMIT_OFFSET, pMsg);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pMsg);
for (int32_t i = 0; i < commitOffsetReq.num; i++) {
SMqOffset *pOffset = &commitOffsetReq.offsets[i];

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

@ -248,7 +248,7 @@ static int32_t mndCreateQnode(SMnode *pMnode, SRpcMsg *pReq, SDnodeObj *pDnode,
qnodeObj.createdTime = taosGetTimestampMs();
qnodeObj.updateTime = qnodeObj.createdTime;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_QNODE, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pReq);
if (pTrans == NULL) goto _OVER;
mDebug("trans:%d, used to create qnode:%d", pTrans->id, pCreate->dnodeId);
@ -365,7 +365,7 @@ static int32_t mndSetDropQnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, SQn
static int32_t mndDropQnode(SMnode *pMnode, SRpcMsg *pReq, SQnodeObj *pObj) {
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_DROP_QNODE, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_NOTHING, pReq);
if (pTrans == NULL) goto _OVER;
mDebug("trans:%d, used to drop qnode:%d", pTrans->id, pObj->id);

6
source/dnode/mnode/impl/src/mndSma.c
View File

@ -508,12 +508,12 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
streamObj.fixedSinkVgId = smaObj.dstVgId;
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_CREATE_SMA, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB, pReq);
if (pTrans == NULL) goto _OVER;
mDebug("trans:%d, used to create sma:%s", pTrans->id, pCreate->name);
mndTransSetDbInfo(pTrans, pDb);
mndTransSetNoParallel(pTrans);
mndTransSetSerial(pTrans);
if (mndSetCreateSmaRedoLogs(pMnode, pTrans, &smaObj) != 0) goto _OVER;
if (mndSetCreateSmaVgroupRedoLogs(pMnode, pTrans, &streamObj.fixedSinkVg) != 0) goto _OVER;
@ -753,7 +753,7 @@ static int32_t mndDropSma(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SSmaObj *p
pVgroup = mndAcquireVgroup(pMnode, pSma->dstVgId);
if (pVgroup == NULL) goto _OVER;
pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_DROP_SMA, pReq);
pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB, pReq);
if (pTrans == NULL) goto _OVER;
mDebug("trans:%d, used to drop sma:%s", pTrans->id, pSma->name);

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

@ -253,7 +253,7 @@ static int32_t mndCreateSnode(SMnode *pMnode, SRpcMsg *pReq, SDnodeObj *pDnode,
snodeObj.createdTime = taosGetTimestampMs();
snodeObj.updateTime = snodeObj.createdTime;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_SNODE, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pReq);
if (pTrans == NULL) goto _OVER;
mDebug("trans:%d, used to create snode:%d", pTrans->id, pCreate->dnodeId);
@ -372,7 +372,7 @@ static int32_t mndSetDropSnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, SSn
static int32_t mndDropSnode(SMnode *pMnode, SRpcMsg *pReq, SSnodeObj *pObj) {
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_DROP_SNODE, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_NOTHING, pReq);
if (pTrans == NULL) goto _OVER;
mDebug("trans:%d, used to drop snode:%d", pTrans->id, pObj->id);

6
source/dnode/mnode/impl/src/mndStb.c
View File

@ -735,7 +735,7 @@ static int32_t mndCreateStb(SMnode *pMnode, SRpcMsg *pReq, SMCreateStbReq *pCrea
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_STB, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB_INSIDE, pReq);
if (pTrans == NULL) goto _OVER;
mDebug("trans:%d, used to create stb:%s", pTrans->id, pCreate->name);
@ -1257,7 +1257,7 @@ static int32_t mndAlterStb(SMnode *pMnode, SRpcMsg *pReq, const SMAlterStbReq *p
if (code != 0) goto _OVER;
code = -1;
pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_ALTER_STB, pReq);
pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq);
if (pTrans == NULL) goto _OVER;
mDebug("trans:%d, used to alter stb:%s", pTrans->id, pAlter->name);
@ -1403,7 +1403,7 @@ static int32_t mndSetDropStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *
static int32_t mndDropStb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb) {
int32_t code = -1;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_DROP_STB, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB_INSIDE, pReq);
if (pTrans == NULL) goto _OVER;
mDebug("trans:%d, used to drop stb:%s", pTrans->id, pStb->name);

2
source/dnode/mnode/impl/src/mndStream.c
View File

@ -402,7 +402,7 @@ static int32_t mndCreateStream(SMnode *pMnode, SRpcMsg *pReq, SCMCreateStreamReq
tstrncpy(streamObj.targetDb, pDb->name, TSDB_DB_FNAME_LEN);
}
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_STREAM, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pReq);
if (pTrans == NULL) {
mError("stream:%s, failed to create since %s", pCreate->name, terrstr());
return -1;

13
source/dnode/mnode/impl/src/mndSubscribe.c
View File

@ -394,8 +394,8 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
mInfo("rebalance calculation completed, rebalanced vg:");
for (int32_t i = 0; i < taosArrayGetSize(pOutput->rebVgs); i++) {
SMqRebOutputVg *pOutputRebVg = taosArrayGet(pOutput->rebVgs, i);
mInfo("vg: %d moved from consumer %ld to consumer %ld", pOutputRebVg->pVgEp->vgId, pOutputRebVg->oldConsumerId,
pOutputRebVg->newConsumerId);
mInfo("vgId:%d moved from consumer %" PRId64 " to consumer %" PRId64, pOutputRebVg->pVgEp->vgId,
pOutputRebVg->oldConsumerId, pOutputRebVg->newConsumerId);
}
// 9. clear
@ -405,10 +405,9 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
}
static int32_t mndPersistRebResult(SMnode *pMnode, SRpcMsg *pMsg, const SMqRebOutputObj *pOutput) {
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_REBALANCE, pMsg);
if (pTrans == NULL) {
return -1;
}
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pMsg);
if (pTrans == NULL) return -1;
// make txn:
// 1. redo action: action to all vg
const SArray *rebVgs = pOutput->rebVgs;
@ -625,7 +624,7 @@ static int32_t mndProcessDropCgroupReq(SRpcMsg *pReq) {
return -1;
}
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_DROP_CGROUP, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pReq);
if (pTrans == NULL) {
mError("cgroup: %s on topic:%s, failed to drop since %s", dropReq.cgroup, dropReq.topic, terrstr());
mndReleaseSubscribe(pMnode, pSub);

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

@ -383,7 +383,7 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
/*topicObj.withSchema = 1;*/
}
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_TOPIC, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pReq);
if (pTrans == NULL) {
mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
taosMemoryFreeClear(topicObj.ast);
@ -551,7 +551,7 @@ static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {
}
#endif
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_DROP_TOPIC, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pReq);
if (pTrans == NULL) {
mError("topic:%s, failed to drop since %s", pTopic->name, terrstr());
return -1;

265
source/dnode/mnode/impl/src/mndTrans.c
View File

@ -88,12 +88,12 @@ static int32_t mndTransGetActionsSize(SArray *pArray) {
for (int32_t i = 0; i < actionNum; ++i) {
STransAction *pAction = taosArrayGet(pArray, i);
if (pAction->isRaw) {
if (pAction->actionType) {
rawDataLen += (sdbGetRawTotalSize(pAction->pRaw) + sizeof(int32_t));
} else {
rawDataLen += (sizeof(STransAction) + pAction->contLen);
}
rawDataLen += sizeof(pAction->isRaw);
rawDataLen += sizeof(pAction->actionType);
}
return rawDataLen;
@ -117,8 +117,8 @@ static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
SDB_SET_INT32(pRaw, dataPos, pTrans->id, _OVER)
SDB_SET_INT16(pRaw, dataPos, pTrans->stage, _OVER)
SDB_SET_INT16(pRaw, dataPos, pTrans->policy, _OVER)
SDB_SET_INT16(pRaw, dataPos, pTrans->type, _OVER)
SDB_SET_INT16(pRaw, dataPos, pTrans->parallel, _OVER)
SDB_SET_INT16(pRaw, dataPos, pTrans->conflict, _OVER)
SDB_SET_INT16(pRaw, dataPos, pTrans->exec, _OVER)
SDB_SET_INT64(pRaw, dataPos, pTrans->createdTime, _OVER)
SDB_SET_BINARY(pRaw, dataPos, pTrans->dbname, TSDB_DB_FNAME_LEN, _OVER)
SDB_SET_INT32(pRaw, dataPos, pTrans->redoActionPos, _OVER)
@ -135,9 +135,9 @@ static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
SDB_SET_INT32(pRaw, dataPos, pAction->id, _OVER)
SDB_SET_INT32(pRaw, dataPos, pAction->errCode, _OVER)
SDB_SET_INT32(pRaw, dataPos, pAction->acceptableCode, _OVER)
SDB_SET_INT8(pRaw, dataPos, pAction->isRaw, _OVER)
SDB_SET_INT8(pRaw, dataPos, pAction->actionType, _OVER)
SDB_SET_INT8(pRaw, dataPos, pAction->stage, _OVER)
if (pAction->isRaw) {
if (pAction->actionType) {
int32_t len = sdbGetRawTotalSize(pAction->pRaw);
SDB_SET_INT8(pRaw, dataPos, pAction->rawWritten, _OVER)
SDB_SET_INT32(pRaw, dataPos, len, _OVER)
@ -157,9 +157,9 @@ static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
SDB_SET_INT32(pRaw, dataPos, pAction->id, _OVER)
SDB_SET_INT32(pRaw, dataPos, pAction->errCode, _OVER)
SDB_SET_INT32(pRaw, dataPos, pAction->acceptableCode, _OVER)
SDB_SET_INT8(pRaw, dataPos, pAction->isRaw, _OVER)
SDB_SET_INT8(pRaw, dataPos, pAction->actionType, _OVER)
SDB_SET_INT8(pRaw, dataPos, pAction->stage, _OVER)
if (pAction->isRaw) {
if (pAction->actionType) {
int32_t len = sdbGetRawTotalSize(pAction->pRaw);
SDB_SET_INT8(pRaw, dataPos, pAction->rawWritten, _OVER)
SDB_SET_INT32(pRaw, dataPos, len, _OVER)
@ -179,9 +179,9 @@ static SSdbRaw *mndTransActionEncode(STrans *pTrans) {
SDB_SET_INT32(pRaw, dataPos, pAction->id, _OVER)
SDB_SET_INT32(pRaw, dataPos, pAction->errCode, _OVER)
SDB_SET_INT32(pRaw, dataPos, pAction->acceptableCode, _OVER)
SDB_SET_INT8(pRaw, dataPos, pAction->isRaw, _OVER)
SDB_SET_INT8(pRaw, dataPos, pAction->actionType, _OVER)
SDB_SET_INT8(pRaw, dataPos, pAction->stage, _OVER)
if (pAction->isRaw) {
if (pAction->actionType) {
int32_t len = sdbGetRawTotalSize(pAction->pRaw);
SDB_SET_INT8(pRaw, dataPos, pAction->rawWritten, _OVER)
SDB_SET_INT32(pRaw, dataPos, len, _OVER)
@ -250,16 +250,16 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
int16_t stage = 0;
int16_t policy = 0;
int16_t type = 0;
int16_t parallel = 0;
int16_t conflict = 0;
int16_t exec = 0;
SDB_GET_INT16(pRaw, dataPos, &stage, _OVER)
SDB_GET_INT16(pRaw, dataPos, &policy, _OVER)
SDB_GET_INT16(pRaw, dataPos, &type, _OVER)
SDB_GET_INT16(pRaw, dataPos, &parallel, _OVER)
SDB_GET_INT16(pRaw, dataPos, &conflict, _OVER)
SDB_GET_INT16(pRaw, dataPos, &exec, _OVER)
pTrans->stage = stage;
pTrans->policy = policy;
pTrans->type = type;
pTrans->parallel = parallel;
pTrans->conflict = conflict;
pTrans->exec = exec;
SDB_GET_INT64(pRaw, dataPos, &pTrans->createdTime, _OVER)
SDB_GET_BINARY(pRaw, dataPos, pTrans->dbname, TSDB_DB_FNAME_LEN, _OVER)
SDB_GET_INT32(pRaw, dataPos, &pTrans->redoActionPos, _OVER)
@ -279,9 +279,9 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
SDB_GET_INT32(pRaw, dataPos, &action.id, _OVER)
SDB_GET_INT32(pRaw, dataPos, &action.errCode, _OVER)
SDB_GET_INT32(pRaw, dataPos, &action.acceptableCode, _OVER)
SDB_GET_INT8(pRaw, dataPos, &action.isRaw, _OVER)
SDB_GET_INT8(pRaw, dataPos, &action.actionType, _OVER)
SDB_GET_INT8(pRaw, dataPos, &action.stage, _OVER)
if (action.isRaw) {
if (action.actionType) {
SDB_GET_INT8(pRaw, dataPos, &action.rawWritten, _OVER)
SDB_GET_INT32(pRaw, dataPos, &dataLen, _OVER)
action.pRaw = taosMemoryMalloc(dataLen);
@ -308,9 +308,9 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
SDB_GET_INT32(pRaw, dataPos, &action.id, _OVER)
SDB_GET_INT32(pRaw, dataPos, &action.errCode, _OVER)
SDB_GET_INT32(pRaw, dataPos, &action.acceptableCode, _OVER)
SDB_GET_INT8(pRaw, dataPos, &action.isRaw, _OVER)
SDB_GET_INT8(pRaw, dataPos, &action.actionType, _OVER)
SDB_GET_INT8(pRaw, dataPos, &action.stage, _OVER)
if (action.isRaw) {
if (action.actionType) {
SDB_GET_INT8(pRaw, dataPos, &action.rawWritten, _OVER)
SDB_GET_INT32(pRaw, dataPos, &dataLen, _OVER)
action.pRaw = taosMemoryMalloc(dataLen);
@ -337,9 +337,9 @@ static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
SDB_GET_INT32(pRaw, dataPos, &action.id, _OVER)
SDB_GET_INT32(pRaw, dataPos, &action.errCode, _OVER)
SDB_GET_INT32(pRaw, dataPos, &action.acceptableCode, _OVER)
SDB_GET_INT8(pRaw, dataPos, &action.isRaw, _OVER)
SDB_GET_INT8(pRaw, dataPos, &action.actionType, _OVER)
SDB_GET_INT8(pRaw, dataPos, &action.stage, _OVER)
if (action.isRaw) {
if (action.actionType) {
SDB_GET_INT8(pRaw, dataPos, &action.rawWritten, _OVER)
SDB_GET_INT32(pRaw, dataPos, &dataLen, _OVER)
action.pRaw = taosMemoryMalloc(dataLen);
@ -408,81 +408,6 @@ static const char *mndTransStr(ETrnStage stage) {
}
}
static const char *mndTransType(ETrnType type) {
switch (type) {
case TRN_TYPE_CREATE_USER:
return "create-user";
case TRN_TYPE_ALTER_USER:
return "alter-user";
case TRN_TYPE_DROP_USER:
return "drop-user";
case TRN_TYPE_CREATE_FUNC:
return "create-func";
case TRN_TYPE_DROP_FUNC:
return "drop-func";
case TRN_TYPE_CREATE_SNODE:
return "create-snode";
case TRN_TYPE_DROP_SNODE:
return "drop-snode";
case TRN_TYPE_CREATE_QNODE:
return "create-qnode";
case TRN_TYPE_DROP_QNODE:
return "drop-qnode";
case TRN_TYPE_CREATE_BNODE:
return "create-bnode";
case TRN_TYPE_DROP_BNODE:
return "drop-bnode";
case TRN_TYPE_CREATE_MNODE:
return "create-mnode";
case TRN_TYPE_DROP_MNODE:
return "drop-mnode";
case TRN_TYPE_CREATE_TOPIC:
return "create-topic";
case TRN_TYPE_DROP_TOPIC:
return "drop-topic";
case TRN_TYPE_SUBSCRIBE:
return "subscribe";
case TRN_TYPE_REBALANCE:
return "rebalance";
case TRN_TYPE_COMMIT_OFFSET:
return "commit-offset";
case TRN_TYPE_CREATE_STREAM:
return "create-stream";
case TRN_TYPE_DROP_STREAM:
return "drop-stream";
case TRN_TYPE_CONSUMER_LOST:
return "consumer-lost";
case TRN_TYPE_CONSUMER_RECOVER:
return "consumer-recover";
case TRN_TYPE_CREATE_DNODE:
return "create-qnode";
case TRN_TYPE_DROP_DNODE:
return "drop-qnode";
case TRN_TYPE_CREATE_DB:
return "create-db";
case TRN_TYPE_ALTER_DB:
return "alter-db";
case TRN_TYPE_DROP_DB:
return "drop-db";
case TRN_TYPE_SPLIT_VGROUP:
return "split-vgroup";
case TRN_TYPE_MERGE_VGROUP:
return "merge-vgroup";
case TRN_TYPE_CREATE_STB:
return "create-stb";
case TRN_TYPE_ALTER_STB:
return "alter-stb";
case TRN_TYPE_DROP_STB:
return "drop-stb";
case TRN_TYPE_CREATE_SMA:
return "create-sma";
case TRN_TYPE_DROP_SMA:
return "drop-sma";
default:
return "invalid";
}
}
static void mndTransTestStartFunc(SMnode *pMnode, void *param, int32_t paramLen) {
mInfo("test trans start, param:%s, len:%d", (char *)param, paramLen);
}
@ -594,7 +519,7 @@ void mndReleaseTrans(SMnode *pMnode, STrans *pTrans) {
sdbRelease(pSdb, pTrans);
}
STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, ETrnType type, const SRpcMsg *pReq) {
STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, ETrnConflct conflict, const SRpcMsg *pReq) {
STrans *pTrans = taosMemoryCalloc(1, sizeof(STrans));
if (pTrans == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
@ -605,8 +530,8 @@ STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, ETrnType type, const S
pTrans->id = sdbGetMaxId(pMnode->pSdb, SDB_TRANS);
pTrans->stage = TRN_STAGE_PREPARE;
pTrans->policy = policy;
pTrans->type = type;
pTrans->parallel = TRN_EXEC_PARALLEL;
pTrans->conflict = conflict;
pTrans->exec = TRN_EXEC_PRARLLEL;
pTrans->createdTime = taosGetTimestampMs();
pTrans->redoActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
pTrans->undoActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
@ -627,7 +552,7 @@ static void mndTransDropActions(SArray *pArray) {
int32_t size = taosArrayGetSize(pArray);
for (int32_t i = 0; i < size; ++i) {
STransAction *pAction = taosArrayGet(pArray, i);
if (pAction->isRaw) {
if (pAction->actionType) {
taosMemoryFreeClear(pAction->pRaw);
} else {
taosMemoryFreeClear(pAction->pCont);
@ -658,17 +583,17 @@ static int32_t mndTransAppendAction(SArray *pArray, STransAction *pAction) {
}
int32_t mndTransAppendRedolog(STrans *pTrans, SSdbRaw *pRaw) {
STransAction action = {.stage = TRN_STAGE_REDO_ACTION, .isRaw = true, .pRaw = pRaw};
STransAction action = {.stage = TRN_STAGE_REDO_ACTION, .actionType = true, .pRaw = pRaw};
return mndTransAppendAction(pTrans->redoActions, &action);
}
int32_t mndTransAppendUndolog(STrans *pTrans, SSdbRaw *pRaw) {
STransAction action = {.stage = TRN_STAGE_UNDO_ACTION, .isRaw = true, .pRaw = pRaw};
STransAction action = {.stage = TRN_STAGE_UNDO_ACTION, .actionType = true, .pRaw = pRaw};
return mndTransAppendAction(pTrans->undoActions, &action);
}
int32_t mndTransAppendCommitlog(STrans *pTrans, SSdbRaw *pRaw) {
STransAction action = {.stage = TRN_STAGE_COMMIT_ACTION, .isRaw = true, .pRaw = pRaw};
STransAction action = {.stage = TRN_STAGE_COMMIT_ACTION, .actionType = true, .pRaw = pRaw};
return mndTransAppendAction(pTrans->commitActions, &action);
}
@ -698,7 +623,7 @@ void mndTransSetDbInfo(STrans *pTrans, SDbObj *pDb) {
memcpy(pTrans->dbname, pDb->name, TSDB_DB_FNAME_LEN);
}
void mndTransSetNoParallel(STrans *pTrans) { pTrans->parallel = TRN_EXEC_NO_PARALLEL; }
void mndTransSetSerial(STrans *pTrans) { pTrans->exec = TRN_EXEC_SERIAL; }
static int32_t mndTransSync(SMnode *pMnode, STrans *pTrans) {
SSdbRaw *pRaw = mndTransActionEncode(pTrans);
@ -721,76 +646,43 @@ static int32_t mndTransSync(SMnode *pMnode, STrans *pTrans) {
return 0;
}
static bool mndIsBasicTrans(STrans *pTrans) {
return pTrans->type > TRN_TYPE_BASIC_SCOPE && pTrans->type < TRN_TYPE_BASIC_SCOPE_END;
}
static bool mndIsGlobalTrans(STrans *pTrans) {
return pTrans->type > TRN_TYPE_GLOBAL_SCOPE && pTrans->type < TRN_TYPE_GLOBAL_SCOPE_END;
}
static bool mndIsDbTrans(STrans *pTrans) {
return pTrans->type > TRN_TYPE_DB_SCOPE && pTrans->type < TRN_TYPE_DB_SCOPE_END;
}
static bool mndIsStbTrans(STrans *pTrans) {
return pTrans->type > TRN_TYPE_STB_SCOPE && pTrans->type < TRN_TYPE_STB_SCOPE_END;
}
static bool mndCheckTransConflict(SMnode *pMnode, STrans *pNewTrans) {
static bool mndCheckTransConflict(SMnode *pMnode, STrans *pNew) {
STrans *pTrans = NULL;
void *pIter = NULL;
bool conflict = false;
if (mndIsBasicTrans(pNewTrans)) return conflict;
if (pNew->conflict == TRN_CONFLICT_NOTHING) return conflict;
while (1) {
pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
if (pIter == NULL) break;
if (mndIsGlobalTrans(pNewTrans)) {
if (mndIsDbTrans(pTrans) || mndIsStbTrans(pTrans)) {
mError("trans:%d, can't execute since trans:%d in progress db:%s", pNewTrans->id, pTrans->id, pTrans->dbname);
conflict = true;
} else {
}
if (pNew->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
if (pNew->conflict == TRN_CONFLICT_DB) {
if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
if (pTrans->conflict == TRN_CONFLICT_DB && strcmp(pNew->dbname, pTrans->dbname) == 0) conflict = true;
if (pTrans->conflict == TRN_CONFLICT_DB_INSIDE && strcmp(pNew->dbname, pTrans->dbname) == 0) conflict = true;
}
else if (mndIsDbTrans(pNewTrans)) {
if (mndIsGlobalTrans(pTrans)) {
mError("trans:%d, can't execute since trans:%d in progress", pNewTrans->id, pTrans->id);
conflict = true;
} else if (mndIsDbTrans(pTrans) || mndIsStbTrans(pTrans)) {
if (strcmp(pNewTrans->dbname, pTrans->dbname) == 0) {
mError("trans:%d, can't execute since trans:%d in progress db:%s", pNewTrans->id, pTrans->id, pTrans->dbname);
conflict = true;
}
} else {
}
if (pNew->conflict == TRN_CONFLICT_DB_INSIDE) {
if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
if (pTrans->conflict == TRN_CONFLICT_DB && strcmp(pNew->dbname, pTrans->dbname) == 0) conflict = true;
}
else if (mndIsStbTrans(pNewTrans)) {
if (mndIsGlobalTrans(pTrans)) {
mError("trans:%d, can't execute since trans:%d in progress", pNewTrans->id, pTrans->id);
conflict = true;
} else if (mndIsDbTrans(pTrans)) {
if (strcmp(pNewTrans->dbname, pTrans->dbname) == 0) {
mError("trans:%d, can't execute since trans:%d in progress db:%s", pNewTrans->id, pTrans->id, pTrans->dbname);
conflict = true;
}
} else {
}
}
mError("trans:%d, can't execute since conflict with trans:%d, db:%s", pNew->id, pTrans->id, pTrans->dbname);
sdbRelease(pMnode->pSdb, pTrans);
}
sdbCancelFetch(pMnode->pSdb, pIter);
sdbRelease(pMnode->pSdb, pTrans);
return conflict;
}
int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans) {
if (pTrans->conflict == TRN_CONFLICT_DB || pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
if (strlen(pTrans->dbname) == 0) {
terrno = TSDB_CODE_MND_TRANS_CONFLICT;
mError("trans:%d, failed to prepare conflict db not set", pTrans->id);
return -1;
}
}
if (mndCheckTransConflict(pMnode, pTrans)) {
terrno = TSDB_CODE_MND_TRANS_CONFLICT;
mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
@ -921,9 +813,6 @@ void mndTransProcessRsp(SRpcMsg *pRsp) {
if (pAction != NULL) {
pAction->msgReceived = 1;
pAction->errCode = pRsp->code;
if (pAction->errCode != 0) {
tstrncpy(pTrans->lastError, tstrerror(pAction->errCode), TSDB_TRANS_ERROR_LEN);
}
}
mDebug("trans:%d, %s:%d response is received, code:0x%x, accept:0x%x", transId, mndTransStr(pAction->stage), action,
@ -1004,7 +893,7 @@ static int32_t mndTransSendSingleMsg(SMnode *pMnode, STrans *pTrans, STransActio
}
static int32_t mndTransExecSingleAction(SMnode *pMnode, STrans *pTrans, STransAction *pAction) {
if (pAction->isRaw) {
if (pAction->actionType) {
return mndTransWriteSingleLog(pMnode, pTrans, pAction);
} else {
return mndTransSendSingleMsg(pMnode, pTrans, pAction);
@ -1032,24 +921,36 @@ static int32_t mndTransExecuteActions(SMnode *pMnode, STrans *pTrans, SArray *pA
return -1;
}
int32_t numOfExecuted = 0;
int32_t errCode = 0;
int32_t numOfExecuted = 0;
int32_t errCode = 0;
STransAction *pErrAction = NULL;
for (int32_t action = 0; action < numOfActions; ++action) {
STransAction *pAction = taosArrayGet(pArray, action);
if (pAction->msgReceived || pAction->rawWritten) {
numOfExecuted++;
if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
errCode = pAction->errCode;
pErrAction = pAction;
}
}
}
if (numOfExecuted == numOfActions) {
if (errCode == 0) {
pTrans->lastErrorAction = 0;
pTrans->lastErrorNo = 0;
pTrans->lastErrorMsgType = 0;
memset(&pTrans->lastErrorEpset, 0, sizeof(pTrans->lastErrorEpset));
mDebug("trans:%d, all %d actions execute successfully", pTrans->id, numOfActions);
return 0;
} else {
mError("trans:%d, all %d actions executed, code:0x%x", pTrans->id, numOfActions, errCode & 0XFFFF);
if (pErrAction != NULL) {
pTrans->lastErrorMsgType = pErrAction->msgType;
pTrans->lastErrorAction = pErrAction->id;
pTrans->lastErrorNo = pErrAction->errCode;
pTrans->lastErrorEpset = pErrAction->epSet;
}
mndTransResetActions(pMnode, pTrans, pArray);
terrno = errCode;
return errCode;
@ -1084,7 +985,7 @@ static int32_t mndTransExecuteCommitActions(SMnode *pMnode, STrans *pTrans) {
return code;
}
static int32_t mndTransExecuteRedoActionsNoParallel(SMnode *pMnode, STrans *pTrans) {
static int32_t mndTransExecuteRedoActionsSerial(SMnode *pMnode, STrans *pTrans) {
int32_t code = 0;
int32_t numOfActions = taosArrayGetSize(pTrans->redoActions);
if (numOfActions == 0) return code;
@ -1111,6 +1012,18 @@ static int32_t mndTransExecuteRedoActionsNoParallel(SMnode *pMnode, STrans *pTra
}
}
if (code == 0) {
pTrans->lastErrorAction = 0;
pTrans->lastErrorNo = 0;
pTrans->lastErrorMsgType = 0;
memset(&pTrans->lastErrorEpset, 0, sizeof(pTrans->lastErrorEpset));
} else {
pTrans->lastErrorMsgType = pAction->msgType;
pTrans->lastErrorAction = action;
pTrans->lastErrorNo = pAction->errCode;
pTrans->lastErrorEpset = pAction->epSet;
}
if (code == 0) {
pTrans->redoActionPos++;
mDebug("trans:%d, %s:%d is executed and need sync to other mnodes", pTrans->id, mndTransStr(pAction->stage),
@ -1144,8 +1057,8 @@ static bool mndTransPerformRedoActionStage(SMnode *pMnode, STrans *pTrans) {
bool continueExec = true;
int32_t code = 0;
if (pTrans->parallel == TRN_EXEC_NO_PARALLEL) {
code = mndTransExecuteRedoActionsNoParallel(pMnode, pTrans);
if (pTrans->exec == TRN_EXEC_SERIAL) {
code = mndTransExecuteRedoActionsSerial(pMnode, pTrans);
} else {
code = mndTransExecuteRedoActions(pMnode, pTrans);
}
@ -1455,11 +1368,6 @@ static int32_t mndRetrieveTrans(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)dbname, false);
char type[TSDB_TRANS_TYPE_LEN + VARSTR_HEADER_SIZE] = {0};
STR_WITH_MAXSIZE_TO_VARSTR(type, mndTransType(pTrans->type), pShow->pMeta->pSchemas[cols].bytes);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)type, false);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)&pTrans->failedTimes, false);
@ -1467,7 +1375,20 @@ static int32_t mndRetrieveTrans(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
colDataAppend(pColInfo, numOfRows, (const char *)&pTrans->lastExecTime, false);
char lastError[TSDB_TRANS_ERROR_LEN + VARSTR_HEADER_SIZE] = {0};
STR_WITH_MAXSIZE_TO_VARSTR(lastError, pTrans->lastError, pShow->pMeta->pSchemas[cols].bytes);
char detail[TSDB_TRANS_ERROR_LEN] = {0};
if (pTrans->lastErrorNo != 0) {
int32_t len = snprintf(detail, sizeof(detail), "action:%d errno:0x%x(%s) ", pTrans->lastErrorAction,
pTrans->lastErrorNo & 0xFFFF, tstrerror(pTrans->lastErrorNo));
SEpSet epset = pTrans->lastErrorEpset;
if (epset.numOfEps > 0) {
len += snprintf(detail + len, sizeof(detail) - len, "msgType:%s numOfEps:%d inUse:%d ",
TMSG_INFO(pTrans->lastErrorMsgType), epset.numOfEps, epset.inUse);
}
for (int32_t i = 0; i < pTrans->lastErrorEpset.numOfEps; ++i) {
len += snprintf(detail + len, sizeof(detail) - len, "ep:%d-%s:%u ", i, epset.eps[i].fqdn, epset.eps[i].port);
}
}
STR_WITH_MAXSIZE_TO_VARSTR(lastError, detail, pShow->pMeta->pSchemas[cols].bytes);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)lastError, false);

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

@ -79,10 +79,7 @@ static int32_t mndCreateDefaultUser(SMnode *pMnode, char *acct, char *user, char
mDebug("user:%s, will be created when deploying, raw:%p", userObj.user, pRaw);
#if 0
return sdbWrite(pMnode->pSdb, pRaw);
#else
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_CREATE_USER, NULL);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_NOTHING, NULL);
if (pTrans == NULL) {
mError("user:%s, failed to create since %s", userObj.user, terrstr());
return -1;
@ -104,7 +101,6 @@ static int32_t mndCreateDefaultUser(SMnode *pMnode, char *acct, char *user, char
mndTransDrop(pTrans);
return 0;
#endif
}
static int32_t mndCreateDefaultUsers(SMnode *pMnode) {
@ -291,7 +287,7 @@ static int32_t mndCreateUser(SMnode *pMnode, char *acct, SCreateUserReq *pCreate
userObj.updateTime = userObj.createdTime;
userObj.superUser = pCreate->superUser;
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_USER, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pReq);
if (pTrans == NULL) {
mError("user:%s, failed to create since %s", pCreate->user, terrstr());
return -1;
@ -371,7 +367,7 @@ _OVER:
}
static int32_t mndAlterUser(SMnode *pMnode, SUserObj *pOld, SUserObj *pNew, SRpcMsg *pReq) {
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_ALTER_USER, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pReq);
if (pTrans == NULL) {
mError("user:%s, failed to alter since %s", pOld->user, terrstr());
return -1;
@ -578,7 +574,7 @@ _OVER:
}
static int32_t mndDropUser(SMnode *pMnode, SRpcMsg *pReq, SUserObj *pUser) {
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_DROP_USER, pReq);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pReq);
if (pTrans == NULL) {
mError("user:%s, failed to drop since %s", pUser->user, terrstr());
return -1;

14
source/dnode/mnode/impl/test/trans/trans2.cpp
View File

@ -11,6 +11,8 @@
#include <gtest/gtest.h>
#if 0
#include "mndTrans.h"
#include "mndUser.h"
#include "tcache.h"
@ -103,7 +105,7 @@ class MndTestTrans2 : public ::testing::Test {
void SetUp() override {}
void TearDown() override {}
int32_t CreateUserLog(const char *acct, const char *user, ETrnType type, SDbObj *pDb) {
int32_t CreateUserLog(const char *acct, const char *user, ETrnConflct conflict, SDbObj *pDb) {
SUserObj userObj = {0};
taosEncryptPass_c((uint8_t *)"taosdata", strlen("taosdata"), userObj.pass);
tstrncpy(userObj.user, user, TSDB_USER_LEN);
@ -113,7 +115,7 @@ class MndTestTrans2 : public ::testing::Test {
userObj.superUser = 1;
SRpcMsg rpcMsg = {0};
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, type, &rpcMsg);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, conflict, &rpcMsg);
SSdbRaw *pRedoRaw = mndUserActionEncode(&userObj);
mndTransAppendRedolog(pTrans, pRedoRaw);
sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY);
@ -135,7 +137,7 @@ class MndTestTrans2 : public ::testing::Test {
return code;
}
int32_t CreateUserAction(const char *acct, const char *user, bool hasUndoAction, ETrnPolicy policy, ETrnType type,
int32_t CreateUserAction(const char *acct, const char *user, bool hasUndoAction, ETrnPolicy policy, ETrnConflct conflict,
SDbObj *pDb) {
SUserObj userObj = {0};
taosEncryptPass_c((uint8_t *)"taosdata", strlen("taosdata"), userObj.pass);
@ -146,7 +148,7 @@ class MndTestTrans2 : public ::testing::Test {
userObj.superUser = 1;
SRpcMsg rpcMsg = {0};
STrans *pTrans = mndTransCreate(pMnode, policy, type, &rpcMsg);
STrans *pTrans = mndTransCreate(pMnode, policy, conflict, &rpcMsg);
SSdbRaw *pRedoRaw = mndUserActionEncode(&userObj);
mndTransAppendRedolog(pTrans, pRedoRaw);
sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY);
@ -218,7 +220,7 @@ class MndTestTrans2 : public ::testing::Test {
userObj.superUser = 1;
SRpcMsg rpcMsg = {0};
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_USER, &rpcMsg);
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, &rpcMsg);
SSdbRaw *pRedoRaw = mndUserActionEncode(&userObj);
mndTransAppendRedolog(pTrans, pRedoRaw);
sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY);
@ -528,3 +530,5 @@ TEST_F(MndTestTrans2, 04_Conflict) {
mndReleaseUser(pMnode, pUser);
}
}
#endif

10
source/dnode/mnode/impl/test/user/CMakeLists.txt
View File

@ -5,7 +5,9 @@ target_link_libraries(
PUBLIC sut
)
add_test(
NAME userTest
COMMAND userTest
)
if(NOT TD_WINDOWS)
add_test(
NAME userTest
COMMAND userTest
)
endif(NOT TD_WINDOWS)

7
source/dnode/vnode/CMakeLists.txt
View File

@ -52,10 +52,11 @@ target_sources(
# tq
"src/tq/tq.c"
"src/tq/tqExec.c"
"src/tq/tqCommit.c"
"src/tq/tqOffset.c"
"src/tq/tqPush.c"
"src/tq/tqMeta.c"
"src/tq/tqRead.c"
"src/tq/tqOffset.c"
#"src/tq/tqPush.c"
#"src/tq/tqCommit.c"
)
target_include_directories(
vnode

19
source/dnode/vnode/src/inc/tq.h
View File

@ -66,12 +66,12 @@ struct STqReadHandle {
// tqPush
typedef struct {
int64_t consumerId;
int32_t epoch;
int32_t skipLogNum;
int64_t reqOffset;
SRWLatch lock;
SRpcMsg* handle;
int64_t consumerId;
int32_t epoch;
int32_t skipLogNum;
int64_t reqOffset;
SRpcHandleInfo info;
SRWLatch lock;
} STqPushHandle;
#if 0
@ -168,6 +168,13 @@ int64_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalHead*
int32_t tqDataExec(STQ* pTq, STqExecHandle* pExec, SSubmitReq* pReq, SMqDataBlkRsp* pRsp, int32_t workerId);
// tqMeta
int32_t tqMetaOpen(STQ* pTq);
int32_t tqMetaClose(STQ* pTq);
int32_t tqMetaSaveHandle(STQ* pTq, const char* key, const STqHandle* pHandle);
int32_t tqMetaDeleteHandle(STQ* pTq, const char* key);
// tqOffset
STqOffsetStore* STqOffsetOpen(STqOffsetCfg*);
void STqOffsetClose(STqOffsetStore*);

209
source/dnode/vnode/src/tq/tq.c
View File

@ -14,7 +14,6 @@
*/
#include "tq.h"
#include "tdbInt.h"
int32_t tqInit() {
int8_t old;
@ -47,51 +46,6 @@ void tqCleanUp() {
}
}
int tqExecKeyCompare(const void* pKey1, int32_t kLen1, const void* pKey2, int32_t kLen2) {
return strcmp(pKey1, pKey2);
}
int32_t tqStoreHandle(STQ* pTq, const char* key, const STqHandle* pHandle) {
int32_t code;
int32_t vlen;
tEncodeSize(tEncodeSTqHandle, pHandle, vlen, code);
ASSERT(code == 0);
void* buf = taosMemoryCalloc(1, vlen);
if (buf == NULL) {
ASSERT(0);
}
SEncoder encoder;
tEncoderInit(&encoder, buf, vlen);
if (tEncodeSTqHandle(&encoder, pHandle) < 0) {
ASSERT(0);
}
TXN txn;
if (tdbTxnOpen(&txn, 0, tdbDefaultMalloc, tdbDefaultFree, NULL, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED) < 0) {
ASSERT(0);
}
if (tdbBegin(pTq->pMetaStore, &txn) < 0) {
ASSERT(0);
}
if (tdbTbUpsert(pTq->pExecStore, key, (int)strlen(key), buf, vlen, &txn) < 0) {
ASSERT(0);
}
if (tdbCommit(pTq->pMetaStore, &txn) < 0) {
ASSERT(0);
}
tEncoderClear(&encoder);
taosMemoryFree(buf);
return 0;
}
STQ* tqOpen(const char* path, SVnode* pVnode, SWal* pWal) {
STQ* pTq = taosMemoryMalloc(sizeof(STQ));
if (pTq == NULL) {
@ -108,60 +62,7 @@ STQ* tqOpen(const char* path, SVnode* pVnode, SWal* pWal) {
pTq->pushMgr = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
if (tdbOpen(path, 16 * 1024, 1, &pTq->pMetaStore) < 0) {
ASSERT(0);
}
if (tdbTbOpen("handles", -1, -1, tqExecKeyCompare, pTq->pMetaStore, &pTq->pExecStore) < 0) {
ASSERT(0);
}
TXN txn;
if (tdbTxnOpen(&txn, 0, tdbDefaultMalloc, tdbDefaultFree, NULL, 0) < 0) {
ASSERT(0);
}
TBC* pCur;
if (tdbTbcOpen(pTq->pExecStore, &pCur, &txn) < 0) {
ASSERT(0);
}
void* pKey;
int kLen;
void* pVal;
int vLen;
tdbTbcMoveToFirst(pCur);
SDecoder decoder;
while (tdbTbcNext(pCur, &pKey, &kLen, &pVal, &vLen) == 0) {
STqHandle handle;
tDecoderInit(&decoder, (uint8_t*)pVal, vLen);
tDecodeSTqHandle(&decoder, &handle);
handle.pWalReader = walOpenReadHandle(pTq->pVnode->pWal);
for (int32_t i = 0; i < 5; i++) {
handle.execHandle.pExecReader[i] = tqInitSubmitMsgScanner(pTq->pVnode->pMeta);
}
if (handle.execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
for (int32_t i = 0; i < 5; i++) {
SReadHandle reader = {
.reader = handle.execHandle.pExecReader[i],
.meta = pTq->pVnode->pMeta,
.pMsgCb = &pTq->pVnode->msgCb,
};
handle.execHandle.exec.execCol.task[i] =
qCreateStreamExecTaskInfo(handle.execHandle.exec.execCol.qmsg, &reader);
ASSERT(handle.execHandle.exec.execCol.task[i]);
}
} else {
handle.execHandle.exec.execDb.pFilterOutTbUid =
taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
}
taosHashPut(pTq->handles, pKey, kLen, &handle, sizeof(STqHandle));
}
if (tdbTxnClose(&txn) < 0) {
if (tqMetaOpen(pTq) < 0) {
ASSERT(0);
}
@ -174,46 +75,12 @@ void tqClose(STQ* pTq) {
taosHashCleanup(pTq->handles);
taosHashCleanup(pTq->pStreamTasks);
taosHashCleanup(pTq->pushMgr);
tdbClose(pTq->pMetaStore);
tqMetaClose(pTq);
taosMemoryFree(pTq);
}
// TODO
}
#if 0
int32_t tEncodeSTqExec(SEncoder* pEncoder, const STqExec* pExec) {
if (tStartEncode(pEncoder) < 0) return -1;
if (tEncodeCStr(pEncoder, pExec->subKey) < 0) return -1;
if (tEncodeI64(pEncoder, pExec->consumerId) < 0) return -1;
if (tEncodeI32(pEncoder, pExec->epoch) < 0) return -1;
if (tEncodeI8(pEncoder, pExec->subType) < 0) return -1;
/*if (tEncodeI8(pEncoder, pExec->withTbName) < 0) return -1;*/
/*if (tEncodeI8(pEncoder, pExec->withSchema) < 0) return -1;*/
/*if (tEncodeI8(pEncoder, pExec->withTag) < 0) return -1;*/
if (pExec->subType == TOPIC_SUB_TYPE__COLUMN) {
if (tEncodeCStr(pEncoder, pExec->qmsg) < 0) return -1;
}
tEndEncode(pEncoder);
return pEncoder->pos;
}
int32_t tDecodeSTqExec(SDecoder* pDecoder, STqExec* pExec) {
if (tStartDecode(pDecoder) < 0) return -1;
if (tDecodeCStrTo(pDecoder, pExec->subKey) < 0) return -1;
if (tDecodeI64(pDecoder, &pExec->consumerId) < 0) return -1;
if (tDecodeI32(pDecoder, &pExec->epoch) < 0) return -1;
if (tDecodeI8(pDecoder, &pExec->subType) < 0) return -1;
/*if (tDecodeI8(pDecoder, &pExec->withTbName) < 0) return -1;*/
/*if (tDecodeI8(pDecoder, &pExec->withSchema) < 0) return -1;*/
/*if (tDecodeI8(pDecoder, &pExec->withTag) < 0) return -1;*/
if (pExec->subType == TOPIC_SUB_TYPE__COLUMN) {
if (tDecodeCStrAlloc(pDecoder, &pExec->qmsg) < 0) return -1;
}
tEndDecode(pDecoder);
return 0;
}
#endif
int32_t tEncodeSTqHandle(SEncoder* pEncoder, const STqHandle* pHandle) {
if (tStartEncode(pEncoder) < 0) return -1;
if (tEncodeCStr(pEncoder, pHandle->subKey) < 0) return -1;
@ -290,9 +157,6 @@ int32_t tqPushMsgNew(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_
taosWLockLatch(&pHandle->pushHandle.lock);
SRpcMsg* pMsg = atomic_load_ptr(&pHandle->pushHandle.handle);
ASSERT(pMsg);
SMqDataBlkRsp rsp = {0};
rsp.reqOffset = pHandle->pushHandle.reqOffset;
rsp.blockData = taosArrayInit(0, sizeof(void*));
@ -318,7 +182,7 @@ int32_t tqPushMsgNew(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_
int32_t tlen = sizeof(SMqRspHead) + tEncodeSMqDataBlkRsp(NULL, &rsp);
void* buf = rpcMallocCont(tlen);
if (buf == NULL) {
pMsg->code = -1;
// todo free
return -1;
}
@ -329,10 +193,15 @@ int32_t tqPushMsgNew(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_
void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
tEncodeSMqDataBlkRsp(&abuf, &rsp);
SRpcMsg resp = {.info = handleInfo, .pCont = buf, .contLen = tlen, .code = 0};
SRpcMsg resp = {
.info = pHandle->pushHandle.info,
.pCont = buf,
.contLen = tlen,
.code = 0,
};
tmsgSendRsp(&resp);
atomic_store_ptr(&pHandle->pushHandle.handle, NULL);
memset(&pHandle->pushHandle.info, 0, sizeof(SRpcHandleInfo));
taosWUnLockLatch(&pHandle->pushHandle.lock);
tqDebug("vg %d offset %ld from consumer %ld (epoch %d) send rsp, block num: %d, reqOffset: %ld, rspOffset: %ld",
@ -374,7 +243,7 @@ int tqCommit(STQ* pTq) {
int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
SMqPollReq* pReq = pMsg->pCont;
int64_t consumerId = pReq->consumerId;
int64_t waitTime = pReq->waitTime;
int64_t waitTime = pReq->timeout;
int32_t reqEpoch = pReq->epoch;
int64_t fetchOffset;
@ -410,24 +279,22 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
rsp.blockData = taosArrayInit(0, sizeof(void*));
rsp.blockDataLen = taosArrayInit(0, sizeof(int32_t));
rsp.blockSchema = taosArrayInit(0, sizeof(void*));
rsp.blockTbName = taosArrayInit(0, sizeof(void*));
rsp.withTbName = pReq->withTbName;
if (rsp.withTbName) {
rsp.blockTbName = taosArrayInit(0, sizeof(void*));
}
if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
rsp.withSchema = false;
rsp.withTag = false;
} else {
rsp.withSchema = true;
rsp.blockSchema = taosArrayInit(0, sizeof(void*));
rsp.withTag = false;
}
/*int8_t withTbName = pExec->withTbName;*/
/*if (pReq->withTbName != -1) {*/
/*withTbName = pReq->withTbName;*/
/*}*/
/*rsp.withTbName = withTbName;*/
while (1) {
consumerEpoch = atomic_load_32(&pHandle->epoch);
if (consumerEpoch > reqEpoch) {
@ -443,15 +310,6 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
SWalReadHead* pHead = &pHeadWithCkSum->head;
#if 0
SWalReadHead* pHead;
if (walReadWithHandle_s(pExec->pWalReader, fetchOffset, &pHead) < 0) {
// TODO: no more log, set timer to wait blocking time
// if data inserted during waiting, launch query and
// response to user
tqDebug("tmq poll: consumer %ld (epoch %d) vg %d offset %ld, no more log to return", consumerId, pReq->epoch,
TD_VID(pTq->pVnode), fetchOffset);
#if 0
// add to pushMgr
taosWLockLatch(&pExec->pushHandle.lock);
@ -473,10 +331,6 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
return 0;
#endif
break;
}
#endif
tqDebug("tmq poll: consumer %ld (epoch %d) iter log, vg %d offset %ld msgType %d", consumerId, pReq->epoch,
TD_VID(pTq->pVnode), fetchOffset, pHead->msgType);
@ -533,8 +387,14 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
// TODO wrap in destroy func
taosArrayDestroy(rsp.blockData);
taosArrayDestroy(rsp.blockDataLen);
taosArrayDestroyP(rsp.blockSchema, (FDelete)tDeleteSSchemaWrapper);
taosArrayDestroyP(rsp.blockTbName, (FDelete)taosMemoryFree);
if (rsp.withSchema) {
taosArrayDestroyP(rsp.blockSchema, (FDelete)tDeleteSSchemaWrapper);
}
if (rsp.withTbName) {
taosArrayDestroyP(rsp.blockTbName, (FDelete)taosMemoryFree);
}
return 0;
}
@ -545,24 +405,9 @@ int32_t tqProcessVgDeleteReq(STQ* pTq, char* msg, int32_t msgLen) {
int32_t code = taosHashRemove(pTq->handles, pReq->subKey, strlen(pReq->subKey));
ASSERT(code == 0);
TXN txn;
if (tdbTxnOpen(&txn, 0, tdbDefaultMalloc, tdbDefaultFree, NULL, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED) < 0) {
if (tqMetaDeleteHandle(pTq, pReq->subKey) < 0) {
ASSERT(0);
}
if (tdbBegin(pTq->pMetaStore, &txn) < 0) {
ASSERT(0);
}
if (tdbTbDelete(pTq->pExecStore, pReq->subKey, (int)strlen(pReq->subKey), &txn) < 0) {
/*ASSERT(0);*/
}
if (tdbCommit(pTq->pMetaStore, &txn) < 0) {
ASSERT(0);
}
return 0;
}
@ -620,7 +465,7 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
atomic_add_fetch_32(&pHandle->epoch, 1);
}
if (tqStoreHandle(pTq, req.subKey, pHandle) < 0) {
if (tqMetaSaveHandle(pTq, req.subKey, pHandle) < 0) {
// TODO
}
return 0;

134
source/dnode/vnode/src/tq/tqMeta.c
View File

@ -12,3 +12,137 @@
* 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/>.
*/
#include "tdbInt.h"
#include "tq.h"
int tqExecKeyCompare(const void* pKey1, int32_t kLen1, const void* pKey2, int32_t kLen2) {
return strcmp(pKey1, pKey2);
}
int32_t tqMetaOpen(STQ* pTq) {
if (tdbOpen(pTq->path, 16 * 1024, 1, &pTq->pMetaStore) < 0) {
ASSERT(0);
}
if (tdbTbOpen("handles", -1, -1, tqExecKeyCompare, pTq->pMetaStore, &pTq->pExecStore) < 0) {
ASSERT(0);
}
TXN txn;
if (tdbTxnOpen(&txn, 0, tdbDefaultMalloc, tdbDefaultFree, NULL, 0) < 0) {
ASSERT(0);
}
TBC* pCur;
if (tdbTbcOpen(pTq->pExecStore, &pCur, &txn) < 0) {
ASSERT(0);
}
void* pKey;
int kLen;
void* pVal;
int vLen;
tdbTbcMoveToFirst(pCur);
SDecoder decoder;
while (tdbTbcNext(pCur, &pKey, &kLen, &pVal, &vLen) == 0) {
STqHandle handle;
tDecoderInit(&decoder, (uint8_t*)pVal, vLen);
tDecodeSTqHandle(&decoder, &handle);
handle.pWalReader = walOpenReadHandle(pTq->pVnode->pWal);
for (int32_t i = 0; i < 5; i++) {
handle.execHandle.pExecReader[i] = tqInitSubmitMsgScanner(pTq->pVnode->pMeta);
}
if (handle.execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
for (int32_t i = 0; i < 5; i++) {
SReadHandle reader = {
.reader = handle.execHandle.pExecReader[i],
.meta = pTq->pVnode->pMeta,
.pMsgCb = &pTq->pVnode->msgCb,
};
handle.execHandle.exec.execCol.task[i] =
qCreateStreamExecTaskInfo(handle.execHandle.exec.execCol.qmsg, &reader);
ASSERT(handle.execHandle.exec.execCol.task[i]);
}
} else {
handle.execHandle.exec.execDb.pFilterOutTbUid =
taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
}
taosHashPut(pTq->handles, pKey, kLen, &handle, sizeof(STqHandle));
}
if (tdbTxnClose(&txn) < 0) {
ASSERT(0);
}
return 0;
}
int32_t tqMetaClose(STQ* pTq) {
tdbClose(pTq->pMetaStore);
return 0;
}
int32_t tqMetaSaveHandle(STQ* pTq, const char* key, const STqHandle* pHandle) {
int32_t code;
int32_t vlen;
tEncodeSize(tEncodeSTqHandle, pHandle, vlen, code);
ASSERT(code == 0);
void* buf = taosMemoryCalloc(1, vlen);
if (buf == NULL) {
ASSERT(0);
}
SEncoder encoder;
tEncoderInit(&encoder, buf, vlen);
if (tEncodeSTqHandle(&encoder, pHandle) < 0) {
ASSERT(0);
}
TXN txn;
if (tdbTxnOpen(&txn, 0, tdbDefaultMalloc, tdbDefaultFree, NULL, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED) < 0) {
ASSERT(0);
}
if (tdbBegin(pTq->pMetaStore, &txn) < 0) {
ASSERT(0);
}
if (tdbTbUpsert(pTq->pExecStore, key, (int)strlen(key), buf, vlen, &txn) < 0) {
ASSERT(0);
}
if (tdbCommit(pTq->pMetaStore, &txn) < 0) {
ASSERT(0);
}
tEncoderClear(&encoder);
taosMemoryFree(buf);
return 0;
}
int32_t tqMetaDeleteHandle(STQ* pTq, const char* key) {
TXN txn;
if (tdbTxnOpen(&txn, 0, tdbDefaultMalloc, tdbDefaultFree, NULL, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED) < 0) {
ASSERT(0);
}
if (tdbBegin(pTq->pMetaStore, &txn) < 0) {
ASSERT(0);
}
if (tdbTbDelete(pTq->pExecStore, key, (int)strlen(key), &txn) < 0) {
/*ASSERT(0);*/
}
if (tdbCommit(pTq->pMetaStore, &txn) < 0) {
ASSERT(0);
}
return 0;
}

2
source/dnode/vnode/src/tq/tqPush.c
View File

@ -12,3 +12,5 @@
* 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/>.
*/
#include "tq.h"

2
source/dnode/vnode/test/tsdbSmaTest.cpp
View File

@ -368,7 +368,7 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
SDiskCfg pDisks = {0};
pDisks.level = 0;
pDisks.primary = 1;
strncpy(pDisks.dir, "/var/lib/taos", TSDB_FILENAME_LEN);
strncpy(pDisks.dir, TD_DATA_DIR_PATH, TSDB_FILENAME_LEN);
int32_t numOfDisks = 1;
pTsdb->pTfs = tfsOpen(&pDisks, numOfDisks);
EXPECT_NE(pTsdb->pTfs, nullptr);

2
source/libs/catalog/test/catalogTests.cpp
View File

@ -137,7 +137,7 @@ void ctgTestInitLogFile() {
tsAsyncLog = 0;
qDebugFlag = 159;
strcpy(tsLogDir, "/var/log/taos");
strcpy(tsLogDir, TD_LOG_DIR_PATH);
ctgdEnableDebug("api");
ctgdEnableDebug("meta");

4
source/libs/function/src/builtinsimpl.c
View File

@ -1646,8 +1646,8 @@ bool leastSQRFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInf
pInfo->startVal = IS_FLOAT_TYPE(pCtx->param[1].param.nType) ? pCtx->param[1].param.d :
(double)pCtx->param[1].param.i;
pInfo->stepVal = IS_FLOAT_TYPE(pCtx->param[1].param.nType) ? pCtx->param[2].param.d :
(double)pCtx->param[1].param.i;
pInfo->stepVal = IS_FLOAT_TYPE(pCtx->param[2].param.nType) ? pCtx->param[2].param.d :
(double)pCtx->param[2].param.i;
return true;
}

29
source/libs/parser/src/parInsert.c
View File

@ -1525,14 +1525,14 @@ int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, char* tN
SKvParam param = {.builder = &tagBuilder};
for (int c = 0; c < tags->numOfBound; ++c) {
SSchema* pTagSchema = &pSchema[tags->boundColumns[c]];
param.schema = pTagSchema;
if (bind[c].is_null && bind[c].is_null[0]) {
KvRowAppend(&pBuf, NULL, 0, &param);
continue;
}
SSchema* pTagSchema = &pSchema[tags->boundColumns[c]];
param.schema = pTagSchema;
int32_t colLen = pTagSchema->bytes;
if (IS_VAR_DATA_TYPE(pTagSchema->type)) {
colLen = bind[c].length[0];
@ -1724,18 +1724,23 @@ int32_t qBindStmtSingleColValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBu
return TSDB_CODE_SUCCESS;
}
int32_t buildBoundFields(SParsedDataColInfo* boundInfo, SSchema* pSchema, int32_t* fieldNum, TAOS_FIELD** fields) {
int32_t buildBoundFields(SParsedDataColInfo* boundInfo, SSchema* pSchema, int32_t* fieldNum, TAOS_FIELD_E** fields, uint8_t timePrec) {
if (fields) {
*fields = taosMemoryCalloc(boundInfo->numOfBound, sizeof(TAOS_FIELD));
if (NULL == *fields) {
return TSDB_CODE_OUT_OF_MEMORY;
}
SSchema* schema = &pSchema[boundInfo->boundColumns[0]];
if (TSDB_DATA_TYPE_TIMESTAMP == schema->type) {
(*fields)[0].precision = timePrec;
}
for (int32_t i = 0; i < boundInfo->numOfBound; ++i) {
SSchema* pTagSchema = &pSchema[boundInfo->boundColumns[i]];
strcpy((*fields)[i].name, pTagSchema->name);
(*fields)[i].type = pTagSchema->type;
(*fields)[i].bytes = pTagSchema->bytes;
schema = &pSchema[boundInfo->boundColumns[i]];
strcpy((*fields)[i].name, schema->name);
(*fields)[i].type = schema->type;
(*fields)[i].bytes = schema->bytes;
}
}
@ -1744,7 +1749,7 @@ int32_t buildBoundFields(SParsedDataColInfo* boundInfo, SSchema* pSchema, int32_
return TSDB_CODE_SUCCESS;
}
int32_t qBuildStmtTagFields(void* pBlock, void* boundTags, int32_t* fieldNum, TAOS_FIELD** fields) {
int32_t qBuildStmtTagFields(void* pBlock, void* boundTags, int32_t* fieldNum, TAOS_FIELD_E** fields) {
STableDataBlocks* pDataBlock = (STableDataBlocks*)pBlock;
SParsedDataColInfo* tags = (SParsedDataColInfo*)boundTags;
if (NULL == tags) {
@ -1759,12 +1764,12 @@ int32_t qBuildStmtTagFields(void* pBlock, void* boundTags, int32_t* fieldNum, TA
return TSDB_CODE_SUCCESS;
}
CHECK_CODE(buildBoundFields(tags, pSchema, fieldNum, fields));
CHECK_CODE(buildBoundFields(tags, pSchema, fieldNum, fields, 0));
return TSDB_CODE_SUCCESS;
}
int32_t qBuildStmtColFields(void* pBlock, int32_t* fieldNum, TAOS_FIELD** fields) {
int32_t qBuildStmtColFields(void* pBlock, int32_t* fieldNum, TAOS_FIELD_E** fields) {
STableDataBlocks* pDataBlock = (STableDataBlocks*)pBlock;
SSchema* pSchema = getTableColumnSchema(pDataBlock->pTableMeta);
if (pDataBlock->boundColumnInfo.numOfBound <= 0) {
@ -1776,7 +1781,7 @@ int32_t qBuildStmtColFields(void* pBlock, int32_t* fieldNum, TAOS_FIELD** fields
return TSDB_CODE_SUCCESS;
}
CHECK_CODE(buildBoundFields(&pDataBlock->boundColumnInfo, pSchema, fieldNum, fields));
CHECK_CODE(buildBoundFields(&pDataBlock->boundColumnInfo, pSchema, fieldNum, fields, pDataBlock->pTableMeta->tableInfo.precision));
return TSDB_CODE_SUCCESS;
}

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

@ -830,7 +830,8 @@ static EDealRes translateComparisonOperator(STranslateContext* pCxt, SOperatorNo
if (!IS_VAR_DATA_TYPE(((SExprNode*)(pOp->pLeft))->resType.type)) {
return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, ((SExprNode*)(pOp->pLeft))->aliasName);
}
if (QUERY_NODE_VALUE != nodeType(pOp->pRight) || !IS_STR_DATA_TYPE(((SExprNode*)(pOp->pRight))->resType.type)) {
if (QUERY_NODE_VALUE != nodeType(pOp->pRight) ||
((!IS_STR_DATA_TYPE(((SExprNode*)(pOp->pRight))->resType.type)) && (((SExprNode*)(pOp->pRight))->resType.type != TSDB_DATA_TYPE_NULL))) {
return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, ((SExprNode*)(pOp->pRight))->aliasName);
}
}

2
source/libs/qworker/test/qworkerTests.cpp
View File

@ -108,7 +108,7 @@ void qwtInitLogFile() {
tsAsyncLog = 0;
qDebugFlag = 159;
strcpy(tsLogDir, "/var/log/taos");
strcpy(tsLogDir, TD_LOG_DIR_PATH);
if (taosInitLog(defaultLogFileNamePrefix, maxLogFileNum) < 0) {
printf("failed to open log file in directory:%s\n", tsLogDir);

2
source/libs/scalar/test/filter/filterTests.cpp
View File

@ -60,7 +60,7 @@ void flttInitLogFile() {
tsAsyncLog = 0;
qDebugFlag = 159;
strcpy(tsLogDir, "/var/log/taos");
strcpy(tsLogDir, TD_LOG_DIR_PATH);
if (taosInitLog(defaultLogFileNamePrefix, maxLogFileNum) < 0) {
printf("failed to open log file in directory:%s\n", tsLogDir);

10
source/libs/scalar/test/scalar/CMakeLists.txt
View File

@ -17,7 +17,9 @@ TARGET_INCLUDE_DIRECTORIES(
PUBLIC "${TD_SOURCE_DIR}/source/libs/parser/inc"
PRIVATE "${TD_SOURCE_DIR}/source/libs/scalar/inc"
)
add_test(
NAME scalarTest
COMMAND scalarTest
)
if(NOT TD_WINDOWS)
add_test(
NAME scalarTest
COMMAND scalarTest
)
endif(NOT TD_WINDOWS)

2
source/libs/scalar/test/scalar/scalarTests.cpp
View File

@ -74,7 +74,7 @@ void scltInitLogFile() {
tsAsyncLog = 0;
qDebugFlag = 159;
strcpy(tsLogDir, "/var/log/taos");
strcpy(tsLogDir, TD_LOG_DIR_PATH);
if (taosInitLog(defaultLogFileNamePrefix, maxLogFileNum) < 0) {
printf("failed to open log file in directory:%s\n", tsLogDir);

2
source/libs/scheduler/test/schedulerTests.cpp
View File

@ -79,7 +79,7 @@ void schtInitLogFile() {
tsAsyncLog = 0;
qDebugFlag = 159;
strcpy(tsLogDir, "/var/log/taos");
strcpy(tsLogDir, TD_LOG_DIR_PATH);
if (taosInitLog(defaultLogFileNamePrefix, maxLogFileNum) < 0) {
printf("failed to open log file in directory:%s\n", tsLogDir);

6
source/os/CMakeLists.txt
View File

@ -10,7 +10,11 @@ target_include_directories(
PUBLIC "${TD_SOURCE_DIR}/contrib/msvcregex"
)
# iconv
find_path(IconvApiIncludes iconv.h PATHS)
if(TD_WINDOWS)
find_path(IconvApiIncludes iconv.h "${TD_SOURCE_DIR}/contrib/iconv")
else()
find_path(IconvApiIncludes iconv.h PATHS)
endif(TD_WINDOWS)
if(NOT IconvApiIncludes)
add_definitions(-DDISALLOW_NCHAR_WITHOUT_ICONV)
endif ()

4
source/os/src/osEnv.c
View File

@ -70,11 +70,11 @@ void osDefaultInit() {
#elif defined(_TD_DARWIN_64)
if (configDir[0] == 0) {
strcpy(configDir, "/tmp/taosd");
strcpy(configDir, "/usr/local/etc/taos");
}
strcpy(tsDataDir, "/usr/local/var/lib/taos");
strcpy(tsLogDir, "/usr/local/var/log/taos");
strcpy(tsTempDir, "/usr/local/etc/taos");
strcpy(tsTempDir, "/tmp/taosd");
strcpy(tsOsName, "Darwin");
#else

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

@ -186,9 +186,9 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_SNODE_ALREADY_EXIST, "Snode already exists"
TAOS_DEFINE_ERROR(TSDB_CODE_MND_SNODE_NOT_EXIST, "Snode not there")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_BNODE_ALREADY_EXIST, "Bnode already exists")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_BNODE_NOT_EXIST, "Bnode not there")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_FEW_MNODES, "Too few mnodes")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_MNODE_DEPLOYED, "Mnode deployed in this dnode")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_CANT_DROP_MASTER, "Can't drop mnode which is master")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_FEW_MNODES, "The replicas of mnode cannot less than 1")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_MNODES, "The replicas of mnode cannot exceed 3")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_CANT_DROP_MASTER, "Can't drop mnode which is LEADER")
// mnode-acct
TAOS_DEFINE_ERROR(TSDB_CODE_MND_ACCT_ALREADY_EXIST, "Account already exists")

318
tests/script/api/batchprepare.c
View File

@ -10,6 +10,9 @@
#include "../../../include/client/taos.h"
#define FUNCTION_TEST_IDX 1
#define TIME_PRECISION_MILLI 0
#define TIME_PRECISION_MICRO 1
#define TIME_PRECISION_NANO 2
int32_t shortColList[] = {TSDB_DATA_TYPE_TIMESTAMP, TSDB_DATA_TYPE_INT};
int32_t fullColList[] = {TSDB_DATA_TYPE_TIMESTAMP, TSDB_DATA_TYPE_BOOL, TSDB_DATA_TYPE_TINYINT, TSDB_DATA_TYPE_UTINYINT, TSDB_DATA_TYPE_SMALLINT, TSDB_DATA_TYPE_USMALLINT, TSDB_DATA_TYPE_INT, TSDB_DATA_TYPE_UINT, TSDB_DATA_TYPE_BIGINT, TSDB_DATA_TYPE_UBIGINT, TSDB_DATA_TYPE_FLOAT, TSDB_DATA_TYPE_DOUBLE, TSDB_DATA_TYPE_BINARY, TSDB_DATA_TYPE_NCHAR};
@ -32,6 +35,8 @@ typedef enum {
BP_BIND_COL,
} BP_BIND_TYPE;
#define BP_BIND_TYPE_STR(t) (((t) == BP_BIND_COL) ? "column" : "tag")
OperInfo operInfo[] = {
{">", 2, false},
{">=", 2, false},
@ -57,11 +62,12 @@ FuncInfo funcInfo[] = {
{"min", 1},
};
#define BP_STARTUP_TS 1591060628000
char *bpStbPrefix = "st";
char *bpTbPrefix = "t";
int32_t bpDefaultStbId = 1;
int64_t bpTs;
//char *operatorList[] = {">", ">=", "<", "<=", "=", "<>", "in", "not in"};
//char *varoperatorList[] = {">", ">=", "<", "<=", "=", "<>", "in", "not in", "like", "not like", "match", "nmatch"};
@ -188,8 +194,10 @@ typedef struct {
bool printCreateTblSql;
bool printQuerySql;
bool printStmtSql;
bool printVerbose;
bool autoCreateTbl;
bool numericParam;
uint8_t precision;
int32_t rowNum; //row num for one table
int32_t bindColNum;
int32_t bindTagNum;
@ -209,12 +217,15 @@ typedef struct {
int32_t caseRunNum; // total run case num
} CaseCtrl;
#if 1
#if 0
CaseCtrl gCaseCtrl = { // default
.precision = TIME_PRECISION_MICRO,
.bindNullNum = 0,
.printCreateTblSql = false,
.printQuerySql = true,
.printStmtSql = true,
.printVerbose = false,
.printRes = false,
.autoCreateTbl = false,
.numericParam = false,
.rowNum = 0,
@ -230,7 +241,6 @@ CaseCtrl gCaseCtrl = { // default
.funcIdxListNum = 0,
.funcIdxList = NULL,
.checkParamNum = false,
.printRes = false,
.runTimes = 0,
.caseIdx = -1,
.caseNum = -1,
@ -240,26 +250,35 @@ CaseCtrl gCaseCtrl = { // default
#endif
#if 0
#if 1
CaseCtrl gCaseCtrl = {
.precision = TIME_PRECISION_MILLI,
.bindNullNum = 0,
.printCreateTblSql = true,
.printCreateTblSql = false,
.printQuerySql = true,
.printStmtSql = true,
.printVerbose = false,
.printRes = true,
.autoCreateTbl = false,
.numericParam = false,
.rowNum = 0,
.bindColNum = 0,
.bindTagNum = 0,
.bindRowNum = 0,
.bindColTypeNum = 0,
.bindColTypeList = NULL,
.bindTagTypeNum = 0,
.bindTagTypeList = NULL,
.optrIdxListNum = 0,
.optrIdxList = NULL,
.funcIdxListNum = 0,
.funcIdxList = NULL,
.checkParamNum = false,
.printRes = false,
.runTimes = 0,
.caseIdx = 1,
.caseNum = 1,
.caseIdx = -1,
.caseNum = -1,
.caseRunIdx = -1,
.caseRunNum = 1,
.caseRunNum = -1,
};
#endif
@ -891,7 +910,6 @@ int32_t prepareColData(BP_BIND_TYPE bType, BindData *data, int32_t bindIdx, int3
int32_t prepareInsertData(BindData *data) {
static int64_t tsData = 1591060628000;
uint64_t allRowNum = gCurCase->rowNum * gCurCase->tblNum;
data->colNum = 0;
@ -918,7 +936,7 @@ int32_t prepareInsertData(BindData *data) {
}
for (int32_t i = 0; i < allRowNum; ++i) {
data->tsData[i] = tsData++;
data->tsData[i] = bpTs++;
data->boolData[i] = (bool)(i % 2);
data->tinyData[i] = (int8_t)i;
data->utinyData[i] = (uint8_t)(i+1);
@ -956,7 +974,6 @@ int32_t prepareInsertData(BindData *data) {
}
int32_t prepareQueryCondData(BindData *data, int32_t tblIdx) {
static int64_t tsData = 1591060628000;
uint64_t bindNum = gCurCase->rowNum / gCurCase->bindRowNum;
data->colNum = 0;
@ -982,7 +999,7 @@ int32_t prepareQueryCondData(BindData *data, int32_t tblIdx) {
}
for (int32_t i = 0; i < bindNum; ++i) {
data->tsData[i] = tsData + tblIdx*gCurCase->rowNum + rand()%gCurCase->rowNum;
data->tsData[i] = bpTs + tblIdx*gCurCase->rowNum + rand()%gCurCase->rowNum;
data->boolData[i] = (bool)(tblIdx*gCurCase->rowNum + rand() % gCurCase->rowNum);
data->tinyData[i] = (int8_t)(tblIdx*gCurCase->rowNum + rand() % gCurCase->rowNum);
data->utinyData[i] = (uint8_t)(tblIdx*gCurCase->rowNum + rand() % gCurCase->rowNum);
@ -1014,7 +1031,6 @@ int32_t prepareQueryCondData(BindData *data, int32_t tblIdx) {
int32_t prepareQueryMiscData(BindData *data, int32_t tblIdx) {
static int64_t tsData = 1591060628000;
uint64_t bindNum = gCurCase->rowNum / gCurCase->bindRowNum;
data->colNum = 0;
@ -1040,7 +1056,7 @@ int32_t prepareQueryMiscData(BindData *data, int32_t tblIdx) {
}
for (int32_t i = 0; i < bindNum; ++i) {
data->tsData[i] = tsData + tblIdx*gCurCase->rowNum + rand()%gCurCase->rowNum;
data->tsData[i] = bpTs + tblIdx*gCurCase->rowNum + rand()%gCurCase->rowNum;
data->boolData[i] = (bool)(tblIdx*gCurCase->rowNum + rand() % gCurCase->rowNum);
data->tinyData[i] = (int8_t)(tblIdx*gCurCase->rowNum + rand() % gCurCase->rowNum);
data->utinyData[i] = (uint8_t)(tblIdx*gCurCase->rowNum + rand() % gCurCase->rowNum);
@ -1202,39 +1218,7 @@ int32_t bpAppendValueString(char *buf, int type, void *value, int32_t valueLen,
}
int32_t bpBindParam(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind) {
static int32_t n = 0;
if (gCurCase->bindRowNum > 1) {
if (0 == (n++%2)) {
if (taos_stmt_bind_param_batch(stmt, bind)) {
printf("!!!taos_stmt_bind_param_batch error:%s\n", taos_stmt_errstr(stmt));
exit(1);
}
} else {
for (int32_t i = 0; i < gCurCase->bindColNum; ++i) {
if (taos_stmt_bind_single_param_batch(stmt, bind++, i)) {
printf("!!!taos_stmt_bind_single_param_batch error:%s\n", taos_stmt_errstr(stmt));
exit(1);
}
}
}
} else {
if (0 == (n++%2)) {
if (taos_stmt_bind_param_batch(stmt, bind)) {
printf("!!!taos_stmt_bind_param_batch error:%s\n", taos_stmt_errstr(stmt));
exit(1);
}
} else {
if (taos_stmt_bind_param(stmt, bind)) {
printf("!!!taos_stmt_bind_param error:%s\n", taos_stmt_errstr(stmt));
exit(1);
}
}
}
return 0;
}
void bpCheckIsInsert(TAOS_STMT *stmt, int32_t insert) {
int32_t isInsert = 0;
@ -1280,15 +1264,12 @@ void bpCheckAffectedRowsOnce(TAOS_STMT *stmt, int32_t expectedNum) {
}
void bpCheckQueryResult(TAOS_STMT *stmt, TAOS *taos, char *stmtSql, TAOS_MULTI_BIND* bind) {
TAOS_RES* res = taos_stmt_use_result(stmt);
int32_t sqlResNum = 0;
int32_t stmtResNum = 0;
bpFetchRows(res, gCaseCtrl.printRes, &stmtResNum);
// query using sql
char sql[1024];
int32_t len = 0;
char* p = stmtSql;
char* s = NULL;
int32_t sqlResNum = 0;
for (int32_t i = 0; true; ++i, p=s+1) {
s = strchr(p, '?');
@ -1313,6 +1294,12 @@ void bpCheckQueryResult(TAOS_STMT *stmt, TAOS *taos, char *stmtSql, TAOS_MULTI_B
}
bpExecQuery(taos, sql, gCaseCtrl.printRes, &sqlResNum);
// query using stmt
TAOS_RES* res = taos_stmt_use_result(stmt);
int32_t stmtResNum = 0;
bpFetchRows(res, gCaseCtrl.printRes, &stmtResNum);
if (sqlResNum != stmtResNum) {
printf("!!!sql res num %d mis-match stmt res num %d\n", sqlResNum, stmtResNum);
exit(1);
@ -1321,9 +1308,165 @@ void bpCheckQueryResult(TAOS_STMT *stmt, TAOS *taos, char *stmtSql, TAOS_MULTI_B
printf("***sql res num match stmt res num %d\n", stmtResNum);
}
void bpCheckColTagFields(TAOS_STMT *stmt, int32_t fieldNum, TAOS_FIELD_E* pFields, int32_t expecteNum, TAOS_MULTI_BIND* pBind, BP_BIND_TYPE type) {
int32_t code = 0;
if (fieldNum != expecteNum) {
printf("!!!%s field num %d mis-match expect num %d\n", BP_BIND_TYPE_STR(type), fieldNum, expecteNum);
exit(1);
}
if (type == BP_BIND_COL) {
if (pFields[0].precision != gCaseCtrl.precision) {
printf("!!!db precision %d mis-match expect %d\n", pFields[0].precision, gCaseCtrl.precision);
exit(1);
}
}
for (int32_t i = 0; i < fieldNum; ++i) {
if (pFields[i].type != pBind[i].buffer_type) {
printf("!!!%s %dth field type %d mis-match expect type %d\n", BP_BIND_TYPE_STR(type), i, pFields[i].type, pBind[i].buffer_type);
exit(1);
}
if (pFields[i].type == TSDB_DATA_TYPE_BINARY) {
if (pFields[i].bytes != (pBind[i].buffer_length + 2)) {
printf("!!!%s %dth field len %d mis-match expect len %d\n", BP_BIND_TYPE_STR(type), i, pFields[i].bytes, (pBind[i].buffer_length + 2));
exit(1);
}
} else if (pFields[i].type == TSDB_DATA_TYPE_NCHAR) {
if (pFields[i].bytes != (pBind[i].buffer_length * 4 + 2)) {
printf("!!!%s %dth field len %d mis-match expect len %d\n", BP_BIND_TYPE_STR(type), i, pFields[i].bytes, (pBind[i].buffer_length + 2));
exit(1);
}
} else if (pFields[i].bytes != pBind[i].buffer_length) {
printf("!!!%s %dth field len %d mis-match expect len %d\n", BP_BIND_TYPE_STR(type), i, pFields[i].bytes, pBind[i].buffer_length);
exit(1);
}
}
if (type == BP_BIND_COL) {
int fieldType = 0;
int fieldBytes = 0;
for (int32_t i = 0; i < fieldNum; ++i) {
code = taos_stmt_get_param(stmt, i, &fieldType, &fieldBytes);
if (code) {
printf("!!!taos_stmt_get_param error:%s\n", taos_stmt_errstr(stmt));
exit(1);
}
if (pFields[i].type != fieldType) {
printf("!!!%s %dth field type %d mis-match expect type %d\n", BP_BIND_TYPE_STR(type), i, fieldType, pFields[i].type);
exit(1);
}
if (pFields[i].bytes != fieldBytes) {
printf("!!!%s %dth field len %d mis-match expect len %d\n", BP_BIND_TYPE_STR(type), i, fieldBytes, pFields[i].bytes);
exit(1);
}
}
}
if (gCaseCtrl.printVerbose) {
printf("%s fields check passed\n", BP_BIND_TYPE_STR(type));
}
}
void bpCheckTagFields(TAOS_STMT *stmt, TAOS_MULTI_BIND* pBind) {
int32_t code = 0;
int fieldNum = 0;
TAOS_FIELD_E* pFields = NULL;
code = taos_stmt_get_tag_fields(stmt, &fieldNum, &pFields);
if (code != 0){
printf("!!!taos_stmt_get_tag_fields error:%s\n", taos_stmt_errstr(stmt));
exit(1);
}
bpCheckColTagFields(stmt, fieldNum, pFields, gCurCase->bindTagNum, pBind, BP_BIND_TAG);
}
void bpCheckColFields(TAOS_STMT *stmt, TAOS_MULTI_BIND* pBind) {
if (gCurCase->testType == TTYPE_QUERY) {
return;
}
int32_t code = 0;
int fieldNum = 0;
TAOS_FIELD_E* pFields = NULL;
code = taos_stmt_get_col_fields(stmt, &fieldNum, &pFields);
if (code != 0){
printf("!!!taos_stmt_get_col_fields error:%s\n", taos_stmt_errstr(stmt));
exit(1);
}
bpCheckColTagFields(stmt, fieldNum, pFields, gCurCase->bindColNum, pBind, BP_BIND_COL);
}
void bpShowBindParam(TAOS_MULTI_BIND *bind, int32_t num) {
for (int32_t i = 0; i < num; ++i) {
TAOS_MULTI_BIND* b = &bind[i];
printf("Bind %d: type[%d],buf[%p],buflen[%d],len[%],null[%d],num[%d]\n",
i, b->buffer_type, b->buffer, b->buffer_length, b->length ? *b->length : 0, b->is_null ? *b->is_null : 0, b->num);
}
}
int32_t bpBindParam(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind) {
static int32_t n = 0;
bpCheckColFields(stmt, bind);
if (gCurCase->bindRowNum > 1) {
if (0 == (n++%2)) {
if (taos_stmt_bind_param_batch(stmt, bind)) {
printf("!!!taos_stmt_bind_param_batch error:%s\n", taos_stmt_errstr(stmt));
bpShowBindParam(bind, gCurCase->bindColNum);
exit(1);
}
} else {
for (int32_t i = 0; i < gCurCase->bindColNum; ++i) {
if (taos_stmt_bind_single_param_batch(stmt, bind+i, i)) {
printf("!!!taos_stmt_bind_single_param_batch %d error:%s\n", taos_stmt_errstr(stmt), i);
bpShowBindParam(bind, gCurCase->bindColNum);
exit(1);
}
}
}
} else {
if (0 == (n++%2)) {
if (taos_stmt_bind_param_batch(stmt, bind)) {
printf("!!!taos_stmt_bind_param_batch error:%s\n", taos_stmt_errstr(stmt));
bpShowBindParam(bind, gCurCase->bindColNum);
exit(1);
}
} else {
if (taos_stmt_bind_param(stmt, bind)) {
printf("!!!taos_stmt_bind_param error:%s\n", taos_stmt_errstr(stmt));
bpShowBindParam(bind, gCurCase->bindColNum);
exit(1);
}
}
}
return 0;
}
int32_t bpSetTableNameTags(BindData *data, int32_t tblIdx, char *tblName, TAOS_STMT *stmt) {
int32_t code = 0;
if (gCurCase->bindTagNum > 0) {
return taos_stmt_set_tbname_tags(stmt, tblName, data->pTags + tblIdx * gCurCase->bindTagNum);
if ((rand() % 2) == 0) {
code = taos_stmt_set_tbname(stmt, tblName);
if (code != 0){
printf("!!!taos_stmt_set_tbname error:%s\n", taos_stmt_errstr(stmt));
exit(1);
}
bpCheckTagFields(stmt, data->pTags + tblIdx * gCurCase->bindTagNum);
return taos_stmt_set_tags(stmt, data->pTags + tblIdx * gCurCase->bindTagNum);
} else {
return taos_stmt_set_tbname_tags(stmt, tblName, data->pTags + tblIdx * gCurCase->bindTagNum);
}
} else {
return taos_stmt_set_tbname(stmt, tblName);
}
@ -1755,7 +1898,7 @@ int insertAUTOTest1(TAOS_STMT *stmt, TAOS *taos) {
if (gCurCase->tblNum > 1) {
char buf[32];
sprintf(buf, "t%d", t);
code = taos_stmt_set_tbname_tags(stmt, buf, data.pTags + t * gCurCase->bindTagNum);
code = bpSetTableNameTags(&data, t, buf, stmt);
if (code != 0){
printf("!!!taos_stmt_set_tbname_tags error:%s\n", taos_stmt_errstr(stmt));
exit(1);
@ -2223,14 +2366,48 @@ void generateCreateTableSQL(char *buf, int32_t tblIdx, int32_t colNum, int32_t *
}
}
char *bpPrecisionStr(uint8_t precision) {
switch (precision) {
case TIME_PRECISION_MILLI:
return "ms";
case TIME_PRECISION_MICRO:
return "us";
case TIME_PRECISION_NANO:
return "ns";
default:
return "unknwon";
}
}
void bpSetStartupTs() {
switch (gCaseCtrl.precision) {
case TIME_PRECISION_MILLI:
bpTs = BP_STARTUP_TS;
break;
case TIME_PRECISION_MICRO:
bpTs = BP_STARTUP_TS * 1000;
break;
case TIME_PRECISION_NANO:
bpTs = BP_STARTUP_TS * 1000000;
break;
default:
bpTs = BP_STARTUP_TS;
break;
}
}
void prepare(TAOS *taos, int32_t colNum, int32_t *colList, int prepareStb) {
TAOS_RES *result;
int code;
char createDbSql[128] = {0};
result = taos_query(taos, "drop database demo");
taos_free_result(result);
result = taos_query(taos, "create database demo keep 36500");
sprintf(createDbSql, "create database demo keep 36500 precision \"%s\"", bpPrecisionStr(gCaseCtrl.precision));
printf("\tCreate Database SQL:%s\n", createDbSql);
result = taos_query(taos, createDbSql);
code = taos_errno(result);
if (code != 0) {
printf("!!!failed to create database, reason:%s\n", taos_errstr(result));
@ -2278,6 +2455,8 @@ int32_t runCase(TAOS *taos, int32_t caseIdx, int32_t caseRunIdx, bool silent) {
CaseCfg cfg = gCase[caseIdx];
CaseCfg cfgBk;
gCurCase = &cfg;
bpSetStartupTs();
if ((gCaseCtrl.bindColTypeNum || gCaseCtrl.bindColNum) && (gCurCase->colNum != gFullColNum)) {
return 1;
@ -2413,22 +2592,28 @@ void* runCaseList(TAOS *taos) {
}
void runAll(TAOS *taos) {
#if 1
strcpy(gCaseCtrl.caseCatalog, "Normal Test");
strcpy(gCaseCtrl.caseCatalog, "Default Test");
printf("%s Begin\n", gCaseCtrl.caseCatalog);
runCaseList(taos);
strcpy(gCaseCtrl.caseCatalog, "Micro DB precision Test");
printf("%s Begin\n", gCaseCtrl.caseCatalog);
gCaseCtrl.precision = TIME_PRECISION_MICRO;
runCaseList(taos);
gCaseCtrl.precision = TIME_PRECISION_MILLI;
strcpy(gCaseCtrl.caseCatalog, "Nano DB precision Test");
printf("%s Begin\n", gCaseCtrl.caseCatalog);
gCaseCtrl.precision = TIME_PRECISION_NANO;
runCaseList(taos);
gCaseCtrl.precision = TIME_PRECISION_MILLI;
strcpy(gCaseCtrl.caseCatalog, "Auto Create Table Test");
gCaseCtrl.autoCreateTbl = true;
printf("%s Begin\n", gCaseCtrl.caseCatalog);
runCaseList(taos);
gCaseCtrl.autoCreateTbl = false;
#endif
/*
strcpy(gCaseCtrl.caseCatalog, "Null Test");
printf("%s Begin\n", gCaseCtrl.caseCatalog);
gCaseCtrl.bindNullNum = 1;
@ -2441,6 +2626,7 @@ void runAll(TAOS *taos) {
runCaseList(taos);
gCaseCtrl.bindRowNum = 0;
#if 0
strcpy(gCaseCtrl.caseCatalog, "Row Num Test");
printf("%s Begin\n", gCaseCtrl.caseCatalog);
gCaseCtrl.rowNum = 1000;
@ -2448,23 +2634,21 @@ void runAll(TAOS *taos) {
runCaseList(taos);
gCaseCtrl.rowNum = 0;
gCaseCtrl.printRes = true;
*/
strcpy(gCaseCtrl.caseCatalog, "Runtimes Test");
printf("%s Begin\n", gCaseCtrl.caseCatalog);
gCaseCtrl.runTimes = 2;
runCaseList(taos);
gCaseCtrl.runTimes = 0;
#endif
#if 1
strcpy(gCaseCtrl.caseCatalog, "Check Param Test");
printf("%s Begin\n", gCaseCtrl.caseCatalog);
gCaseCtrl.checkParamNum = true;
runCaseList(taos);
gCaseCtrl.checkParamNum = false;
#endif
/*
#if 0
strcpy(gCaseCtrl.caseCatalog, "Bind Col Num Test");
printf("%s Begin\n", gCaseCtrl.caseCatalog);
gCaseCtrl.bindColNum = 6;
@ -2476,7 +2660,7 @@ void runAll(TAOS *taos) {
gCaseCtrl.bindColTypeNum = tListLen(bindColTypeList);
gCaseCtrl.bindColTypeList = bindColTypeList;
runCaseList(taos);
*/
#endif
printf("All Test End\n");
}

1
tests/script/jenkins/basic.txt
View File

@ -57,7 +57,6 @@
# ---- mnode
./test.sh -f tsim/mnode/basic1.sim
./test.sh -f tsim/mnode/basic2.sim
./test.sh -f tsim/mnode/basic3.sim
# ---- show
./test.sh -f tsim/show/basic.sim

11
tests/script/tsim/mnode/basic3.sim
View File

@ -2,14 +2,17 @@ system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/deploy.sh -n dnode2 -i 2
system sh/deploy.sh -n dnode3 -i 3
system sh/deploy.sh -n dnode4 -i 4
system sh/exec.sh -n dnode1 -s start
system sh/exec.sh -n dnode2 -s start
system sh/exec.sh -n dnode3 -s start
system sh/exec.sh -n dnode4 -s start
sql connect
print =============== step1: create dnodes
sql create dnode $hostname port 7200
sql create dnode $hostname port 7300
sql create dnode $hostname port 7400
$x = 0
step1:
@ -32,6 +35,7 @@ endi
print =============== step2: create mnode 2
sql create mnode on dnode 2
sql create mnode on dnode 3
sql_error create mnode on dnode 4
$x = 0
step2:
@ -106,6 +110,10 @@ print $data(1)[0] $data(1)[1] $data(1)[2]
print $data(2)[0] $data(2)[1] $data(2)[2]
print $data(3)[0] $data(3)[1] $data(3)[2]
if $data(2)[2] != OFFLINE then
goto step5
endi
sql show users
if $rows != 2 then
return -1
@ -134,4 +142,5 @@ endi
system sh/exec.sh -n dnode1 -s stop
system sh/exec.sh -n dnode2 -s stop
system sh/exec.sh -n dnode3 -s stop
system sh/exec.sh -n dnode3 -s stop
system sh/exec.sh -n dnode4 -s stop

18
tests/script/tsim/trans/create_db.sim
View File

@ -76,14 +76,6 @@ if $data[0][3] != d1 then
return -1
endi
if $data[0][4] != create-db then
return -1
endi
if $data[0][7] != @Unable to establish connection@ then
return -1
endi
sql_error create database d1 vgroups 2;
print =============== start dnode2
@ -125,15 +117,7 @@ endi
if $data[0][3] != d2 then
return -1
endi
if $data[0][4] != create-db then
return -1
endi
if $data[0][7] != @Unable to establish connection@ then
return -1
endi
return
sql_error create database d2 vgroups 2;
print =============== kill transaction

4
tests/test/c/sdbDump.c
View File

@ -279,9 +279,9 @@ void dumpTrans(SSdb *pSdb, SJson *json) {
tjsonAddIntegerToObject(item, "id", pObj->id);
tjsonAddIntegerToObject(item, "stage", pObj->stage);
tjsonAddIntegerToObject(item, "policy", pObj->policy);
tjsonAddIntegerToObject(item, "type", pObj->type);
tjsonAddIntegerToObject(item, "conflict", pObj->conflict);
tjsonAddIntegerToObject(item, "exec", pObj->exec);
tjsonAddStringToObject(item, "createdTime", i642str(pObj->createdTime));
tjsonAddStringToObject(item, "dbUid", i642str(pObj->dbUid));
tjsonAddStringToObject(item, "dbname", pObj->dbname);
tjsonAddIntegerToObject(item, "commitLogNum", taosArrayGetSize(pObj->commitActions));
tjsonAddIntegerToObject(item, "redoActionNum", taosArrayGetSize(pObj->redoActions));

2
tools/taos-tools

@ -1 +1 @@
Subproject commit 4d83d8c62973506f760bcaa3a33f4665ed9046d0
Subproject commit 717f5aaa5f0a1b4d92bb2ae68858fec554fb5eda