Skip to content

T
TDengine

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

通知 1192
Star 22018
Fork 4786
T
TDengine
“0ca664d876118a9ee5879cfe5c0b715d13d82ed5”上不存在“source/server/dnode/inc/dnodeStatus.h”
提交 9a4ccd2c 编写于 作者: K Kaili Xu
浏览文件
操作

Merge branch 'develop' into fix/TD-4324 

sync with develop
上级 b5fa8bf6 6b3e339c
显示空白变更内容
内联 并排
Showing with 1227 addition and 240 deletion
packaging/tools/makepkg.sh
浏览文件 @ 9a4ccd2c
...@@ -114,6 +114,25 @@ mkdir -p ${install_dir}/examples ...@@ -114,6 +114,25 @@ mkdir -p ${install_dir}/examples
examples_dir="${top_dir}/tests/examples" examples_dir="${top_dir}/tests/examples"
cp -r ${examples_dir}/c ${install_dir}/examples cp -r ${examples_dir}/c ${install_dir}/examples
if [[ "$pagMode" != "lite" ]] && [[ "$cpuType" != "aarch32" ]]; then if [[ "$pagMode" != "lite" ]] && [[ "$cpuType" != "aarch32" ]]; then
if [ -d ${examples_dir}/JDBC/connectionPools/target ]; then
rm -rf ${examples_dir}/JDBC/connectionPools/target
fi
if [ -d ${examples_dir}/JDBC/JDBCDemo/target ]; then
rm -rf ${examples_dir}/JDBC/JDBCDemo/target
fi
if [ -d ${examples_dir}/JDBC/mybatisplus-demo/target ]; then
rm -rf ${examples_dir}/JDBC/mybatisplus-demo/target
fi
if [ -d ${examples_dir}/JDBC/springbootdemo/target ]; then
rm -rf ${examples_dir}/JDBC/springbootdemo/target
fi
if [ -d ${examples_dir}/JDBC/SpringJdbcTemplate/target ]; then
rm -rf ${examples_dir}/JDBC/SpringJdbcTemplate/target
fi
if [ -d ${examples_dir}/JDBC/taosdemo/target ]; then
rm -rf ${examples_dir}/JDBC/taosdemo/target
fi
cp -r ${examples_dir}/JDBC ${install_dir}/examples cp -r ${examples_dir}/JDBC ${install_dir}/examples
cp -r ${examples_dir}/matlab ${install_dir}/examples cp -r ${examples_dir}/matlab ${install_dir}/examples
cp -r ${examples_dir}/python ${install_dir}/examples cp -r ${examples_dir}/python ${install_dir}/examples
......
snap/snapcraft.yaml
浏览文件 @ 9a4ccd2c
name: tdengine name: tdengine
base: core18 base: core18
version: '2.1.1.0' version: '2.1.1.0'
icon: snap/gui/t-dengine.svg icon: snap/gui/t-dengine.svg
summary: an open-source big data platform designed and optimized for IoT. summary: an open-source big data platform designed and optimized for IoT.
......
src/client/src/tscSQLParser.c
浏览文件 @ 9a4ccd2c
...@@ -7187,6 +7187,11 @@ int32_t validateSqlNode(SSqlObj* pSql, SSqlNode* pSqlNode, int32_t index) { ...@@ -7187,6 +7187,11 @@ int32_t validateSqlNode(SSqlObj* pSql, SSqlNode* pSqlNode, int32_t index) {
const char* msg1 = "point interpolation query needs timestamp"; const char* msg1 = "point interpolation query needs timestamp";
const char* msg2 = "too many tables in from clause"; const char* msg2 = "too many tables in from clause";
const char* msg3 = "start(end) time of query range required or time range too large"; const char* msg3 = "start(end) time of query range required or time range too large";
// const char* msg5 = "too many columns in selection clause";
// const char* msg6 = "too many tables in from clause";
// const char* msg7 = "invalid table alias name";
// const char* msg8 = "alias name too long";
const char* msg9 = "only tag query not compatible with normal column filter";
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
...@@ -7326,6 +7331,20 @@ int32_t validateSqlNode(SSqlObj* pSql, SSqlNode* pSqlNode, int32_t index) { ...@@ -7326,6 +7331,20 @@ int32_t validateSqlNode(SSqlObj* pSql, SSqlNode* pSqlNode, int32_t index) {
} }
} }
if (tscQueryTags(pQueryInfo)) {
SExprInfo* pExpr1 = tscSqlExprGet(pQueryInfo, 0);
if (pExpr1->base.functionId != TSDB_FUNC_TID_TAG) {
int32_t numOfCols = (int32_t)taosArrayGetSize(pQueryInfo->colList);
for (int32_t i = 0; i < numOfCols; ++i) {
SColumn* pCols = taosArrayGetP(pQueryInfo->colList, i);
if (pCols->info.flist.numOfFilters > 0) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg9);
}
}
}
}
// parse the having clause in the first place // parse the having clause in the first place
if (validateHavingClause(pQueryInfo, pSqlNode->pHaving, pCmd, pSqlNode->pSelNodeList, joinQuery, timeWindowQuery) != if (validateHavingClause(pQueryInfo, pSqlNode->pHaving, pCmd, pSqlNode->pSelNodeList, joinQuery, timeWindowQuery) !=
TSDB_CODE_SUCCESS) { TSDB_CODE_SUCCESS) {
......
src/client/src/tscServer.c
浏览文件 @ 9a4ccd2c
...@@ -1928,8 +1928,9 @@ int tscProcessTableMetaRsp(SSqlObj *pSql) { ...@@ -1928,8 +1928,9 @@ int tscProcessTableMetaRsp(SSqlObj *pSql) {
} }
} }
tscDebug("0x%"PRIx64" recv table meta, uid:%" PRIu64 ", tid:%d, name:%s", pSql->self, pTableMeta->id.uid, pTableMeta->id.tid, tscDebug("0x%"PRIx64" recv table meta, uid:%" PRIu64 ", tid:%d, name:%s, numOfCols:%d, numOfTags:%d", pSql->self,
tNameGetTableName(&pTableMetaInfo->name)); pTableMeta->id.uid, pTableMeta->id.tid, tNameGetTableName(&pTableMetaInfo->name), pTableMeta->tableInfo.numOfColumns,
pTableMeta->tableInfo.numOfTags);
free(pTableMeta); free(pTableMeta);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
...@@ -2072,7 +2073,7 @@ int tscProcessSTableVgroupRsp(SSqlObj *pSql) { ...@@ -2072,7 +2073,7 @@ int tscProcessSTableVgroupRsp(SSqlObj *pSql) {
pInfo->vgroupList->numOfVgroups = pVgroupMsg->numOfVgroups; pInfo->vgroupList->numOfVgroups = pVgroupMsg->numOfVgroups;
if (pInfo->vgroupList->numOfVgroups <= 0) { if (pInfo->vgroupList->numOfVgroups <= 0) {
tscDebug("0x%"PRIx64" empty vgroup info, no corresponding tables for stable", pSql->self); tscDebug("0x%" PRIx64 " empty vgroup info, no corresponding tables for stable", pSql->self);
} else { } else {
for (int32_t j = 0; j < pInfo->vgroupList->numOfVgroups; ++j) { for (int32_t j = 0; j < pInfo->vgroupList->numOfVgroups; ++j) {
// just init, no need to lock // just init, no need to lock
......
src/client/src/tscStream.c
浏览文件 @ 9a4ccd2c
...@@ -627,6 +627,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p ...@@ -627,6 +627,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
if (pSql->sqlstr == NULL) { if (pSql->sqlstr == NULL) {
tscError("0x%"PRIx64" failed to malloc sql string buffer", pSql->self); tscError("0x%"PRIx64" failed to malloc sql string buffer", pSql->self);
tscFreeSqlObj(pSql); tscFreeSqlObj(pSql);
free(pStream);
return NULL; return NULL;
} }
......
src/client/src/tscSub.c
浏览文件 @ 9a4ccd2c
...@@ -215,7 +215,7 @@ static void tscProcessSubscriptionTimer(void *handle, void *tmrId) { ...@@ -215,7 +215,7 @@ static void tscProcessSubscriptionTimer(void *handle, void *tmrId) {
taosTmrReset(tscProcessSubscriptionTimer, pSub->interval, pSub, tscTmr, &pSub->pTimer); taosTmrReset(tscProcessSubscriptionTimer, pSub->interval, pSub, tscTmr, &pSub->pTimer);
} }
//TODO refactor: extract table list name not simply from the sql
static SArray* getTableList( SSqlObj* pSql ) { static SArray* getTableList( SSqlObj* pSql ) {
const char* p = strstr( pSql->sqlstr, " from " ); const char* p = strstr( pSql->sqlstr, " from " );
assert(p != NULL); // we are sure this is a 'select' statement assert(p != NULL); // we are sure this is a 'select' statement
...@@ -224,11 +224,11 @@ static SArray* getTableList( SSqlObj* pSql ) { ...@@ -224,11 +224,11 @@ static SArray* getTableList( SSqlObj* pSql ) {
SSqlObj* pNew = taos_query(pSql->pTscObj, sql); SSqlObj* pNew = taos_query(pSql->pTscObj, sql);
if (pNew == NULL) { if (pNew == NULL) {
tscError("0x%"PRIx64"failed to retrieve table id: cannot create new sql object.", pSql->self); tscError("0x%"PRIx64" failed to retrieve table id: cannot create new sql object.", pSql->self);
return NULL; return NULL;
} else if (taos_errno(pNew) != TSDB_CODE_SUCCESS) { } else if (taos_errno(pNew) != TSDB_CODE_SUCCESS) {
tscError("0x%"PRIx64"failed to retrieve table id,error: %s", pSql->self, tstrerror(taos_errno(pNew))); tscError("0x%"PRIx64" failed to retrieve table id,error: %s", pSql->self, tstrerror(taos_errno(pNew)));
return NULL; return NULL;
} }
......
src/inc/taoserror.h
浏览文件 @ 9a4ccd2c
...@@ -219,6 +219,7 @@ int32_t* taosGetErrno(); ...@@ -219,6 +219,7 @@ int32_t* taosGetErrno();
#define TSDB_CODE_VND_NO_WRITE_AUTH TAOS_DEF_ERROR_CODE(0, 0x0512) //"Database write operation denied") #define TSDB_CODE_VND_NO_WRITE_AUTH TAOS_DEF_ERROR_CODE(0, 0x0512) //"Database write operation denied")
#define TSDB_CODE_VND_IS_SYNCING TAOS_DEF_ERROR_CODE(0, 0x0513) //"Database is syncing") #define TSDB_CODE_VND_IS_SYNCING TAOS_DEF_ERROR_CODE(0, 0x0513) //"Database is syncing")
#define TSDB_CODE_VND_INVALID_TSDB_STATE TAOS_DEF_ERROR_CODE(0, 0x0514) //"Invalid tsdb state") #define TSDB_CODE_VND_INVALID_TSDB_STATE TAOS_DEF_ERROR_CODE(0, 0x0514) //"Invalid tsdb state")
#define TSDB_CODE_VND_IS_CLOSING TAOS_DEF_ERROR_CODE(0, 0x0515) //"Database is closing")
// tsdb // tsdb
#define TSDB_CODE_TDB_INVALID_TABLE_ID TAOS_DEF_ERROR_CODE(0, 0x0600) //"Invalid table ID") #define TSDB_CODE_TDB_INVALID_TABLE_ID TAOS_DEF_ERROR_CODE(0, 0x0600) //"Invalid table ID")
......
src/kit/taosdemo/taosdemo.c
浏览文件 @ 9a4ccd2c
...@@ -19,6 +19,7 @@ ...@@ -19,6 +19,7 @@
*/ */
#include <stdint.h> #include <stdint.h>
#include <taos.h>
#define _GNU_SOURCE #define _GNU_SOURCE
#define CURL_STATICLIB #define CURL_STATICLIB
...@@ -229,13 +230,13 @@ typedef struct SArguments_S { ...@@ -229,13 +230,13 @@ typedef struct SArguments_S {
uint32_t num_of_threads; uint32_t num_of_threads;
uint64_t insert_interval; uint64_t insert_interval;
int64_t query_times; int64_t query_times;
uint64_t interlace_rows; uint32_t interlace_rows;
uint64_t num_of_RPR; // num_of_records_per_req uint32_t num_of_RPR; // num_of_records_per_req
uint64_t max_sql_len; uint64_t max_sql_len;
int64_t num_of_tables; int64_t num_of_tables;
int64_t num_of_DPT; int64_t num_of_DPT;
int abort; int abort;
int disorderRatio; // 0: no disorder, >0: x% uint32_t disorderRatio; // 0: no disorder, >0: x%
int disorderRange; // ms or us by database precision int disorderRange; // ms or us by database precision
uint32_t method_of_delete; uint32_t method_of_delete;
char ** arg_list; char ** arg_list;
...@@ -258,12 +259,12 @@ typedef struct SSuperTable_S { ...@@ -258,12 +259,12 @@ typedef struct SSuperTable_S {
uint8_t autoCreateTable; // 0: create sub table, 1: auto create sub table uint8_t autoCreateTable; // 0: create sub table, 1: auto create sub table
char childTblPrefix[MAX_TB_NAME_SIZE]; char childTblPrefix[MAX_TB_NAME_SIZE];
char dataSource[MAX_TB_NAME_SIZE+1]; // rand_gen or sample char dataSource[MAX_TB_NAME_SIZE+1]; // rand_gen or sample
uint16_t insertMode; // 0: taosc, 1: rest, 2: stmt uint16_t iface; // 0: taosc, 1: rest, 2: stmt
int64_t childTblLimit; int64_t childTblLimit;
uint64_t childTblOffset; uint64_t childTblOffset;
// int multiThreadWriteOneTbl; // 0: no, 1: yes // int multiThreadWriteOneTbl; // 0: no, 1: yes
uint64_t interlaceRows; // uint32_t interlaceRows; //
int disorderRatio; // 0: no disorder, >0: x% int disorderRatio; // 0: no disorder, >0: x%
int disorderRange; // ms or us by database precision int disorderRange; // ms or us by database precision
uint64_t maxSqlLen; // uint64_t maxSqlLen; //
...@@ -469,6 +470,7 @@ typedef struct SThreadInfo_S { ...@@ -469,6 +470,7 @@ typedef struct SThreadInfo_S {
// seq of query or subscribe // seq of query or subscribe
uint64_t querySeq; // sequence number of sql command uint64_t querySeq; // sequence number of sql command
TAOS_SUB* tsub;
} threadInfo; } threadInfo;
...@@ -551,6 +553,8 @@ static void createChildTables(); ...@@ -551,6 +553,8 @@ static void createChildTables();
static int queryDbExec(TAOS *taos, char *command, QUERY_TYPE type, bool quiet); static int queryDbExec(TAOS *taos, char *command, QUERY_TYPE type, bool quiet);
static int postProceSql(char *host, struct sockaddr_in *pServAddr, static int postProceSql(char *host, struct sockaddr_in *pServAddr,
uint16_t port, char* sqlstr, threadInfo *pThreadInfo); uint16_t port, char* sqlstr, threadInfo *pThreadInfo);
static int64_t getTSRandTail(int64_t timeStampStep, int32_t seq,
int disorderRatio, int disorderRange);
/* ************ Global variables ************ */ /* ************ Global variables ************ */
...@@ -926,6 +930,7 @@ static void parse_args(int argc, char *argv[], SArguments *arguments) { ...@@ -926,6 +930,7 @@ static void parse_args(int argc, char *argv[], SArguments *arguments) {
&& strcasecmp(argv[i], "BIGINT") && strcasecmp(argv[i], "BIGINT")
&& strcasecmp(argv[i], "DOUBLE") && strcasecmp(argv[i], "DOUBLE")
&& strcasecmp(argv[i], "BINARY") && strcasecmp(argv[i], "BINARY")
&& strcasecmp(argv[i], "TIMESTAMP")
&& strcasecmp(argv[i], "NCHAR")) { && strcasecmp(argv[i], "NCHAR")) {
printHelp(); printHelp();
errorPrint("%s", "-b: Invalid data_type!\n"); errorPrint("%s", "-b: Invalid data_type!\n");
...@@ -947,6 +952,7 @@ static void parse_args(int argc, char *argv[], SArguments *arguments) { ...@@ -947,6 +952,7 @@ static void parse_args(int argc, char *argv[], SArguments *arguments) {
&& strcasecmp(token, "BIGINT") && strcasecmp(token, "BIGINT")
&& strcasecmp(token, "DOUBLE") && strcasecmp(token, "DOUBLE")
&& strcasecmp(token, "BINARY") && strcasecmp(token, "BINARY")
&& strcasecmp(token, "TIMESTAMP")
&& strcasecmp(token, "NCHAR")) { && strcasecmp(token, "NCHAR")) {
printHelp(); printHelp();
free(g_dupstr); free(g_dupstr);
...@@ -1070,7 +1076,7 @@ static void parse_args(int argc, char *argv[], SArguments *arguments) { ...@@ -1070,7 +1076,7 @@ static void parse_args(int argc, char *argv[], SArguments *arguments) {
} }
printf("# Insertion interval: %"PRIu64"\n", printf("# Insertion interval: %"PRIu64"\n",
arguments->insert_interval); arguments->insert_interval);
printf("# Number of records per req: %"PRIu64"\n", printf("# Number of records per req: %ud\n",
arguments->num_of_RPR); arguments->num_of_RPR);
printf("# Max SQL length: %"PRIu64"\n", printf("# Max SQL length: %"PRIu64"\n",
arguments->max_sql_len); arguments->max_sql_len);
...@@ -1129,11 +1135,11 @@ static int queryDbExec(TAOS *taos, char *command, QUERY_TYPE type, bool quiet) { ...@@ -1129,11 +1135,11 @@ static int queryDbExec(TAOS *taos, char *command, QUERY_TYPE type, bool quiet) {
} }
} }
verbosePrint("%s() LN%d - command: %s\n", __func__, __LINE__, command);
if (code != 0) { if (code != 0) {
if (!quiet) { if (!quiet) {
debugPrint("%s() LN%d - command: %s\n", __func__, __LINE__, command); errorPrint("Failed to execute %s, reason: %s\n",
errorPrint("Failed to execute %s, command length: %d, reason: %s\n", command, taos_errstr(res));
command, (int)strlen(command), taos_errstr(res));
} }
taos_free_result(res); taos_free_result(res);
//taos_close(taos); //taos_close(taos);
...@@ -1362,7 +1368,7 @@ static int printfInsertMeta() { ...@@ -1362,7 +1368,7 @@ static int printfInsertMeta() {
g_Dbs.threadCountByCreateTbl); g_Dbs.threadCountByCreateTbl);
printf("top insert interval: \033[33m%"PRIu64"\033[0m\n", printf("top insert interval: \033[33m%"PRIu64"\033[0m\n",
g_args.insert_interval); g_args.insert_interval);
printf("number of records per req: \033[33m%"PRIu64"\033[0m\n", printf("number of records per req: \033[33m%ud\033[0m\n",
g_args.num_of_RPR); g_args.num_of_RPR);
printf("max sql length: \033[33m%"PRIu64"\033[0m\n", printf("max sql length: \033[33m%"PRIu64"\033[0m\n",
g_args.max_sql_len); g_args.max_sql_len);
...@@ -1468,9 +1474,9 @@ static int printfInsertMeta() { ...@@ -1468,9 +1474,9 @@ static int printfInsertMeta() {
g_Dbs.db[i].superTbls[j].childTblPrefix); g_Dbs.db[i].superTbls[j].childTblPrefix);
printf(" dataSource: \033[33m%s\033[0m\n", printf(" dataSource: \033[33m%s\033[0m\n",
g_Dbs.db[i].superTbls[j].dataSource); g_Dbs.db[i].superTbls[j].dataSource);
printf(" insertMode: \033[33m%s\033[0m\n", printf(" iface: \033[33m%s\033[0m\n",
(g_Dbs.db[i].superTbls[j].insertMode==TAOSC_IFACE)?"taosc": (g_Dbs.db[i].superTbls[j].iface==TAOSC_IFACE)?"taosc":
(g_Dbs.db[i].superTbls[j].insertMode==REST_IFACE)?"rest":"stmt"); (g_Dbs.db[i].superTbls[j].iface==REST_IFACE)?"rest":"stmt");
if (g_Dbs.db[i].superTbls[j].childTblLimit > 0) { if (g_Dbs.db[i].superTbls[j].childTblLimit > 0) {
printf(" childTblLimit: \033[33m%"PRId64"\033[0m\n", printf(" childTblLimit: \033[33m%"PRId64"\033[0m\n",
g_Dbs.db[i].superTbls[j].childTblLimit); g_Dbs.db[i].superTbls[j].childTblLimit);
...@@ -1488,7 +1494,7 @@ static int printfInsertMeta() { ...@@ -1488,7 +1494,7 @@ static int printfInsertMeta() {
printf(" multiThreadWriteOneTbl: \033[33myes\033[0m\n"); printf(" multiThreadWriteOneTbl: \033[33myes\033[0m\n");
} }
*/ */
printf(" interlaceRows: \033[33m%"PRIu64"\033[0m\n", printf(" interlaceRows: \033[33m%ud\033[0m\n",
g_Dbs.db[i].superTbls[j].interlaceRows); g_Dbs.db[i].superTbls[j].interlaceRows);
if (g_Dbs.db[i].superTbls[j].interlaceRows > 0) { if (g_Dbs.db[i].superTbls[j].interlaceRows > 0) {
...@@ -1566,7 +1572,7 @@ static void printfInsertMetaToFile(FILE* fp) { ...@@ -1566,7 +1572,7 @@ static void printfInsertMetaToFile(FILE* fp) {
fprintf(fp, "resultFile: %s\n", g_Dbs.resultFile); fprintf(fp, "resultFile: %s\n", g_Dbs.resultFile);
fprintf(fp, "thread num of insert data: %d\n", g_Dbs.threadCount); fprintf(fp, "thread num of insert data: %d\n", g_Dbs.threadCount);
fprintf(fp, "thread num of create table: %d\n", g_Dbs.threadCountByCreateTbl); fprintf(fp, "thread num of create table: %d\n", g_Dbs.threadCountByCreateTbl);
fprintf(fp, "number of records per req: %"PRIu64"\n", g_args.num_of_RPR); fprintf(fp, "number of records per req: %ud\n", g_args.num_of_RPR);
fprintf(fp, "max sql length: %"PRIu64"\n", g_args.max_sql_len); fprintf(fp, "max sql length: %"PRIu64"\n", g_args.max_sql_len);
fprintf(fp, "database count: %d\n", g_Dbs.dbCount); fprintf(fp, "database count: %d\n", g_Dbs.dbCount);
...@@ -1658,12 +1664,12 @@ static void printfInsertMetaToFile(FILE* fp) { ...@@ -1658,12 +1664,12 @@ static void printfInsertMetaToFile(FILE* fp) {
g_Dbs.db[i].superTbls[j].childTblPrefix); g_Dbs.db[i].superTbls[j].childTblPrefix);
fprintf(fp, " dataSource: %s\n", fprintf(fp, " dataSource: %s\n",
g_Dbs.db[i].superTbls[j].dataSource); g_Dbs.db[i].superTbls[j].dataSource);
fprintf(fp, " insertMode: %s\n", fprintf(fp, " iface: %s\n",
(g_Dbs.db[i].superTbls[j].insertMode==TAOSC_IFACE)?"taosc": (g_Dbs.db[i].superTbls[j].iface==TAOSC_IFACE)?"taosc":
(g_Dbs.db[i].superTbls[j].insertMode==REST_IFACE)?"rest":"stmt"); (g_Dbs.db[i].superTbls[j].iface==REST_IFACE)?"rest":"stmt");
fprintf(fp, " insertRows: %"PRId64"\n", fprintf(fp, " insertRows: %"PRId64"\n",
g_Dbs.db[i].superTbls[j].insertRows); g_Dbs.db[i].superTbls[j].insertRows);
fprintf(fp, " interlace rows: %"PRIu64"\n", fprintf(fp, " interlace rows: %ud\n",
g_Dbs.db[i].superTbls[j].interlaceRows); g_Dbs.db[i].superTbls[j].interlaceRows);
if (g_Dbs.db[i].superTbls[j].interlaceRows > 0) { if (g_Dbs.db[i].superTbls[j].interlaceRows > 0) {
fprintf(fp, " stable insert interval: %"PRIu64"\n", fprintf(fp, " stable insert interval: %"PRIu64"\n",
...@@ -1676,7 +1682,7 @@ static void printfInsertMetaToFile(FILE* fp) { ...@@ -1676,7 +1682,7 @@ static void printfInsertMetaToFile(FILE* fp) {
fprintf(fp, " multiThreadWriteOneTbl: yes\n"); fprintf(fp, " multiThreadWriteOneTbl: yes\n");
} }
*/ */
fprintf(fp, " interlaceRows: %"PRIu64"\n", fprintf(fp, " interlaceRows: %ud\n",
g_Dbs.db[i].superTbls[j].interlaceRows); g_Dbs.db[i].superTbls[j].interlaceRows);
fprintf(fp, " disorderRange: %d\n", fprintf(fp, " disorderRange: %d\n",
g_Dbs.db[i].superTbls[j].disorderRange); g_Dbs.db[i].superTbls[j].disorderRange);
...@@ -2729,8 +2735,6 @@ static int createSuperTable( ...@@ -2729,8 +2735,6 @@ static int createSuperTable(
snprintf(command, BUFFER_SIZE, snprintf(command, BUFFER_SIZE,
"create table if not exists %s.%s (ts timestamp%s) tags %s", "create table if not exists %s.%s (ts timestamp%s) tags %s",
dbName, superTbl->sTblName, cols, tags); dbName, superTbl->sTblName, cols, tags);
verbosePrint("%s() LN%d: %s\n", __func__, __LINE__, command);
if (0 != queryDbExec(taos, command, NO_INSERT_TYPE, false)) { if (0 != queryDbExec(taos, command, NO_INSERT_TYPE, false)) {
errorPrint( "create supertable %s failed!\n\n", errorPrint( "create supertable %s failed!\n\n",
superTbl->sTblName); superTbl->sTblName);
...@@ -2753,7 +2757,6 @@ static int createDatabasesAndStables() { ...@@ -2753,7 +2757,6 @@ static int createDatabasesAndStables() {
for (int i = 0; i < g_Dbs.dbCount; i++) { for (int i = 0; i < g_Dbs.dbCount; i++) {
if (g_Dbs.db[i].drop) { if (g_Dbs.db[i].drop) {
sprintf(command, "drop database if exists %s;", g_Dbs.db[i].dbName); sprintf(command, "drop database if exists %s;", g_Dbs.db[i].dbName);
verbosePrint("%s() %d command: %s\n", __func__, __LINE__, command);
if (0 != queryDbExec(taos, command, NO_INSERT_TYPE, false)) { if (0 != queryDbExec(taos, command, NO_INSERT_TYPE, false)) {
taos_close(taos); taos_close(taos);
return -1; return -1;
...@@ -2826,7 +2829,6 @@ static int createDatabasesAndStables() { ...@@ -2826,7 +2829,6 @@ static int createDatabasesAndStables() {
" precision \'%s\';", g_Dbs.db[i].dbCfg.precision); " precision \'%s\';", g_Dbs.db[i].dbCfg.precision);
} }
debugPrint("%s() %d command: %s\n", __func__, __LINE__, command);
if (0 != queryDbExec(taos, command, NO_INSERT_TYPE, false)) { if (0 != queryDbExec(taos, command, NO_INSERT_TYPE, false)) {
taos_close(taos); taos_close(taos);
errorPrint( "\ncreate database %s failed!\n\n", g_Dbs.db[i].dbName); errorPrint( "\ncreate database %s failed!\n\n", g_Dbs.db[i].dbName);
...@@ -2843,8 +2845,6 @@ static int createDatabasesAndStables() { ...@@ -2843,8 +2845,6 @@ static int createDatabasesAndStables() {
for (uint64_t j = 0; j < g_Dbs.db[i].superTblCount; j++) { for (uint64_t j = 0; j < g_Dbs.db[i].superTblCount; j++) {
sprintf(command, "describe %s.%s;", g_Dbs.db[i].dbName, sprintf(command, "describe %s.%s;", g_Dbs.db[i].dbName,
g_Dbs.db[i].superTbls[j].sTblName); g_Dbs.db[i].superTbls[j].sTblName);
verbosePrint("%s() %d command: %s\n", __func__, __LINE__, command);
ret = queryDbExec(taos, command, NO_INSERT_TYPE, true); ret = queryDbExec(taos, command, NO_INSERT_TYPE, true);
if ((ret != 0) || (g_Dbs.db[i].drop)) { if ((ret != 0) || (g_Dbs.db[i].drop)) {
...@@ -2948,7 +2948,6 @@ static void* createTable(void *sarg) ...@@ -2948,7 +2948,6 @@ static void* createTable(void *sarg)
} }
len = 0; len = 0;
verbosePrint("%s() LN%d %s\n", __func__, __LINE__, buffer);
if (0 != queryDbExec(pThreadInfo->taos, buffer, NO_INSERT_TYPE, false)){ if (0 != queryDbExec(pThreadInfo->taos, buffer, NO_INSERT_TYPE, false)){
errorPrint( "queryDbExec() failed. buffer:\n%s\n", buffer); errorPrint( "queryDbExec() failed. buffer:\n%s\n", buffer);
free(buffer); free(buffer);
...@@ -2964,7 +2963,6 @@ static void* createTable(void *sarg) ...@@ -2964,7 +2963,6 @@ static void* createTable(void *sarg)
} }
if (0 != len) { if (0 != len) {
verbosePrint("%s() %d buffer: %s\n", __func__, __LINE__, buffer);
if (0 != queryDbExec(pThreadInfo->taos, buffer, NO_INSERT_TYPE, false)) { if (0 != queryDbExec(pThreadInfo->taos, buffer, NO_INSERT_TYPE, false)) {
errorPrint( "queryDbExec() failed. buffer:\n%s\n", buffer); errorPrint( "queryDbExec() failed. buffer:\n%s\n", buffer);
} }
...@@ -2975,7 +2973,7 @@ static void* createTable(void *sarg) ...@@ -2975,7 +2973,7 @@ static void* createTable(void *sarg)
} }
static int startMultiThreadCreateChildTable( static int startMultiThreadCreateChildTable(
char* cols, int threads, uint64_t startFrom, int64_t ntables, char* cols, int threads, uint64_t tableFrom, int64_t ntables,
char* db_name, SSuperTable* superTblInfo) { char* db_name, SSuperTable* superTblInfo) {
pthread_t *pids = malloc(threads * sizeof(pthread_t)); pthread_t *pids = malloc(threads * sizeof(pthread_t));
...@@ -3019,10 +3017,10 @@ static int startMultiThreadCreateChildTable( ...@@ -3019,10 +3017,10 @@ static int startMultiThreadCreateChildTable(
return -1; return -1;
} }
pThreadInfo->start_table_from = startFrom; pThreadInfo->start_table_from = tableFrom;
pThreadInfo->ntables = i<b?a+1:a; pThreadInfo->ntables = i<b?a+1:a;
pThreadInfo->end_table_to = i < b ? startFrom + a : startFrom + a - 1; pThreadInfo->end_table_to = i < b ? tableFrom + a : tableFrom + a - 1;
startFrom = pThreadInfo->end_table_to + 1; tableFrom = pThreadInfo->end_table_to + 1;
pThreadInfo->use_metric = true; pThreadInfo->use_metric = true;
pThreadInfo->cols = cols; pThreadInfo->cols = cols;
pThreadInfo->minDelay = UINT64_MAX; pThreadInfo->minDelay = UINT64_MAX;
...@@ -3060,15 +3058,15 @@ static void createChildTables() { ...@@ -3060,15 +3058,15 @@ static void createChildTables() {
verbosePrint("%s() LN%d: %s\n", __func__, __LINE__, verbosePrint("%s() LN%d: %s\n", __func__, __LINE__,
g_Dbs.db[i].superTbls[j].colsOfCreateChildTable); g_Dbs.db[i].superTbls[j].colsOfCreateChildTable);
uint64_t startFrom = 0; uint64_t tableFrom = 0;
g_totalChildTables += g_Dbs.db[i].superTbls[j].childTblCount; g_totalChildTables += g_Dbs.db[i].superTbls[j].childTblCount;
verbosePrint("%s() LN%d: create %"PRId64" child tables from %"PRIu64"\n", verbosePrint("%s() LN%d: create %"PRId64" child tables from %"PRIu64"\n",
__func__, __LINE__, g_totalChildTables, startFrom); __func__, __LINE__, g_totalChildTables, tableFrom);
startMultiThreadCreateChildTable( startMultiThreadCreateChildTable(
g_Dbs.db[i].superTbls[j].colsOfCreateChildTable, g_Dbs.db[i].superTbls[j].colsOfCreateChildTable,
g_Dbs.threadCountByCreateTbl, g_Dbs.threadCountByCreateTbl,
startFrom, tableFrom,
g_Dbs.db[i].superTbls[j].childTblCount, g_Dbs.db[i].superTbls[j].childTblCount,
g_Dbs.db[i].dbName, &(g_Dbs.db[i].superTbls[j])); g_Dbs.db[i].dbName, &(g_Dbs.db[i].superTbls[j]));
} }
...@@ -3563,9 +3561,9 @@ static bool getMetaFromInsertJsonFile(cJSON* root) { ...@@ -3563,9 +3561,9 @@ static bool getMetaFromInsertJsonFile(cJSON* root) {
// rows per table need be less than insert batch // rows per table need be less than insert batch
if (g_args.interlace_rows > g_args.num_of_RPR) { if (g_args.interlace_rows > g_args.num_of_RPR) {
printf("NOTICE: interlace rows value %"PRIu64" > num_of_records_per_req %"PRIu64"\n\n", printf("NOTICE: interlace rows value %ud > num_of_records_per_req %ud\n\n",
g_args.interlace_rows, g_args.num_of_RPR); g_args.interlace_rows, g_args.num_of_RPR);
printf(" interlace rows value will be set to num_of_records_per_req %"PRIu64"\n\n", printf(" interlace rows value will be set to num_of_records_per_req %ud\n\n",
g_args.num_of_RPR); g_args.num_of_RPR);
prompt(); prompt();
g_args.interlace_rows = g_args.num_of_RPR; g_args.interlace_rows = g_args.num_of_RPR;
...@@ -3880,22 +3878,22 @@ static bool getMetaFromInsertJsonFile(cJSON* root) { ...@@ -3880,22 +3878,22 @@ static bool getMetaFromInsertJsonFile(cJSON* root) {
goto PARSE_OVER; goto PARSE_OVER;
} }
cJSON *insertMode = cJSON_GetObjectItem(stbInfo, "insert_mode"); // taosc , rest, stmt cJSON *stbIface = cJSON_GetObjectItem(stbInfo, "insert_mode"); // taosc , rest, stmt
if (insertMode && insertMode->type == cJSON_String if (stbIface && stbIface->type == cJSON_String
&& insertMode->valuestring != NULL) { && stbIface->valuestring != NULL) {
if (0 == strcasecmp(insertMode->valuestring, "taosc")) { if (0 == strcasecmp(stbIface->valuestring, "taosc")) {
g_Dbs.db[i].superTbls[j].insertMode = TAOSC_IFACE; g_Dbs.db[i].superTbls[j].iface= TAOSC_IFACE;
} else if (0 == strcasecmp(insertMode->valuestring, "rest")) { } else if (0 == strcasecmp(stbIface->valuestring, "rest")) {
g_Dbs.db[i].superTbls[j].insertMode = REST_IFACE; g_Dbs.db[i].superTbls[j].iface= REST_IFACE;
} else if (0 == strcasecmp(insertMode->valuestring, "stmt")) { } else if (0 == strcasecmp(stbIface->valuestring, "stmt")) {
g_Dbs.db[i].superTbls[j].insertMode = STMT_IFACE; g_Dbs.db[i].superTbls[j].iface= STMT_IFACE;
} else { } else {
errorPrint("%s() LN%d, failed to read json, insert_mode %s not recognized\n", errorPrint("%s() LN%d, failed to read json, insert_mode %s not recognized\n",
__func__, __LINE__, insertMode->valuestring); __func__, __LINE__, stbIface->valuestring);
goto PARSE_OVER; goto PARSE_OVER;
} }
} else if (!insertMode) { } else if (!stbIface) {
g_Dbs.db[i].superTbls[j].insertMode = TAOSC_IFACE; g_Dbs.db[i].superTbls[j].iface = TAOSC_IFACE;
} else { } else {
errorPrint("%s", "failed to read json, insert_mode not found\n"); errorPrint("%s", "failed to read json, insert_mode not found\n");
goto PARSE_OVER; goto PARSE_OVER;
...@@ -4032,9 +4030,10 @@ static bool getMetaFromInsertJsonFile(cJSON* root) { ...@@ -4032,9 +4030,10 @@ static bool getMetaFromInsertJsonFile(cJSON* root) {
g_Dbs.db[i].superTbls[j].interlaceRows = stbInterlaceRows->valueint; g_Dbs.db[i].superTbls[j].interlaceRows = stbInterlaceRows->valueint;
// rows per table need be less than insert batch // rows per table need be less than insert batch
if (g_Dbs.db[i].superTbls[j].interlaceRows > g_args.num_of_RPR) { if (g_Dbs.db[i].superTbls[j].interlaceRows > g_args.num_of_RPR) {
printf("NOTICE: db[%d].superTbl[%d]'s interlace rows value %"PRIu64" > num_of_records_per_req %"PRIu64"\n\n", printf("NOTICE: db[%d].superTbl[%d]'s interlace rows value %ud > num_of_records_per_req %ud\n\n",
i, j, g_Dbs.db[i].superTbls[j].interlaceRows, g_args.num_of_RPR); i, j, g_Dbs.db[i].superTbls[j].interlaceRows,
printf(" interlace rows value will be set to num_of_records_per_req %"PRIu64"\n\n", g_args.num_of_RPR);
printf(" interlace rows value will be set to num_of_records_per_req %ud\n\n",
g_args.num_of_RPR); g_args.num_of_RPR);
prompt(); prompt();
g_Dbs.db[i].superTbls[j].interlaceRows = g_args.num_of_RPR; g_Dbs.db[i].superTbls[j].interlaceRows = g_args.num_of_RPR;
...@@ -4695,8 +4694,8 @@ static int getRowDataFromSample( ...@@ -4695,8 +4694,8 @@ static int getRowDataFromSample(
static int64_t generateStbRowData( static int64_t generateStbRowData(
SSuperTable* stbInfo, SSuperTable* stbInfo,
char* recBuf, int64_t timestamp char* recBuf, int64_t timestamp)
) { {
int64_t dataLen = 0; int64_t dataLen = 0;
char *pstr = recBuf; char *pstr = recBuf;
int64_t maxLen = MAX_DATA_SIZE; int64_t maxLen = MAX_DATA_SIZE;
...@@ -4724,23 +4723,23 @@ static int64_t generateStbRowData( ...@@ -4724,23 +4723,23 @@ static int64_t generateStbRowData(
dataLen += snprintf(pstr + dataLen, maxLen - dataLen, "\'%s\',", buf); dataLen += snprintf(pstr + dataLen, maxLen - dataLen, "\'%s\',", buf);
tmfree(buf); tmfree(buf);
} else if (0 == strncasecmp(stbInfo->columns[i].dataType, } else if (0 == strncasecmp(stbInfo->columns[i].dataType,
"INT", 3)) { "INT", strlen("INT"))) {
dataLen += snprintf(pstr + dataLen, maxLen - dataLen, dataLen += snprintf(pstr + dataLen, maxLen - dataLen,
"%d,", rand_int()); "%d,", rand_int());
} else if (0 == strncasecmp(stbInfo->columns[i].dataType, } else if (0 == strncasecmp(stbInfo->columns[i].dataType,
"BIGINT", 6)) { "BIGINT", strlen("BIGINT"))) {
dataLen += snprintf(pstr + dataLen, maxLen - dataLen, dataLen += snprintf(pstr + dataLen, maxLen - dataLen,
"%"PRId64",", rand_bigint()); "%"PRId64",", rand_bigint());
} else if (0 == strncasecmp(stbInfo->columns[i].dataType, } else if (0 == strncasecmp(stbInfo->columns[i].dataType,
"FLOAT", 5)) { "FLOAT", strlen("FLOAT"))) {
dataLen += snprintf(pstr + dataLen, maxLen - dataLen, dataLen += snprintf(pstr + dataLen, maxLen - dataLen,
"%f,", rand_float()); "%f,", rand_float());
} else if (0 == strncasecmp(stbInfo->columns[i].dataType, } else if (0 == strncasecmp(stbInfo->columns[i].dataType,
"DOUBLE", 6)) { "DOUBLE", strlen("DOUBLE"))) {
dataLen += snprintf(pstr + dataLen, maxLen - dataLen, dataLen += snprintf(pstr + dataLen, maxLen - dataLen,
"%f,", rand_double()); "%f,", rand_double());
} else if (0 == strncasecmp(stbInfo->columns[i].dataType, } else if (0 == strncasecmp(stbInfo->columns[i].dataType,
"SMALLINT", 8)) { "SMALLINT", strlen("SMALLINT"))) {
dataLen += snprintf(pstr + dataLen, maxLen - dataLen, dataLen += snprintf(pstr + dataLen, maxLen - dataLen,
"%d,", rand_smallint()); "%d,", rand_smallint());
} else if (0 == strncasecmp(stbInfo->columns[i].dataType, } else if (0 == strncasecmp(stbInfo->columns[i].dataType,
...@@ -4796,6 +4795,8 @@ static int64_t generateData(char *recBuf, char **data_type, ...@@ -4796,6 +4795,8 @@ static int64_t generateData(char *recBuf, char **data_type,
pstr += sprintf(pstr, ",%d", rand_int()); pstr += sprintf(pstr, ",%d", rand_int());
} else if (strcasecmp(data_type[i % c], "BIGINT") == 0) { } else if (strcasecmp(data_type[i % c], "BIGINT") == 0) {
pstr += sprintf(pstr, ",%" PRId64, rand_bigint()); pstr += sprintf(pstr, ",%" PRId64, rand_bigint());
} else if (strcasecmp(data_type[i % c], "TIMESTAMP") == 0) {
pstr += sprintf(pstr, ",%" PRId64, rand_bigint());
} else if (strcasecmp(data_type[i % c], "FLOAT") == 0) { } else if (strcasecmp(data_type[i % c], "FLOAT") == 0) {
pstr += sprintf(pstr, ",%10.4f", rand_float()); pstr += sprintf(pstr, ",%10.4f", rand_float());
} else if (strcasecmp(data_type[i % c], "DOUBLE") == 0) { } else if (strcasecmp(data_type[i % c], "DOUBLE") == 0) {
...@@ -4864,11 +4865,11 @@ static int64_t execInsert(threadInfo *pThreadInfo, uint64_t k) ...@@ -4864,11 +4865,11 @@ static int64_t execInsert(threadInfo *pThreadInfo, uint64_t k)
verbosePrint("[%d] %s() LN%d %s\n", pThreadInfo->threadID, verbosePrint("[%d] %s() LN%d %s\n", pThreadInfo->threadID,
__func__, __LINE__, pThreadInfo->buffer); __func__, __LINE__, pThreadInfo->buffer);
if (superTblInfo) { if (superTblInfo) {
if (superTblInfo->insertMode == TAOSC_IFACE) { if (superTblInfo->iface == TAOSC_IFACE) {
affectedRows = queryDbExec( affectedRows = queryDbExec(
pThreadInfo->taos, pThreadInfo->taos,
pThreadInfo->buffer, INSERT_TYPE, false); pThreadInfo->buffer, INSERT_TYPE, false);
} else if (superTblInfo->insertMode == REST_IFACE) { } else if (superTblInfo->iface == REST_IFACE) {
if (0 != postProceSql(g_Dbs.host, &g_Dbs.serv_addr, g_Dbs.port, if (0 != postProceSql(g_Dbs.host, &g_Dbs.serv_addr, g_Dbs.port,
pThreadInfo->buffer, NULL /* not set result file */)) { pThreadInfo->buffer, NULL /* not set result file */)) {
affectedRows = -1; affectedRows = -1;
...@@ -4877,7 +4878,7 @@ static int64_t execInsert(threadInfo *pThreadInfo, uint64_t k) ...@@ -4877,7 +4878,7 @@ static int64_t execInsert(threadInfo *pThreadInfo, uint64_t k)
} else { } else {
affectedRows = k; affectedRows = k;
} }
} else if (superTblInfo->insertMode == STMT_IFACE) { } else if (superTblInfo->iface == STMT_IFACE) {
debugPrint("%s() LN%d, stmt=%p", __func__, __LINE__, pThreadInfo->stmt); debugPrint("%s() LN%d, stmt=%p", __func__, __LINE__, pThreadInfo->stmt);
if (0 != taos_stmt_execute(pThreadInfo->stmt)) { if (0 != taos_stmt_execute(pThreadInfo->stmt)) {
errorPrint("%s() LN%d, failied to execute insert statement\n", errorPrint("%s() LN%d, failied to execute insert statement\n",
...@@ -4888,7 +4889,7 @@ static int64_t execInsert(threadInfo *pThreadInfo, uint64_t k) ...@@ -4888,7 +4889,7 @@ static int64_t execInsert(threadInfo *pThreadInfo, uint64_t k)
affectedRows = k; affectedRows = k;
} else { } else {
errorPrint("%s() LN%d: unknown insert mode: %d\n", errorPrint("%s() LN%d: unknown insert mode: %d\n",
__func__, __LINE__, superTblInfo->insertMode); __func__, __LINE__, superTblInfo->iface);
affectedRows = 0; affectedRows = 0;
} }
} else { } else {
...@@ -4924,15 +4925,15 @@ static void getTableName(char *pTblName, ...@@ -4924,15 +4925,15 @@ static void getTableName(char *pTblName,
} }
static int64_t generateDataTailWithoutStb( static int64_t generateDataTailWithoutStb(
uint64_t batch, char* buffer, uint32_t batch, char* buffer,
int64_t remainderBufLen, int64_t insertRows, int64_t remainderBufLen, int64_t insertRows,
uint64_t startFrom, int64_t startTime, uint64_t recordFrom, int64_t startTime,
/* int64_t *pSamplePos, */int64_t *dataLen) { /* int64_t *pSamplePos, */int64_t *dataLen) {
uint64_t len = 0; uint64_t len = 0;
char *pstr = buffer; char *pstr = buffer;
verbosePrint("%s() LN%d batch=%"PRIu64"\n", __func__, __LINE__, batch); verbosePrint("%s() LN%d batch=%d\n", __func__, __LINE__, batch);
int64_t k = 0; int64_t k = 0;
for (k = 0; k < batch;) { for (k = 0; k < batch;) {
...@@ -4944,22 +4945,11 @@ static int64_t generateDataTailWithoutStb( ...@@ -4944,22 +4945,11 @@ static int64_t generateDataTailWithoutStb(
char **data_type = g_args.datatype; char **data_type = g_args.datatype;
int lenOfBinary = g_args.len_of_binary; int lenOfBinary = g_args.len_of_binary;
int64_t randTail = DEFAULT_TIMESTAMP_STEP * k;
if (g_args.disorderRatio != 0) {
int rand_num = taosRandom() % 100;
if (rand_num < g_args.disorderRatio) {
randTail = (randTail +
(taosRandom() % g_args.disorderRange + 1)) * (-1);
debugPrint("rand data generated, back %"PRId64"\n", randTail);
}
} else {
randTail = DEFAULT_TIMESTAMP_STEP * k;
}
retLen = generateData(data, data_type, retLen = generateData(data, data_type,
startTime + randTail, startTime + getTSRandTail(
(int64_t)DEFAULT_TIMESTAMP_STEP, k,
g_args.disorderRatio,
g_args.disorderRange),
lenOfBinary); lenOfBinary);
if (len > remainderBufLen) if (len > remainderBufLen)
...@@ -4973,9 +4963,9 @@ static int64_t generateDataTailWithoutStb( ...@@ -4973,9 +4963,9 @@ static int64_t generateDataTailWithoutStb(
verbosePrint("%s() LN%d len=%"PRIu64" k=%"PRIu64" \nbuffer=%s\n", verbosePrint("%s() LN%d len=%"PRIu64" k=%"PRIu64" \nbuffer=%s\n",
__func__, __LINE__, len, k, buffer); __func__, __LINE__, len, k, buffer);
startFrom ++; recordFrom ++;
if (startFrom >= insertRows) { if (recordFrom >= insertRows) {
break; break;
} }
} }
...@@ -4984,47 +4974,61 @@ static int64_t generateDataTailWithoutStb( ...@@ -4984,47 +4974,61 @@ static int64_t generateDataTailWithoutStb(
return k; return k;
} }
static int64_t generateStbDataTail( static int64_t getTSRandTail(int64_t timeStampStep, int32_t seq,
int disorderRatio, int disorderRange)
{
int64_t randTail = timeStampStep * seq;
if (disorderRatio > 0) {
int rand_num = taosRandom() % 100;
if(rand_num < disorderRatio) {
randTail = (randTail +
(taosRandom() % disorderRange + 1)) * (-1);
debugPrint("rand data generated, back %"PRId64"\n", randTail);
}
}
return randTail;
}
static int32_t generateStbDataTail(
SSuperTable* superTblInfo, SSuperTable* superTblInfo,
uint64_t batch, char* buffer, uint32_t batch, char* buffer,
int64_t remainderBufLen, int64_t insertRows, int64_t remainderBufLen, int64_t insertRows,
uint64_t startFrom, int64_t startTime, uint64_t recordFrom, int64_t startTime,
int64_t *pSamplePos, int64_t *dataLen) { int64_t *pSamplePos, int64_t *dataLen) {
uint64_t len = 0; uint64_t len = 0;
char *pstr = buffer; char *pstr = buffer;
verbosePrint("%s() LN%d batch=%"PRIu64"\n", __func__, __LINE__, batch); bool tsRand;
if (0 == strncasecmp(superTblInfo->dataSource, "rand", strlen("rand"))) {
tsRand = true;
} else {
tsRand = false;
}
verbosePrint("%s() LN%d batch=%ud\n", __func__, __LINE__, batch);
int64_t k = 0; int32_t k = 0;
for (k = 0; k < batch;) { for (k = 0; k < batch;) {
char data[MAX_DATA_SIZE]; char data[MAX_DATA_SIZE];
memset(data, 0, MAX_DATA_SIZE); memset(data, 0, MAX_DATA_SIZE);
int64_t retLen = 0; int64_t retLen = 0;
if (0 == strncasecmp(superTblInfo->dataSource, if (tsRand) {
"sample", strlen("sample"))) { retLen = generateStbRowData(superTblInfo, data,
startTime + getTSRandTail(
superTblInfo->timeStampStep, k,
superTblInfo->disorderRatio,
superTblInfo->disorderRange)
);
} else {
retLen = getRowDataFromSample( retLen = getRowDataFromSample(
data, data,
remainderBufLen, remainderBufLen,
startTime + superTblInfo->timeStampStep * k, startTime + superTblInfo->timeStampStep * k,
superTblInfo, superTblInfo,
pSamplePos); pSamplePos);
} else if (0 == strncasecmp(superTblInfo->dataSource,
"rand", strlen("rand"))) {
int64_t randTail = superTblInfo->timeStampStep * k;
if (superTblInfo->disorderRatio > 0) {
int rand_num = taosRandom() % 100;
if(rand_num < superTblInfo->disorderRatio) {
randTail = (randTail +
(taosRandom() % superTblInfo->disorderRange + 1)) * (-1);
debugPrint("rand data generated, back %"PRId64"\n", randTail);
}
}
int64_t d = startTime + randTail;
retLen = generateStbRowData(superTblInfo, data, d);
} }
if (retLen > remainderBufLen) { if (retLen > remainderBufLen) {
...@@ -5036,12 +5040,12 @@ static int64_t generateStbDataTail( ...@@ -5036,12 +5040,12 @@ static int64_t generateStbDataTail(
len += retLen; len += retLen;
remainderBufLen -= retLen; remainderBufLen -= retLen;
verbosePrint("%s() LN%d len=%"PRIu64" k=%"PRIu64" \nbuffer=%s\n", verbosePrint("%s() LN%d len=%"PRIu64" k=%ud \nbuffer=%s\n",
__func__, __LINE__, len, k, buffer); __func__, __LINE__, len, k, buffer);
startFrom ++; recordFrom ++;
if (startFrom >= insertRows) { if (recordFrom >= insertRows) {
break; break;
} }
} }
...@@ -5133,9 +5137,11 @@ static int generateStbSQLHead( ...@@ -5133,9 +5137,11 @@ static int generateStbSQLHead(
return len; return len;
} }
static int64_t generateStbInterlaceData( static int32_t generateStbInterlaceData(
SSuperTable *superTblInfo, SSuperTable *superTblInfo,
char *tableName, uint64_t batchPerTbl, uint64_t i, uint64_t batchPerTblTimes, char *tableName, uint32_t batchPerTbl,
uint64_t i,
uint32_t batchPerTblTimes,
uint64_t tableSeq, uint64_t tableSeq,
threadInfo *pThreadInfo, char *buffer, threadInfo *pThreadInfo, char *buffer,
int64_t insertRows, int64_t insertRows,
...@@ -5162,7 +5168,7 @@ static int64_t generateStbInterlaceData( ...@@ -5162,7 +5168,7 @@ static int64_t generateStbInterlaceData(
int64_t dataLen = 0; int64_t dataLen = 0;
verbosePrint("[%d] %s() LN%d i=%"PRIu64" batchPerTblTimes=%"PRIu64" batchPerTbl = %"PRIu64"\n", verbosePrint("[%d] %s() LN%d i=%"PRIu64" batchPerTblTimes=%ud batchPerTbl = %ud\n",
pThreadInfo->threadID, __func__, __LINE__, pThreadInfo->threadID, __func__, __LINE__,
i, batchPerTblTimes, batchPerTbl); i, batchPerTblTimes, batchPerTbl);
...@@ -5170,7 +5176,7 @@ static int64_t generateStbInterlaceData( ...@@ -5170,7 +5176,7 @@ static int64_t generateStbInterlaceData(
startTime = taosGetTimestamp(pThreadInfo->time_precision); startTime = taosGetTimestamp(pThreadInfo->time_precision);
} }
int64_t k = generateStbDataTail( int32_t k = generateStbDataTail(
superTblInfo, superTblInfo,
batchPerTbl, pstr, *pRemainderBufLen, insertRows, 0, batchPerTbl, pstr, *pRemainderBufLen, insertRows, 0,
startTime, startTime,
...@@ -5180,7 +5186,7 @@ static int64_t generateStbInterlaceData( ...@@ -5180,7 +5186,7 @@ static int64_t generateStbInterlaceData(
pstr += dataLen; pstr += dataLen;
*pRemainderBufLen -= dataLen; *pRemainderBufLen -= dataLen;
} else { } else {
debugPrint("%s() LN%d, generated data tail: %"PRIu64", not equal batch per table: %"PRIu64"\n", debugPrint("%s() LN%d, generated data tail: %ud, not equal batch per table: %ud\n",
__func__, __LINE__, k, batchPerTbl); __func__, __LINE__, k, batchPerTbl);
pstr -= headLen; pstr -= headLen;
pstr[0] = '\0'; pstr[0] = '\0';
...@@ -5191,7 +5197,7 @@ static int64_t generateStbInterlaceData( ...@@ -5191,7 +5197,7 @@ static int64_t generateStbInterlaceData(
} }
static int64_t generateInterlaceDataWithoutStb( static int64_t generateInterlaceDataWithoutStb(
char *tableName, uint64_t batchPerTbl, char *tableName, uint32_t batchPerTbl,
uint64_t tableSeq, uint64_t tableSeq,
char *dbName, char *buffer, char *dbName, char *buffer,
int64_t insertRows, int64_t insertRows,
...@@ -5223,7 +5229,7 @@ static int64_t generateInterlaceDataWithoutStb( ...@@ -5223,7 +5229,7 @@ static int64_t generateInterlaceDataWithoutStb(
pstr += dataLen; pstr += dataLen;
*pRemainderBufLen -= dataLen; *pRemainderBufLen -= dataLen;
} else { } else {
debugPrint("%s() LN%d, generated data tail: %"PRIu64", not equal batch per table: %"PRIu64"\n", debugPrint("%s() LN%d, generated data tail: %"PRIu64", not equal batch per table: %ud\n",
__func__, __LINE__, k, batchPerTbl); __func__, __LINE__, k, batchPerTbl);
pstr -= headLen; pstr -= headLen;
pstr[0] = '\0'; pstr[0] = '\0';
...@@ -5233,13 +5239,222 @@ static int64_t generateInterlaceDataWithoutStb( ...@@ -5233,13 +5239,222 @@ static int64_t generateInterlaceDataWithoutStb(
return k; return k;
} }
static int64_t generateStbProgressiveData( static int32_t prepareStbStmt(SSuperTable *stbInfo,
TAOS_STMT *stmt,
char *tableName, uint32_t batch, uint64_t insertRows,
uint64_t recordFrom,
int64_t startTime, char *buffer)
{
uint32_t k;
int ret;
char *pstr = buffer;
pstr += sprintf(pstr, "INSERT INTO %s values(?", tableName);
for (int i = 0; i < stbInfo->columnCount; i++) {
pstr += sprintf(pstr, ",?");
}
pstr += sprintf(pstr, ")");
ret = taos_stmt_prepare(stmt, buffer, 0);
if (ret != 0){
errorPrint("failed to execute taos_stmt_prepare. return 0x%x. reason: %s\n",
ret, taos_errstr(NULL));
return ret;
}
char *bindArray = malloc(sizeof(TAOS_BIND) * (stbInfo->columnCount + 1));
if (bindArray == NULL) {
errorPrint("Failed to allocate %d bind params\n", batch);
return -1;
}
bool tsRand;
if (0 == strncasecmp(stbInfo->dataSource, "rand", strlen("rand"))) {
tsRand = true;
} else {
tsRand = false;
}
for (k = 0; k < batch;) {
/* columnCount + 1 (ts) */
char data[MAX_DATA_SIZE];
memset(data, 0, MAX_DATA_SIZE);
char *ptr = data;
TAOS_BIND *bind = (TAOS_BIND *)(bindArray + 0);
int64_t *bind_ts;
bind_ts = (int64_t *)ptr;
bind->buffer_type = TSDB_DATA_TYPE_TIMESTAMP;
if (tsRand) {
*bind_ts = startTime + getTSRandTail(
stbInfo->timeStampStep, k,
stbInfo->disorderRatio,
stbInfo->disorderRange);
} else {
*bind_ts = startTime + stbInfo->timeStampStep * k;
}
bind->buffer_length = sizeof(int64_t);
bind->buffer = bind_ts;
bind->length = &bind->buffer_length;
bind->is_null = NULL;
ptr += bind->buffer_length;
for (int i = 0; i < stbInfo->columnCount; i ++) {
bind = (TAOS_BIND *)((char *)bindArray + (sizeof(TAOS_BIND) * (i + 1)));
if (0 == strncasecmp(stbInfo->columns[i].dataType,
"BINARY", strlen("BINARY"))) {
if (stbInfo->columns[i].dataLen > TSDB_MAX_BINARY_LEN) {
errorPrint( "binary length overflow, max size:%u\n",
(uint32_t)TSDB_MAX_BINARY_LEN);
return -1;
}
char *bind_binary = (char *)ptr;
rand_string(bind_binary, stbInfo->columns[i].dataLen);
bind->buffer_type = TSDB_DATA_TYPE_BINARY;
bind->buffer_length = stbInfo->columns[i].dataLen;
bind->buffer = bind_binary;
bind->length = &bind->buffer_length;
bind->is_null = NULL;
ptr += bind->buffer_length;
} else if (0 == strncasecmp(stbInfo->columns[i].dataType,
"NCHAR", strlen("NCHAR"))) {
if (stbInfo->columns[i].dataLen > TSDB_MAX_BINARY_LEN) {
errorPrint( "nchar length overflow, max size:%u\n",
(uint32_t)TSDB_MAX_BINARY_LEN);
return -1;
}
char *bind_nchar = (char *)ptr;
rand_string(bind_nchar, stbInfo->columns[i].dataLen);
bind->buffer_type = TSDB_DATA_TYPE_NCHAR;
bind->buffer_length = strlen(bind_nchar);
bind->buffer = bind_nchar;
bind->length = &bind->buffer_length;
bind->is_null = NULL;
ptr += bind->buffer_length;
} else if (0 == strncasecmp(stbInfo->columns[i].dataType,
"INT", strlen("INT"))) {
int32_t *bind_int = (int32_t *)ptr;
*bind_int = rand_int();
bind->buffer_type = TSDB_DATA_TYPE_INT;
bind->buffer_length = sizeof(int32_t);
bind->buffer = bind_int;
bind->length = &bind->buffer_length;
bind->is_null = NULL;
ptr += bind->buffer_length;
} else if (0 == strncasecmp(stbInfo->columns[i].dataType,
"BIGINT", strlen("BIGINT"))) {
int64_t *bind_bigint = (int64_t *)ptr;
*bind_bigint = rand_bigint();
bind->buffer_type = TSDB_DATA_TYPE_BIGINT;
bind->buffer_length = sizeof(int64_t);
bind->buffer = bind_bigint;
bind->length = &bind->buffer_length;
bind->is_null = NULL;
ptr += bind->buffer_length;
} else if (0 == strncasecmp(stbInfo->columns[i].dataType,
"FLOAT", strlen("FLOAT"))) {
float *bind_float = (float *)ptr;
*bind_float = rand_float();
bind->buffer_type = TSDB_DATA_TYPE_FLOAT;
bind->buffer_length = sizeof(float);
bind->buffer = bind_float;
bind->length = &bind->buffer_length;
bind->is_null = NULL;
ptr += bind->buffer_length;
} else if (0 == strncasecmp(stbInfo->columns[i].dataType,
"DOUBLE", strlen("DOUBLE"))) {
double *bind_double = (double *)ptr;
*bind_double = rand_double();
bind->buffer_type = TSDB_DATA_TYPE_DOUBLE;
bind->buffer_length = sizeof(double);
bind->buffer = bind_double;
bind->length = &bind->buffer_length;
bind->is_null = NULL;
ptr += bind->buffer_length;
} else if (0 == strncasecmp(stbInfo->columns[i].dataType,
"SMALLINT", strlen("SMALLINT"))) {
int16_t *bind_smallint = (int16_t *)ptr;
*bind_smallint = rand_smallint();
bind->buffer_type = TSDB_DATA_TYPE_SMALLINT;
bind->buffer_length = sizeof(int16_t);
bind->buffer = bind_smallint;
bind->length = &bind->buffer_length;
bind->is_null = NULL;
ptr += bind->buffer_length;
} else if (0 == strncasecmp(stbInfo->columns[i].dataType,
"TINYINT", strlen("TINYINT"))) {
int8_t *bind_tinyint = (int8_t *)ptr;
*bind_tinyint = rand_tinyint();
bind->buffer_type = TSDB_DATA_TYPE_TINYINT;
bind->buffer_length = sizeof(int8_t);
bind->buffer = bind_tinyint;
bind->length = &bind->buffer_length;
bind->is_null = NULL;
ptr += bind->buffer_length;
} else if (0 == strncasecmp(stbInfo->columns[i].dataType,
"BOOL", strlen("BOOL"))) {
int8_t *bind_bool = (int8_t *)ptr;
*bind_bool = rand_bool();
bind->buffer_type = TSDB_DATA_TYPE_BOOL;
bind->buffer_length = sizeof(int8_t);
bind->buffer = bind_bool;
bind->length = &bind->buffer_length;
bind->is_null = NULL;
ptr += bind->buffer_length;
} else if (0 == strncasecmp(stbInfo->columns[i].dataType,
"TIMESTAMP", strlen("TIMESTAMP"))) {
int64_t *bind_ts2 = (int64_t *)ptr;
*bind_ts2 = rand_bigint();
bind->buffer_type = TSDB_DATA_TYPE_TIMESTAMP;
bind->buffer_length = sizeof(int64_t);
bind->buffer = bind_ts2;
bind->length = &bind->buffer_length;
bind->is_null = NULL;
ptr += bind->buffer_length;
} else {
errorPrint( "No support data type: %s\n",
stbInfo->columns[i].dataType);
return -1;
}
}
taos_stmt_bind_param(stmt, (TAOS_BIND *)bindArray);
// if msg > 3MB, break
taos_stmt_add_batch(stmt);
k++;
recordFrom ++;
if (recordFrom >= insertRows) {
break;
}
}
return k;
}
static int32_t generateStbProgressiveData(
SSuperTable *superTblInfo, SSuperTable *superTblInfo,
char *tableName, char *tableName,
int64_t tableSeq, int64_t tableSeq,
char *dbName, char *buffer, char *dbName, char *buffer,
int64_t insertRows, int64_t insertRows,
uint64_t startFrom, int64_t startTime, int64_t *pSamplePos, uint64_t recordFrom, int64_t startTime, int64_t *pSamplePos,
int64_t *pRemainderBufLen) int64_t *pRemainderBufLen)
{ {
assert(buffer != NULL); assert(buffer != NULL);
...@@ -5262,17 +5477,22 @@ static int64_t generateStbProgressiveData( ...@@ -5262,17 +5477,22 @@ static int64_t generateStbProgressiveData(
return generateStbDataTail(superTblInfo, return generateStbDataTail(superTblInfo,
g_args.num_of_RPR, pstr, *pRemainderBufLen, g_args.num_of_RPR, pstr, *pRemainderBufLen,
insertRows, startFrom, insertRows, recordFrom,
startTime, startTime,
pSamplePos, &dataLen); pSamplePos, &dataLen);
} }
static int64_t prepareStmtWithoutStb(char *tableName)
{
return -1;
}
static int64_t generateProgressiveDataWithoutStb( static int64_t generateProgressiveDataWithoutStb(
char *tableName, char *tableName,
int64_t tableSeq, /* int64_t tableSeq, */
threadInfo *pThreadInfo, char *buffer, threadInfo *pThreadInfo, char *buffer,
int64_t insertRows, int64_t insertRows,
uint64_t startFrom, int64_t startTime, int64_t *pSamplePos, uint64_t recordFrom, int64_t startTime, /*int64_t *pSamplePos, */
int64_t *pRemainderBufLen) int64_t *pRemainderBufLen)
{ {
assert(buffer != NULL); assert(buffer != NULL);
...@@ -5293,7 +5513,7 @@ static int64_t generateProgressiveDataWithoutStb( ...@@ -5293,7 +5513,7 @@ static int64_t generateProgressiveDataWithoutStb(
int64_t dataLen; int64_t dataLen;
return generateDataTailWithoutStb( return generateDataTailWithoutStb(
g_args.num_of_RPR, pstr, *pRemainderBufLen, insertRows, startFrom, g_args.num_of_RPR, pstr, *pRemainderBufLen, insertRows, recordFrom,
startTime, startTime,
/*pSamplePos, */&dataLen); /*pSamplePos, */&dataLen);
} }
...@@ -5313,7 +5533,7 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) { ...@@ -5313,7 +5533,7 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) {
pThreadInfo->threadID, __func__, __LINE__); pThreadInfo->threadID, __func__, __LINE__);
int64_t insertRows; int64_t insertRows;
uint64_t interlaceRows; uint32_t interlaceRows;
uint64_t maxSqlLen; uint64_t maxSqlLen;
int64_t nTimeStampStep; int64_t nTimeStampStep;
uint64_t insert_interval; uint64_t insert_interval;
...@@ -5351,8 +5571,8 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) { ...@@ -5351,8 +5571,8 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) {
if (interlaceRows > g_args.num_of_RPR) if (interlaceRows > g_args.num_of_RPR)
interlaceRows = g_args.num_of_RPR; interlaceRows = g_args.num_of_RPR;
uint64_t batchPerTbl = interlaceRows; uint32_t batchPerTbl = interlaceRows;
uint64_t batchPerTblTimes; uint32_t batchPerTblTimes;
if ((interlaceRows > 0) && (pThreadInfo->ntables > 1)) { if ((interlaceRows > 0) && (pThreadInfo->ntables > 1)) {
batchPerTblTimes = batchPerTblTimes =
...@@ -5401,9 +5621,9 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) { ...@@ -5401,9 +5621,9 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) {
pstr += len; pstr += len;
remainderBufLen -= len; remainderBufLen -= len;
uint64_t recOfBatch = 0; uint32_t recOfBatch = 0;
for (uint64_t i = 0; i < batchPerTblTimes; i ++) { for (uint32_t i = 0; i < batchPerTblTimes; i ++) {
char tableName[TSDB_TABLE_NAME_LEN]; char tableName[TSDB_TABLE_NAME_LEN];
getTableName(tableName, pThreadInfo, tableSeq); getTableName(tableName, pThreadInfo, tableSeq);
...@@ -5416,11 +5636,12 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) { ...@@ -5416,11 +5636,12 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) {
uint64_t oldRemainderLen = remainderBufLen; uint64_t oldRemainderLen = remainderBufLen;
int64_t generated; int32_t generated;
if (superTblInfo) { if (superTblInfo) {
generated = generateStbInterlaceData( generated = generateStbInterlaceData(
superTblInfo, superTblInfo,
tableName, batchPerTbl, i, batchPerTblTimes, tableName, batchPerTbl, i,
batchPerTblTimes,
tableSeq, tableSeq,
pThreadInfo, pstr, pThreadInfo, pstr,
insertRows, insertRows,
...@@ -5435,10 +5656,10 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) { ...@@ -5435,10 +5656,10 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) {
&remainderBufLen); &remainderBufLen);
} }
debugPrint("[%d] %s() LN%d, generated records is %"PRId64"\n", debugPrint("[%d] %s() LN%d, generated records is %d\n",
pThreadInfo->threadID, __func__, __LINE__, generated); pThreadInfo->threadID, __func__, __LINE__, generated);
if (generated < 0) { if (generated < 0) {
errorPrint("[%d] %s() LN%d, generated records is %"PRId64"\n", errorPrint("[%d] %s() LN%d, generated records is %d\n",
pThreadInfo->threadID, __func__, __LINE__, generated); pThreadInfo->threadID, __func__, __LINE__, generated);
goto free_of_interlace; goto free_of_interlace;
} else if (generated == 0) { } else if (generated == 0) {
...@@ -5450,7 +5671,7 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) { ...@@ -5450,7 +5671,7 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) {
pstr += (oldRemainderLen - remainderBufLen); pstr += (oldRemainderLen - remainderBufLen);
// startTime += batchPerTbl * superTblInfo->timeStampStep; // startTime += batchPerTbl * superTblInfo->timeStampStep;
pThreadInfo->totalInsertRows += batchPerTbl; pThreadInfo->totalInsertRows += batchPerTbl;
verbosePrint("[%d] %s() LN%d batchPerTbl=%"PRId64" recOfBatch=%"PRId64"\n", verbosePrint("[%d] %s() LN%d batchPerTbl=%d recOfBatch=%d\n",
pThreadInfo->threadID, __func__, __LINE__, pThreadInfo->threadID, __func__, __LINE__,
batchPerTbl, recOfBatch); batchPerTbl, recOfBatch);
...@@ -5466,7 +5687,7 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) { ...@@ -5466,7 +5687,7 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) {
if (generatedRecPerTbl >= insertRows) if (generatedRecPerTbl >= insertRows)
break; break;
int remainRows = insertRows - generatedRecPerTbl; int64_t remainRows = insertRows - generatedRecPerTbl;
if ((remainRows > 0) && (batchPerTbl > remainRows)) if ((remainRows > 0) && (batchPerTbl > remainRows))
batchPerTbl = remainRows; batchPerTbl = remainRows;
...@@ -5482,7 +5703,7 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) { ...@@ -5482,7 +5703,7 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) {
break; break;
} }
verbosePrint("[%d] %s() LN%d recOfBatch=%"PRIu64" totalInsertRows=%"PRIu64"\n", verbosePrint("[%d] %s() LN%d recOfBatch=%d totalInsertRows=%"PRIu64"\n",
pThreadInfo->threadID, __func__, __LINE__, recOfBatch, pThreadInfo->threadID, __func__, __LINE__, recOfBatch,
pThreadInfo->totalInsertRows); pThreadInfo->totalInsertRows);
verbosePrint("[%d] %s() LN%d, buffer=%s\n", verbosePrint("[%d] %s() LN%d, buffer=%s\n",
...@@ -5491,7 +5712,7 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) { ...@@ -5491,7 +5712,7 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) {
startTs = taosGetTimestampMs(); startTs = taosGetTimestampMs();
if (recOfBatch == 0) { if (recOfBatch == 0) {
errorPrint("[%d] %s() LN%d try inserting records of batch is %"PRIu64"\n", errorPrint("[%d] %s() LN%d try inserting records of batch is %d\n",
pThreadInfo->threadID, __func__, __LINE__, pThreadInfo->threadID, __func__, __LINE__,
recOfBatch); recOfBatch);
errorPrint("%s\n", "\tPlease check if the batch or the buffer length is proper value!\n"); errorPrint("%s\n", "\tPlease check if the batch or the buffer length is proper value!\n");
...@@ -5513,7 +5734,7 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) { ...@@ -5513,7 +5734,7 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) {
pThreadInfo->totalDelay += delay; pThreadInfo->totalDelay += delay;
if (recOfBatch != affectedRows) { if (recOfBatch != affectedRows) {
errorPrint("[%d] %s() LN%d execInsert insert %"PRIu64", affected rows: %"PRId64"\n%s\n", errorPrint("[%d] %s() LN%d execInsert insert %d, affected rows: %"PRId64"\n%s\n",
pThreadInfo->threadID, __func__, __LINE__, pThreadInfo->threadID, __func__, __LINE__,
recOfBatch, affectedRows, pThreadInfo->buffer); recOfBatch, affectedRows, pThreadInfo->buffer);
goto free_of_interlace; goto free_of_interlace;
...@@ -5574,12 +5795,6 @@ static void* syncWriteProgressive(threadInfo *pThreadInfo) { ...@@ -5574,12 +5795,6 @@ static void* syncWriteProgressive(threadInfo *pThreadInfo) {
uint64_t startTs = taosGetTimestampMs(); uint64_t startTs = taosGetTimestampMs();
uint64_t endTs; uint64_t endTs;
/* int insert_interval =
superTblInfo?superTblInfo->insertInterval:g_args.insert_interval;
uint64_t st = 0;
uint64_t et = 0xffffffff;
*/
pThreadInfo->totalInsertRows = 0; pThreadInfo->totalInsertRows = 0;
pThreadInfo->totalAffectedRows = 0; pThreadInfo->totalAffectedRows = 0;
...@@ -5606,21 +5821,34 @@ static void* syncWriteProgressive(threadInfo *pThreadInfo) { ...@@ -5606,21 +5821,34 @@ static void* syncWriteProgressive(threadInfo *pThreadInfo) {
pstr += len; pstr += len;
remainderBufLen -= len; remainderBufLen -= len;
int64_t generated; int32_t generated;
if (superTblInfo) { if (superTblInfo) {
if (superTblInfo->iface == STMT_IFACE) {
generated = prepareStbStmt(superTblInfo,
pThreadInfo->stmt,
tableName, g_args.num_of_RPR,
insertRows, i, start_time, pstr);
} else {
generated = generateStbProgressiveData( generated = generateStbProgressiveData(
superTblInfo, superTblInfo,
tableName, tableSeq, pThreadInfo->db_name, pstr, insertRows, tableName, tableSeq, pThreadInfo->db_name, pstr,
i, start_time, insertRows, i, start_time,
&(pThreadInfo->samplePos), &(pThreadInfo->samplePos),
&remainderBufLen); &remainderBufLen);
}
} else {
if (g_args.iface == STMT_IFACE) {
generated = prepareStmtWithoutStb(tableName);
} else { } else {
generated = generateProgressiveDataWithoutStb( generated = generateProgressiveDataWithoutStb(
tableName, tableSeq, pThreadInfo, pstr, insertRows, tableName,
/* tableSeq, */
pThreadInfo, pstr, insertRows,
i, start_time, i, start_time,
&(pThreadInfo->samplePos), /* &(pThreadInfo->samplePos), */
&remainderBufLen); &remainderBufLen);
} }
}
if (generated > 0) if (generated > 0)
i += generated; i += generated;
else else
...@@ -5687,7 +5915,7 @@ static void* syncWrite(void *sarg) { ...@@ -5687,7 +5915,7 @@ static void* syncWrite(void *sarg) {
threadInfo *pThreadInfo = (threadInfo *)sarg; threadInfo *pThreadInfo = (threadInfo *)sarg;
SSuperTable* superTblInfo = pThreadInfo->superTblInfo; SSuperTable* superTblInfo = pThreadInfo->superTblInfo;
int interlaceRows; uint32_t interlaceRows;
if (superTblInfo) { if (superTblInfo) {
if ((superTblInfo->interlaceRows == 0) if ((superTblInfo->interlaceRows == 0)
...@@ -5818,7 +6046,7 @@ static void startMultiThreadInsertData(int threads, char* db_name, ...@@ -5818,7 +6046,7 @@ static void startMultiThreadInsertData(int threads, char* db_name,
char* precision,SSuperTable* superTblInfo) { char* precision,SSuperTable* superTblInfo) {
//TAOS* taos; //TAOS* taos;
//if (0 == strncasecmp(superTblInfo->insertMode, "taosc", 5)) { //if (0 == strncasecmp(superTblInfo->iface, "taosc", 5)) {
// taos = taos_connect(g_Dbs.host, g_Dbs.user, g_Dbs.password, db_name, g_Dbs.port); // taos = taos_connect(g_Dbs.host, g_Dbs.user, g_Dbs.password, db_name, g_Dbs.port);
// if (NULL == taos) { // if (NULL == taos) {
// printf("connect to server fail, reason: %s\n", taos_errstr(NULL)); // printf("connect to server fail, reason: %s\n", taos_errstr(NULL));
...@@ -5887,7 +6115,7 @@ static void startMultiThreadInsertData(int threads, char* db_name, ...@@ -5887,7 +6115,7 @@ static void startMultiThreadInsertData(int threads, char* db_name,
} }
int64_t ntables = 0; int64_t ntables = 0;
uint64_t startFrom; uint64_t tableFrom;
if (superTblInfo) { if (superTblInfo) {
int64_t limit; int64_t limit;
...@@ -5914,7 +6142,7 @@ static void startMultiThreadInsertData(int threads, char* db_name, ...@@ -5914,7 +6142,7 @@ static void startMultiThreadInsertData(int threads, char* db_name,
} }
ntables = limit; ntables = limit;
startFrom = offset; tableFrom = offset;
if ((superTblInfo->childTblExists != TBL_NO_EXISTS) if ((superTblInfo->childTblExists != TBL_NO_EXISTS)
&& ((superTblInfo->childTblOffset + superTblInfo->childTblLimit ) && ((superTblInfo->childTblOffset + superTblInfo->childTblLimit )
...@@ -5946,7 +6174,7 @@ static void startMultiThreadInsertData(int threads, char* db_name, ...@@ -5946,7 +6174,7 @@ static void startMultiThreadInsertData(int threads, char* db_name,
offset); offset);
} else { } else {
ntables = g_args.num_of_tables; ntables = g_args.num_of_tables;
startFrom = 0; tableFrom = 0;
} }
taos_close(taos0); taos_close(taos0);
...@@ -5963,8 +6191,9 @@ static void startMultiThreadInsertData(int threads, char* db_name, ...@@ -5963,8 +6191,9 @@ static void startMultiThreadInsertData(int threads, char* db_name,
} }
if ((superTblInfo) if ((superTblInfo)
&& (superTblInfo->insertMode == REST_IFACE)) { && (superTblInfo->iface == REST_IFACE)) {
if (convertHostToServAddr(g_Dbs.host, g_Dbs.port, &(g_Dbs.serv_addr)) != 0) { if (convertHostToServAddr(
g_Dbs.host, g_Dbs.port, &(g_Dbs.serv_addr)) != 0) {
exit(-1); exit(-1);
} }
} }
...@@ -5989,8 +6218,8 @@ static void startMultiThreadInsertData(int threads, char* db_name, ...@@ -5989,8 +6218,8 @@ static void startMultiThreadInsertData(int threads, char* db_name,
pThreadInfo->minDelay = UINT64_MAX; pThreadInfo->minDelay = UINT64_MAX;
if ((NULL == superTblInfo) || if ((NULL == superTblInfo) ||
(superTblInfo->insertMode != REST_IFACE)) { (superTblInfo->iface != REST_IFACE)) {
//pThreadInfo->taos = taos; //t_info->taos = taos;
pThreadInfo->taos = taos_connect( pThreadInfo->taos = taos_connect(
g_Dbs.host, g_Dbs.user, g_Dbs.host, g_Dbs.user,
g_Dbs.password, db_name, g_Dbs.port); g_Dbs.password, db_name, g_Dbs.port);
...@@ -6003,7 +6232,7 @@ static void startMultiThreadInsertData(int threads, char* db_name, ...@@ -6003,7 +6232,7 @@ static void startMultiThreadInsertData(int threads, char* db_name,
exit(-1); exit(-1);
} }
if ((superTblInfo) && (superTblInfo->insertMode == STMT_IFACE)) { if ((superTblInfo) && (superTblInfo->iface == STMT_IFACE)) {
pThreadInfo->stmt = taos_stmt_init(pThreadInfo->taos); pThreadInfo->stmt = taos_stmt_init(pThreadInfo->taos);
if (NULL == pThreadInfo->stmt) { if (NULL == pThreadInfo->stmt) {
errorPrint( errorPrint(
...@@ -6022,10 +6251,10 @@ static void startMultiThreadInsertData(int threads, char* db_name, ...@@ -6022,10 +6251,10 @@ static void startMultiThreadInsertData(int threads, char* db_name,
/* if ((NULL == superTblInfo) /* if ((NULL == superTblInfo)
|| (0 == superTblInfo->multiThreadWriteOneTbl)) { || (0 == superTblInfo->multiThreadWriteOneTbl)) {
*/ */
pThreadInfo->start_table_from = startFrom; pThreadInfo->start_table_from = tableFrom;
pThreadInfo->ntables = i<b?a+1:a; pThreadInfo->ntables = i<b?a+1:a;
pThreadInfo->end_table_to = i < b ? startFrom + a : startFrom + a - 1; pThreadInfo->end_table_to = i < b ? tableFrom + a : tableFrom + a - 1;
startFrom = pThreadInfo->end_table_to + 1; tableFrom = pThreadInfo->end_table_to + 1;
/* } else { /* } else {
pThreadInfo->start_table_from = 0; pThreadInfo->start_table_from = 0;
pThreadInfo->ntables = superTblInfo->childTblCount; pThreadInfo->ntables = superTblInfo->childTblCount;
...@@ -6613,7 +6842,6 @@ static int queryTestProcess() { ...@@ -6613,7 +6842,6 @@ static int queryTestProcess() {
char sqlStr[MAX_TB_NAME_SIZE*2]; char sqlStr[MAX_TB_NAME_SIZE*2];
sprintf(sqlStr, "use %s", g_queryInfo.dbName); sprintf(sqlStr, "use %s", g_queryInfo.dbName);
verbosePrint("%s() %d sqlStr: %s\n", __func__, __LINE__, sqlStr);
if (0 != queryDbExec(taos, sqlStr, NO_INSERT_TYPE, false)) { if (0 != queryDbExec(taos, sqlStr, NO_INSERT_TYPE, false)) {
taos_close(taos); taos_close(taos);
free(infos); free(infos);
...@@ -6665,15 +6893,15 @@ static int queryTestProcess() { ...@@ -6665,15 +6893,15 @@ static int queryTestProcess() {
b = ntables % threads; b = ntables % threads;
} }
uint64_t startFrom = 0; uint64_t tableFrom = 0;
for (int i = 0; i < threads; i++) { for (int i = 0; i < threads; i++) {
threadInfo *pThreadInfo = infosOfSub + i; threadInfo *pThreadInfo = infosOfSub + i;
pThreadInfo->threadID = i; pThreadInfo->threadID = i;
pThreadInfo->start_table_from = startFrom; pThreadInfo->start_table_from = tableFrom;
pThreadInfo->ntables = i<b?a+1:a; pThreadInfo->ntables = i<b?a+1:a;
pThreadInfo->end_table_to = i < b ? startFrom + a : startFrom + a - 1; pThreadInfo->end_table_to = i < b ? tableFrom + a : tableFrom + a - 1;
startFrom = pThreadInfo->end_table_to + 1; tableFrom = pThreadInfo->end_table_to + 1;
pThreadInfo->taos = NULL; // TODO: workaround to use separate taos connection; pThreadInfo->taos = NULL; // TODO: workaround to use separate taos connection;
pthread_create(pidsOfSub + i, NULL, superTableQuery, pThreadInfo); pthread_create(pidsOfSub + i, NULL, superTableQuery, pThreadInfo);
} }
...@@ -6937,7 +7165,6 @@ static void *specifiedSubscribe(void *sarg) { ...@@ -6937,7 +7165,6 @@ static void *specifiedSubscribe(void *sarg) {
char sqlStr[MAX_TB_NAME_SIZE*2]; char sqlStr[MAX_TB_NAME_SIZE*2];
sprintf(sqlStr, "use %s", g_queryInfo.dbName); sprintf(sqlStr, "use %s", g_queryInfo.dbName);
debugPrint("%s() %d sqlStr: %s\n", __func__, __LINE__, sqlStr);
if (0 != queryDbExec(pThreadInfo->taos, sqlStr, NO_INSERT_TYPE, false)) { if (0 != queryDbExec(pThreadInfo->taos, sqlStr, NO_INSERT_TYPE, false)) {
taos_close(pThreadInfo->taos); taos_close(pThreadInfo->taos);
return NULL; return NULL;
...@@ -7123,17 +7350,17 @@ static int subscribeTestProcess() { ...@@ -7123,17 +7350,17 @@ static int subscribeTestProcess() {
} }
for (uint64_t i = 0; i < g_queryInfo.superQueryInfo.sqlCount; i++) { for (uint64_t i = 0; i < g_queryInfo.superQueryInfo.sqlCount; i++) {
uint64_t startFrom = 0; uint64_t tableFrom = 0;
for (int j = 0; j < threads; j++) { for (int j = 0; j < threads; j++) {
uint64_t seq = i * threads + j; uint64_t seq = i * threads + j;
threadInfo *pThreadInfo = infosOfStable + seq; threadInfo *pThreadInfo = infosOfStable + seq;
pThreadInfo->threadID = seq; pThreadInfo->threadID = seq;
pThreadInfo->querySeq = i; pThreadInfo->querySeq = i;
pThreadInfo->start_table_from = startFrom; pThreadInfo->start_table_from = tableFrom;
pThreadInfo->ntables = j<b?a+1:a; pThreadInfo->ntables = j<b?a+1:a;
pThreadInfo->end_table_to = j<b?startFrom+a:startFrom+a-1; pThreadInfo->end_table_to = j<b?tableFrom+a:tableFrom+a-1;
startFrom = pThreadInfo->end_table_to + 1; tableFrom = pThreadInfo->end_table_to + 1;
pThreadInfo->taos = NULL; // TODO: workaround to use separate taos connection; pThreadInfo->taos = NULL; // TODO: workaround to use separate taos connection;
pthread_create(pidsOfStable + seq, pthread_create(pidsOfStable + seq,
NULL, superSubscribe, pThreadInfo); NULL, superSubscribe, pThreadInfo);
...@@ -7252,7 +7479,7 @@ static void setParaFromArg(){ ...@@ -7252,7 +7479,7 @@ static void setParaFromArg(){
tstrncpy(g_Dbs.db[0].superTbls[0].childTblPrefix, tstrncpy(g_Dbs.db[0].superTbls[0].childTblPrefix,
g_args.tb_prefix, MAX_TB_NAME_SIZE); g_args.tb_prefix, MAX_TB_NAME_SIZE);
tstrncpy(g_Dbs.db[0].superTbls[0].dataSource, "rand", MAX_TB_NAME_SIZE); tstrncpy(g_Dbs.db[0].superTbls[0].dataSource, "rand", MAX_TB_NAME_SIZE);
g_Dbs.db[0].superTbls[0].insertMode = g_args.iface; g_Dbs.db[0].superTbls[0].iface = g_args.iface;
tstrncpy(g_Dbs.db[0].superTbls[0].startTimestamp, tstrncpy(g_Dbs.db[0].superTbls[0].startTimestamp,
"2017-07-14 10:40:00.000", MAX_TB_NAME_SIZE); "2017-07-14 10:40:00.000", MAX_TB_NAME_SIZE);
g_Dbs.db[0].superTbls[0].timeStampStep = DEFAULT_TIMESTAMP_STEP; g_Dbs.db[0].superTbls[0].timeStampStep = DEFAULT_TIMESTAMP_STEP;
...@@ -7365,7 +7592,6 @@ static void querySqlFile(TAOS* taos, char* sqlFile) ...@@ -7365,7 +7592,6 @@ static void querySqlFile(TAOS* taos, char* sqlFile)
} }
memcpy(cmd + cmd_len, line, read_len); memcpy(cmd + cmd_len, line, read_len);
verbosePrint("%s() LN%d cmd: %s\n", __func__, __LINE__, cmd);
if (0 != queryDbExec(taos, cmd, NO_INSERT_TYPE, false)) { if (0 != queryDbExec(taos, cmd, NO_INSERT_TYPE, false)) {
errorPrint("%s() LN%d, queryDbExec %s failed!\n", errorPrint("%s() LN%d, queryDbExec %s failed!\n",
__func__, __LINE__, cmd); __func__, __LINE__, cmd);
......
src/mnode/src/mnodeSdb.c
浏览文件 @ 9a4ccd2c
...@@ -719,13 +719,13 @@ static int32_t sdbProcessWrite(void *wparam, void *hparam, int32_t qtype, void * ...@@ -719,13 +719,13 @@ static int32_t sdbProcessWrite(void *wparam, void *hparam, int32_t qtype, void *
if (action == SDB_ACTION_INSERT) { if (action == SDB_ACTION_INSERT) {
return sdbPerformInsertAction(pHead, pTable); return sdbPerformInsertAction(pHead, pTable);
} else if (action == SDB_ACTION_DELETE) { } else if (action == SDB_ACTION_DELETE) {
if (qtype == TAOS_QTYPE_FWD) { //if (qtype == TAOS_QTYPE_FWD) {
// Drop database/stable may take a long time and cause a timeout, so we confirm first then reput it into queue // Drop database/stable may take a long time and cause a timeout, so we confirm first then reput it into queue
sdbWriteFwdToQueue(1, hparam, TAOS_QTYPE_QUERY, unused); // sdbWriteFwdToQueue(1, hparam, TAOS_QTYPE_QUERY, unused);
return TSDB_CODE_SUCCESS; // return TSDB_CODE_SUCCESS;
} else { //} else {
return sdbPerformDeleteAction(pHead, pTable); return sdbPerformDeleteAction(pHead, pTable);
} //}
} else if (action == SDB_ACTION_UPDATE) { } else if (action == SDB_ACTION_UPDATE) {
return sdbPerformUpdateAction(pHead, pTable); return sdbPerformUpdateAction(pHead, pTable);
} else { } else {
......
src/mnode/src/mnodeTable.c
浏览文件 @ 9a4ccd2c
...@@ -1189,8 +1189,8 @@ static int32_t mnodeFindSuperTableTagIndex(SSTableObj *pStable, const char *tagN ...@@ -1189,8 +1189,8 @@ static int32_t mnodeFindSuperTableTagIndex(SSTableObj *pStable, const char *tagN
static int32_t mnodeAddSuperTableTagCb(SMnodeMsg *pMsg, int32_t code) { static int32_t mnodeAddSuperTableTagCb(SMnodeMsg *pMsg, int32_t code) {
SSTableObj *pStable = (SSTableObj *)pMsg->pTable; SSTableObj *pStable = (SSTableObj *)pMsg->pTable;
mLInfo("msg:%p, app:%p stable %s, add tag result:%s", pMsg, pMsg->rpcMsg.ahandle, pStable->info.tableId, mLInfo("msg:%p, app:%p stable %s, add tag result:%s, numOfTags:%d", pMsg, pMsg->rpcMsg.ahandle, pStable->info.tableId,
tstrerror(code)); tstrerror(code), pStable->numOfTags);
return code; return code;
} }
......
src/mnode/src/mnodeVgroup.c
浏览文件 @ 9a4ccd2c
...@@ -121,7 +121,7 @@ static int32_t mnodeVgroupActionDelete(SSdbRow *pRow) { ...@@ -121,7 +121,7 @@ static int32_t mnodeVgroupActionDelete(SSdbRow *pRow) {
SVgObj *pVgroup = pRow->pObj; SVgObj *pVgroup = pRow->pObj;
if (pVgroup->pDb == NULL) { if (pVgroup->pDb == NULL) {
mError("vgId:%d, db:%s is not exist while insert into hash", pVgroup->vgId, pVgroup->dbName); mError("vgId:%d, db:%s is not exist while delete from hash", pVgroup->vgId, pVgroup->dbName);
return TSDB_CODE_MND_VGROUP_NOT_EXIST; return TSDB_CODE_MND_VGROUP_NOT_EXIST;
} }
......
src/sync/src/syncMain.c
浏览文件 @ 9a4ccd2c
...@@ -709,7 +709,7 @@ static void syncChooseMaster(SSyncNode *pNode) { ...@@ -709,7 +709,7 @@ static void syncChooseMaster(SSyncNode *pNode) {
} }
static SSyncPeer *syncCheckMaster(SSyncNode *pNode) { static SSyncPeer *syncCheckMaster(SSyncNode *pNode) {
int32_t onlineNum = 0; int32_t onlineNum = 0, arbOnlineNum = 0;
int32_t masterIndex = -1; int32_t masterIndex = -1;
int32_t replica = pNode->replica; int32_t replica = pNode->replica;
...@@ -723,13 +723,15 @@ static SSyncPeer *syncCheckMaster(SSyncNode *pNode) { ...@@ -723,13 +723,15 @@ static SSyncPeer *syncCheckMaster(SSyncNode *pNode) {
SSyncPeer *pArb = pNode->peerInfo[TAOS_SYNC_MAX_REPLICA]; SSyncPeer *pArb = pNode->peerInfo[TAOS_SYNC_MAX_REPLICA];
if (pArb && pArb->role != TAOS_SYNC_ROLE_OFFLINE) { if (pArb && pArb->role != TAOS_SYNC_ROLE_OFFLINE) {
onlineNum++; onlineNum++;
++arbOnlineNum;
replica = pNode->replica + 1; replica = pNode->replica + 1;
} }
if (onlineNum <= replica * 0.5) { if (onlineNum <= replica * 0.5) {
if (nodeRole != TAOS_SYNC_ROLE_UNSYNCED) { if (nodeRole != TAOS_SYNC_ROLE_UNSYNCED) {
if (nodeRole == TAOS_SYNC_ROLE_MASTER && onlineNum == replica * 0.5 && onlineNum >= 1) { if (nodeRole == TAOS_SYNC_ROLE_MASTER && onlineNum == replica * 0.5 && ((replica > 2 && onlineNum - arbOnlineNum > 1) || pNode->replica < 3)) {
sInfo("vgId:%d, self keep work as master, online:%d replica:%d", pNode->vgId, onlineNum, replica); sInfo("vgId:%d, self keep work as master, online:%d replica:%d", pNode->vgId, onlineNum, replica);
masterIndex = pNode->selfIndex;
} else { } else {
nodeRole = TAOS_SYNC_ROLE_UNSYNCED; nodeRole = TAOS_SYNC_ROLE_UNSYNCED;
sInfo("vgId:%d, self change to unsynced state, online:%d replica:%d", pNode->vgId, onlineNum, replica); sInfo("vgId:%d, self change to unsynced state, online:%d replica:%d", pNode->vgId, onlineNum, replica);
...@@ -1002,6 +1004,7 @@ static void syncProcessForwardFromPeer(char *cont, SSyncPeer *pPeer) { ...@@ -1002,6 +1004,7 @@ static void syncProcessForwardFromPeer(char *cont, SSyncPeer *pPeer) {
if (nodeRole == TAOS_SYNC_ROLE_SLAVE) { if (nodeRole == TAOS_SYNC_ROLE_SLAVE) {
// nodeVersion = pHead->version; // nodeVersion = pHead->version;
code = (*pNode->writeToCacheFp)(pNode->vgId, pHead, TAOS_QTYPE_FWD, NULL); code = (*pNode->writeToCacheFp)(pNode->vgId, pHead, TAOS_QTYPE_FWD, NULL);
syncConfirmForward(pNode->rid, pHead->version, code, false);
} else { } else {
if (nodeSStatus != TAOS_SYNC_STATUS_INIT) { if (nodeSStatus != TAOS_SYNC_STATUS_INIT) {
code = syncSaveIntoBuffer(pPeer, pHead); code = syncSaveIntoBuffer(pPeer, pHead);
...@@ -1404,7 +1407,7 @@ static void syncMonitorFwdInfos(void *param, void *tmrId) { ...@@ -1404,7 +1407,7 @@ static void syncMonitorFwdInfos(void *param, void *tmrId) {
pthread_mutex_lock(&pNode->mutex); pthread_mutex_lock(&pNode->mutex);
for (int32_t i = 0; i < pSyncFwds->fwds; ++i) { for (int32_t i = 0; i < pSyncFwds->fwds; ++i) {
SFwdInfo *pFwdInfo = pSyncFwds->fwdInfo + (pSyncFwds->first + i) % SYNC_MAX_FWDS; SFwdInfo *pFwdInfo = pSyncFwds->fwdInfo + (pSyncFwds->first + i) % SYNC_MAX_FWDS;
if (ABS(time - pFwdInfo->time) < 2000) break; if (ABS(time - pFwdInfo->time) < 10000) break;
sDebug("vgId:%d, forward info expired, hver:%" PRIu64 " curtime:%" PRIu64 " savetime:%" PRIu64, pNode->vgId, sDebug("vgId:%d, forward info expired, hver:%" PRIu64 " curtime:%" PRIu64 " savetime:%" PRIu64, pNode->vgId,
pFwdInfo->version, time, pFwdInfo->time); pFwdInfo->version, time, pFwdInfo->time);
......
src/util/src/tcache.c
浏览文件 @ 9a4ccd2c
...@@ -613,7 +613,7 @@ void doCleanupDataCache(SCacheObj *pCacheObj) { ...@@ -613,7 +613,7 @@ void doCleanupDataCache(SCacheObj *pCacheObj) {
// todo memory leak if there are object with refcount greater than 0 in hash table? // todo memory leak if there are object with refcount greater than 0 in hash table?
taosHashCleanup(pCacheObj->pHashTable); taosHashCleanup(pCacheObj->pHashTable);
taosTrashcanEmpty(pCacheObj, true); taosTrashcanEmpty(pCacheObj, false);
__cache_lock_destroy(pCacheObj); __cache_lock_destroy(pCacheObj);
......
src/vnode/src/vnodeMain.c
浏览文件 @ 9a4ccd2c
...@@ -454,7 +454,11 @@ void vnodeDestroy(SVnodeObj *pVnode) { ...@@ -454,7 +454,11 @@ void vnodeDestroy(SVnodeObj *pVnode) {
} }
if (pVnode->tsdb) { if (pVnode->tsdb) {
code = tsdbCloseRepo(pVnode->tsdb, 1); // the deleted vnode does not need to commit, so as to speed up the deletion
int toCommit = 1;
if (pVnode->dropped) toCommit = 0;
code = tsdbCloseRepo(pVnode->tsdb, toCommit);
pVnode->tsdb = NULL; pVnode->tsdb = NULL;
} }
......
src/vnode/src/vnodeSync.c
浏览文件 @ 9a4ccd2c
...@@ -126,11 +126,16 @@ void vnodeStopSyncFile(int32_t vgId, uint64_t fversion) { ...@@ -126,11 +126,16 @@ void vnodeStopSyncFile(int32_t vgId, uint64_t fversion) {
} }
void vnodeConfirmForard(int32_t vgId, void *wparam, int32_t code) { void vnodeConfirmForard(int32_t vgId, void *wparam, int32_t code) {
void *pVnode = vnodeAcquire(vgId); SVnodeObj *pVnode = vnodeAcquire(vgId);
if (pVnode == NULL) { if (pVnode == NULL) {
vError("vgId:%d, vnode not found while confirm forward", vgId); vError("vgId:%d, vnode not found while confirm forward", vgId);
} }
if (code == TSDB_CODE_SYN_CONFIRM_EXPIRED && pVnode->status == TAOS_VN_STATUS_CLOSING) {
vDebug("vgId:%d, db:%s, vnode is closing while confirm forward", vgId, pVnode->db);
code = TSDB_CODE_VND_IS_CLOSING;
}
dnodeSendRpcVWriteRsp(pVnode, wparam, code); dnodeSendRpcVWriteRsp(pVnode, wparam, code);
vnodeRelease(pVnode); vnodeRelease(pVnode);
} }
......
tests/mas/Jenkinsfile 0 → 100644
浏览文件 @ 9a4ccd2c
def pre_test(){
sh '''
sudo rmtaos||echo 'no taosd installed'
'''
sh '''
cd ${WKC}
git reset --hard
git checkout $BRANCH_NAME
git pull
git submodule update
cd ${WK}
git reset --hard
git checkout $BRANCH_NAME
git pull
export TZ=Asia/Harbin
date
rm -rf ${WK}/debug
mkdir debug
cd debug
cmake .. > /dev/null
make > /dev/null
make install > /dev/null
pip3 install ${WKC}/src/connector/python/linux/python3/
'''
return 1
}
pipeline {
agent none
environment{
WK = '/var/lib/jenkins/workspace/TDinternal'
WKC= '/var/lib/jenkins/workspace/TDinternal/community'
}
stages {
stage('Parallel test stage') {
parallel {
stage('pytest') {
agent{label 'slam1'}
steps {
pre_test()
sh '''
cd ${WKC}/tests
find pytest -name '*'sql|xargs rm -rf
./test-all.sh pytest
date'''
}
}
stage('test_b1') {
agent{label 'slam2'}
steps {
pre_test()
sh '''
cd ${WKC}/tests
./test-all.sh b1
date'''
}
}
stage('test_crash_gen') {
agent{label "slam3"}
steps {
pre_test()
sh '''
cd ${WKC}/tests/pytest
'''
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
sh '''
cd ${WKC}/tests/pytest
./crash_gen.sh -a -p -t 4 -s 2000
'''
}
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
sh '''
cd ${WKC}/tests/pytest
rm -rf /var/lib/taos/*
rm -rf /var/log/taos/*
./handle_crash_gen_val_log.sh
'''
}
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
sh '''
cd ${WKC}/tests/pytest
rm -rf /var/lib/taos/*
rm -rf /var/log/taos/*
./handle_taosd_val_log.sh
'''
}
sh'''
systemctl start taosd
sleep 10
'''
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
sh '''
cd ${WKC}/tests/gotest
bash batchtest.sh
'''
}
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
sh '''
cd ${WKC}/tests/examples/python/PYTHONConnectorChecker
python3 PythonChecker.py
'''
}
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
sh '''
cd ${WKC}/tests/examples/JDBC/JDBCDemo/
mvn clean package assembly:single -DskipTests >/dev/null
java -jar target/JDBCDemo-SNAPSHOT-jar-with-dependencies.jar -host 127.0.0.1
'''
}
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
sh '''
cd ${WKC}/src/connector/jdbc
mvn clean package -Dmaven.test.skip=true >/dev/null
cd ${WKC}/tests/examples/JDBC/JDBCDemo/
java --class-path=../../../../src/connector/jdbc/target:$JAVA_HOME/jre/lib/ext -jar target/JDBCDemo-SNAPSHOT-jar-with-dependencies.jar -host 127.0.0.1
'''
}
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
sh '''
cp -rf ${WKC}/tests/examples/nodejs ${JENKINS_HOME}/workspace/
cd ${JENKINS_HOME}/workspace/nodejs
node nodejsChecker.js host=localhost
'''
}
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
sh '''
cd ${JENKINS_HOME}/workspace/C#NET/src/CheckC#
dotnet run
'''
}
sh '''
systemctl stop taosd
cd ${WKC}/tests
./test-all.sh b2
date
'''
sh '''
cd ${WKC}/tests
./test-all.sh full unit
date'''
}
}
stage('test_valgrind') {
agent{label "slam4"}
steps {
pre_test()
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
sh '''
cd ${WKC}/tests/pytest
nohup taosd >/dev/null &
sleep 10
python3 concurrent_inquiry.py -c 1
'''
}
sh '''
cd ${WKC}/tests
./test-all.sh full jdbc
date'''
sh '''
cd ${WKC}/tests/pytest
./valgrind-test.sh 2>&1 > mem-error-out.log
./handle_val_log.sh
date
cd ${WKC}/tests
./test-all.sh b3
date'''
sh '''
date
cd ${WKC}/tests
./test-all.sh full example
date'''
}
}
stage('arm64_build'){
agent{label 'arm64'}
steps{
sh '''
cd ${WK}
git fetch
git checkout develop
git pull
cd ${WKC}
git fetch
git checkout develop
git pull
git submodule update
cd ${WKC}/packaging
./release.sh -v cluster -c aarch64 -n 2.0.0.0 -m 2.0.0.0
'''
}
}
stage('arm32_build'){
agent{label 'arm32'}
steps{
catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') {
sh '''
cd ${WK}
git fetch
git checkout develop
git pull
cd ${WKC}
git fetch
git checkout develop
git pull
git submodule update
cd ${WKC}/packaging
./release.sh -v cluster -c aarch32 -n 2.0.0.0 -m 2.0.0.0
'''
}
}
}
}
}
}
post {
success {
emailext (
subject: "SUCCESSFUL: Job '${env.JOB_NAME} [${env.BUILD_NUMBER}]'",
body: '''<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
</head>
<body leftmargin="8" marginwidth="0" topmargin="8" marginheight="4" offset="0">
<table width="95%" cellpadding="0" cellspacing="0" style="font-size: 16pt; font-family: Tahoma, Arial, Helvetica, sans-serif">
<tr>
<td><br />
<b><font color="#0B610B"><font size="6">构建信息</font></font></b>
<hr size="2" width="100%" align="center" /></td>
</tr>
<tr>
<td>
<ul>
<div style="font-size:18px">
<li>构建名称>>分支：${PROJECT_NAME}</li>
<li>构建结果：<span style="color:green"> Successful </span></li>
<li>构建编号：${BUILD_NUMBER}</li>
<li>触发用户：${CAUSE}</li>
<li>变更概要：${CHANGES}</li>
<li>构建地址：<a href=${BUILD_URL}>${BUILD_URL}</a></li>
<li>构建日志：<a href=${BUILD_URL}console>${BUILD_URL}console</a></li>
<li>变更集：${JELLY_SCRIPT}</li>
</div>
</ul>
</td>
</tr>
</table></font>
</body>
</html>''',
to: "yqliu@taosdata.com,pxiao@taosdata.com",
from: "support@taosdata.com"
)
}
failure {
emailext (
subject: "FAILED: Job '${env.JOB_NAME} [${env.BUILD_NUMBER}]'",
body: '''<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
</head>
<body leftmargin="8" marginwidth="0" topmargin="8" marginheight="4" offset="0">
<table width="95%" cellpadding="0" cellspacing="0" style="font-size: 16pt; font-family: Tahoma, Arial, Helvetica, sans-serif">
<tr>
<td><br />
<b><font color="#0B610B"><font size="6">构建信息</font></font></b>
<hr size="2" width="100%" align="center" /></td>
</tr>
<tr>
<td>
<ul>
<div style="font-size:18px">
<li>构建名称>>分支：${PROJECT_NAME}</li>
<li>构建结果：<span style="color:green"> Successful </span></li>
<li>构建编号：${BUILD_NUMBER}</li>
<li>触发用户：${CAUSE}</li>
<li>变更概要：${CHANGES}</li>
<li>构建地址：<a href=${BUILD_URL}>${BUILD_URL}</a></li>
<li>构建日志：<a href=${BUILD_URL}console>${BUILD_URL}console</a></li>
<li>变更集：${JELLY_SCRIPT}</li>
</div>
</ul>
</td>
</tr>
</table></font>
</body>
</html>''',
to: "yqliu@taosdata.com,pxiao@taosdata.com",
from: "support@taosdata.com"
)
}
}
}
\ No newline at end of file
tests/perftest-scripts/perftest-query.sh
浏览文件 @ 9a4ccd2c
...@@ -64,18 +64,25 @@ function runQueryPerfTest { ...@@ -64,18 +64,25 @@ function runQueryPerfTest {
[ -f $PERFORMANCE_TEST_REPORT ] && rm $PERFORMANCE_TEST_REPORT [ -f $PERFORMANCE_TEST_REPORT ] && rm $PERFORMANCE_TEST_REPORT
nohup $WORK_DIR/TDengine/debug/build/bin/taosd -c /etc/taosperf/ > /dev/null 2>&1 & nohup $WORK_DIR/TDengine/debug/build/bin/taosd -c /etc/taosperf/ > /dev/null 2>&1 &
echoInfo "Wait TDengine to start" echoInfo "Wait TDengine to start"
sleep 300 sleep 60
echoInfo "Run Performance Test" echoInfo "Run Performance Test"
cd $WORK_DIR/TDengine/tests/pytest cd $WORK_DIR/TDengine/tests/pytest
python3 query/queryPerformance.py -c $LOCAL_COMMIT | tee -a $PERFORMANCE_TEST_REPORT python3 query/queryPerformance.py -c $LOCAL_COMMIT | tee -a $PERFORMANCE_TEST_REPORT
mkdir -p /var/lib/perf/
mkdir -p /var/log/perf/
rm -rf /var/lib/perf/*
rm -rf /var/log/perf/*
nohup $WORK_DIR/TDengine/debug/build/bin/taosd -c /etc/perf/ > /dev/null 2>&1 &
echoInfo "Wait TDengine to start"
sleep 10
echoInfo "Run Performance Test"
cd $WORK_DIR/TDengine/tests/pytest
python3 insert/insertFromCSVPerformance.py -c $LOCAL_COMMIT | tee -a $PERFORMANCE_TEST_REPORT python3 insert/insertFromCSVPerformance.py -c $LOCAL_COMMIT | tee -a $PERFORMANCE_TEST_REPORT
python3 tools/taosdemoPerformance.py -c $LOCAL_COMMIT | tee -a $PERFORMANCE_TEST_REPORT python3 tools/taosdemoPerformance.py -c $LOCAL_COMMIT | tee -a $PERFORMANCE_TEST_REPORT
#python3 perfbenchmark/joinPerformance.py | tee -a $PERFORMANCE_TEST_REPORT
} }
......
tests/pytest/crash_gen/service_manager.py
浏览文件 @ 9a4ccd2c
...@@ -22,7 +22,7 @@ from queue import Queue, Empty ...@@ -22,7 +22,7 @@ from queue import Queue, Empty
from .shared.config import Config from .shared.config import Config
from .shared.db import DbTarget, DbConn from .shared.db import DbTarget, DbConn
from .shared.misc import Logging, Helper, CrashGenError, Status, Progress, Dice from .shared.misc import Logging, Helper, CrashGenError, Status, Progress, Dice
from .shared.types import DirPath from .shared.types import DirPath, IpcStream
# from crash_gen.misc import CrashGenError, Dice, Helper, Logging, Progress, Status # from crash_gen.misc import CrashGenError, Dice, Helper, Logging, Progress, Status
# from crash_gen.db import DbConn, DbTarget # from crash_gen.db import DbConn, DbTarget
...@@ -177,13 +177,12 @@ quorum 2 ...@@ -177,13 +177,12 @@ quorum 2
return "127.0.0.1" return "127.0.0.1"
def getServiceCmdLine(self): # to start the instance def getServiceCmdLine(self): # to start the instance
cmdLine = []
if Config.getConfig().track_memory_leaks: if Config.getConfig().track_memory_leaks:
Logging.info("Invoking VALGRIND on service...") Logging.info("Invoking VALGRIND on service...")
cmdLine = ['valgrind', '--leak-check=yes'] return ['exec /usr/bin/valgrind', '--leak-check=yes', self.getExecFile(), '-c', self.getCfgDir()]
else:
# TODO: move "exec -c" into Popen(), we can both "use shell" and NOT fork so ask to lose kill control # TODO: move "exec -c" into Popen(), we can both "use shell" and NOT fork so ask to lose kill control
cmdLine += ["exec " + self.getExecFile(), '-c', self.getCfgDir()] # used in subproce.Popen() return ["exec " + self.getExecFile(), '-c', self.getCfgDir()] # used in subproce.Popen()
return cmdLine
def _getDnodes(self, dbc): def _getDnodes(self, dbc):
dbc.query("show dnodes") dbc.query("show dnodes")
...@@ -281,16 +280,16 @@ class TdeSubProcess: ...@@ -281,16 +280,16 @@ class TdeSubProcess:
return '[TdeSubProc: pid = {}, status = {}]'.format( return '[TdeSubProc: pid = {}, status = {}]'.format(
self.getPid(), self.getStatus() ) self.getPid(), self.getStatus() )
def getStdOut(self) -> BinaryIO : def getIpcStdOut(self) -> IpcStream :
if self._popen.universal_newlines : # alias of text_mode if self._popen.universal_newlines : # alias of text_mode
raise CrashGenError("We need binary mode for STDOUT IPC") raise CrashGenError("We need binary mode for STDOUT IPC")
# Logging.info("Type of stdout is: {}".format(type(self._popen.stdout))) # Logging.info("Type of stdout is: {}".format(type(self._popen.stdout)))
return typing.cast(BinaryIO, self._popen.stdout) return typing.cast(IpcStream, self._popen.stdout)
def getStdErr(self) -> BinaryIO : def getIpcStdErr(self) -> IpcStream :
if self._popen.universal_newlines : # alias of text_mode if self._popen.universal_newlines : # alias of text_mode
raise CrashGenError("We need binary mode for STDERR IPC") raise CrashGenError("We need binary mode for STDERR IPC")
return typing.cast(BinaryIO, self._popen.stderr) return typing.cast(IpcStream, self._popen.stderr)
# Now it's always running, since we matched the life cycle # Now it's always running, since we matched the life cycle
# def isRunning(self): # def isRunning(self):
...@@ -302,11 +301,6 @@ class TdeSubProcess: ...@@ -302,11 +301,6 @@ class TdeSubProcess:
def _start(self, cmdLine) -> Popen : def _start(self, cmdLine) -> Popen :
ON_POSIX = 'posix' in sys.builtin_module_names ON_POSIX = 'posix' in sys.builtin_module_names
# Sanity check
# if self.subProcess: # already there
# raise RuntimeError("Corrupt process state")
# Prepare environment variables for coverage information # Prepare environment variables for coverage information
# Ref: https://stackoverflow.com/questions/2231227/python-subprocess-popen-with-a-modified-environment # Ref: https://stackoverflow.com/questions/2231227/python-subprocess-popen-with-a-modified-environment
myEnv = os.environ.copy() myEnv = os.environ.copy()
...@@ -314,9 +308,8 @@ class TdeSubProcess: ...@@ -314,9 +308,8 @@ class TdeSubProcess:
# print(myEnv) # print(myEnv)
# print("Starting TDengine with env: ", myEnv.items()) # print("Starting TDengine with env: ", myEnv.items())
# print("Starting TDengine via Shell: {}".format(cmdLineStr)) print("Starting TDengine: {}".format(cmdLine))
# useShell = True # Needed to pass environments into it
return Popen( return Popen(
' '.join(cmdLine), # ' '.join(cmdLine) if useShell else cmdLine, ' '.join(cmdLine), # ' '.join(cmdLine) if useShell else cmdLine,
shell=True, # Always use shell, since we need to pass ENV vars shell=True, # Always use shell, since we need to pass ENV vars
...@@ -732,19 +725,19 @@ class ServiceManagerThread: ...@@ -732,19 +725,19 @@ class ServiceManagerThread:
self._ipcQueue = Queue() # type: Queue self._ipcQueue = Queue() # type: Queue
self._thread = threading.Thread( # First thread captures server OUTPUT self._thread = threading.Thread( # First thread captures server OUTPUT
target=self.svcOutputReader, target=self.svcOutputReader,
args=(subProc.getStdOut(), self._ipcQueue, logDir)) args=(subProc.getIpcStdOut(), self._ipcQueue, logDir))
self._thread.daemon = True # thread dies with the program self._thread.daemon = True # thread dies with the program
self._thread.start() self._thread.start()
time.sleep(0.01) time.sleep(0.01)
if not self._thread.is_alive(): # What happened? if not self._thread.is_alive(): # What happened?
Logging.info("Failed to started process to monitor STDOUT") Logging.info("Failed to start process to monitor STDOUT")
self.stop() self.stop()
raise CrashGenError("Failed to start thread to monitor STDOUT") raise CrashGenError("Failed to start thread to monitor STDOUT")
Logging.info("Successfully started process to monitor STDOUT") Logging.info("Successfully started process to monitor STDOUT")
self._thread2 = threading.Thread( # 2nd thread captures server ERRORs self._thread2 = threading.Thread( # 2nd thread captures server ERRORs
target=self.svcErrorReader, target=self.svcErrorReader,
args=(subProc.getStdErr(), self._ipcQueue, logDir)) args=(subProc.getIpcStdErr(), self._ipcQueue, logDir))
self._thread2.daemon = True # thread dies with the program self._thread2.daemon = True # thread dies with the program
self._thread2.start() self._thread2.start()
time.sleep(0.01) time.sleep(0.01)
...@@ -887,14 +880,19 @@ class ServiceManagerThread: ...@@ -887,14 +880,19 @@ class ServiceManagerThread:
print("\nNon-UTF8 server output: {}\n".format(bChunk.decode('cp437'))) print("\nNon-UTF8 server output: {}\n".format(bChunk.decode('cp437')))
return None return None
def _textChunkGenerator(self, streamIn: BinaryIO, logDir: str, logFile: str def _textChunkGenerator(self, streamIn: IpcStream, logDir: str, logFile: str
) -> Generator[TextChunk, None, None]: ) -> Generator[TextChunk, None, None]:
''' '''
Take an input stream with binary data, produced a generator of decoded Take an input stream with binary data (likely from Popen), produced a generator of decoded
"text chunks", and also save the original binary data in a log file. "text chunks".
Side effect: it also save the original binary data in a log file.
''' '''
os.makedirs(logDir, exist_ok=True) os.makedirs(logDir, exist_ok=True)
logF = open(os.path.join(logDir, logFile), 'wb') logF = open(os.path.join(logDir, logFile), 'wb')
if logF is None:
Logging.error("Failed to open log file (binary write): {}/{}".format(logDir, logFile))
return
for bChunk in iter(streamIn.readline, b''): for bChunk in iter(streamIn.readline, b''):
logF.write(bChunk) # Write to log file immediately logF.write(bChunk) # Write to log file immediately
tChunk = self._decodeBinaryChunk(bChunk) # decode tChunk = self._decodeBinaryChunk(bChunk) # decode
...@@ -902,14 +900,14 @@ class ServiceManagerThread: ...@@ -902,14 +900,14 @@ class ServiceManagerThread:
yield tChunk # TODO: split into actual text lines yield tChunk # TODO: split into actual text lines
# At the end... # At the end...
streamIn.close() # Close the stream streamIn.close() # Close the incoming stream
logF.close() # Close the output file logF.close() # Close the log file
def svcOutputReader(self, stdOut: BinaryIO, queue, logDir: str): def svcOutputReader(self, ipcStdOut: IpcStream, queue, logDir: str):
''' '''
The infinite routine that processes the STDOUT stream for the sub process being managed. The infinite routine that processes the STDOUT stream for the sub process being managed.
:param stdOut: the IO stream object used to fetch the data from :param ipcStdOut: the IO stream object used to fetch the data from
:param queue: the queue where we dump the roughly parsed chunk-by-chunk text data :param queue: the queue where we dump the roughly parsed chunk-by-chunk text data
:param logDir: where we should dump a verbatim output file :param logDir: where we should dump a verbatim output file
''' '''
...@@ -917,7 +915,7 @@ class ServiceManagerThread: ...@@ -917,7 +915,7 @@ class ServiceManagerThread:
# Important Reference: https://stackoverflow.com/questions/375427/non-blocking-read-on-a-subprocess-pipe-in-python # Important Reference: https://stackoverflow.com/questions/375427/non-blocking-read-on-a-subprocess-pipe-in-python
# print("This is the svcOutput Reader...") # print("This is the svcOutput Reader...")
# stdOut.readline() # Skip the first output? TODO: remove? # stdOut.readline() # Skip the first output? TODO: remove?
for tChunk in self._textChunkGenerator(stdOut, logDir, 'stdout.log') : for tChunk in self._textChunkGenerator(ipcStdOut, logDir, 'stdout.log') :
queue.put(tChunk) # tChunk garanteed not to be None queue.put(tChunk) # tChunk garanteed not to be None
self._printProgress("_i") self._printProgress("_i")
...@@ -940,12 +938,12 @@ class ServiceManagerThread: ...@@ -940,12 +938,12 @@ class ServiceManagerThread:
Logging.info("EOF found TDengine STDOUT, marking the process as terminated") Logging.info("EOF found TDengine STDOUT, marking the process as terminated")
self.setStatus(Status.STATUS_STOPPED) self.setStatus(Status.STATUS_STOPPED)
def svcErrorReader(self, stdErr: BinaryIO, queue, logDir: str): def svcErrorReader(self, ipcStdErr: IpcStream, queue, logDir: str):
# os.makedirs(logDir, exist_ok=True) # os.makedirs(logDir, exist_ok=True)
# logFile = os.path.join(logDir,'stderr.log') # logFile = os.path.join(logDir,'stderr.log')
# fErr = open(logFile, 'wb') # fErr = open(logFile, 'wb')
# for line in iter(err.readline, b''): # for line in iter(err.readline, b''):
for tChunk in self._textChunkGenerator(stdErr, logDir, 'stderr.log') : for tChunk in self._textChunkGenerator(ipcStdErr, logDir, 'stderr.log') :
queue.put(tChunk) # tChunk garanteed not to be None queue.put(tChunk) # tChunk garanteed not to be None
# fErr.write(line) # fErr.write(line)
Logging.info("TDengine STDERR: {}".format(tChunk)) Logging.info("TDengine STDERR: {}".format(tChunk))
......
tests/pytest/crash_gen/shared/types.py
浏览文件 @ 9a4ccd2c
from typing import Any, List, Dict, NewType from typing import Any, BinaryIO, List, Dict, NewType
from enum import Enum from enum import Enum
DirPath = NewType('DirPath', str) DirPath = NewType('DirPath', str)
...@@ -26,3 +26,5 @@ class TdDataType(Enum): ...@@ -26,3 +26,5 @@ class TdDataType(Enum):
TdColumns = Dict[str, TdDataType] TdColumns = Dict[str, TdDataType]
TdTags = Dict[str, TdDataType] TdTags = Dict[str, TdDataType]
IpcStream = NewType('IpcStream', BinaryIO)
\ No newline at end of file
tests/pytest/fulltest.sh
浏览文件 @ 9a4ccd2c
...@@ -183,7 +183,7 @@ python3 ./test.py -f stable/query_after_reset.py ...@@ -183,7 +183,7 @@ python3 ./test.py -f stable/query_after_reset.py
# perfbenchmark # perfbenchmark
python3 ./test.py -f perfbenchmark/bug3433.py python3 ./test.py -f perfbenchmark/bug3433.py
#python3 ./test.py -f perfbenchmark/bug3589.py #python3 ./test.py -f perfbenchmark/bug3589.py
python3 ./test.py -f perfbenchmark/taosdemoInsert.py
#query #query
python3 ./test.py -f query/filter.py python3 ./test.py -f query/filter.py
......
tests/pytest/insert/insertFromCSVPerformance.py
浏览文件 @ 9a4ccd2c
...@@ -31,7 +31,7 @@ class insertFromCSVPerformace: ...@@ -31,7 +31,7 @@ class insertFromCSVPerformace:
self.host = "127.0.0.1" self.host = "127.0.0.1"
self.user = "root" self.user = "root"
self.password = "taosdata" self.password = "taosdata"
self.config = "/etc/taosperf" self.config = "/etc/perf"
self.conn = taos.connect( self.conn = taos.connect(
self.host, self.host,
self.user, self.user,
......
tests/pytest/perfbenchmark/taosdemoInsert.py 0 → 100644
浏览文件 @ 9a4ccd2c
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import taos
import sys
import os
import json
import argparse
import subprocess
import datetime
import re
from multiprocessing import cpu_count
from util.log import *
from util.sql import *
from util.cases import *
from util.dnodes import *
from util.dnodes import TDDnode
class Taosdemo:
def __init__(self, clearCache, dbName, keep):
self.clearCache = clearCache
self.dbname = dbName
self.drop = "yes"
self.keep = keep
self.host = "127.0.0.1"
self.user = "root"
self.password = "taosdata"
# self.config = "/etc/taosperf"
# self.conn = taos.connect(
# self.host,
# self.user,
# self.password,
# self.config)
# env config
def getBuildPath(self) -> str:
selfPath = os.path.dirname(os.path.realpath(__file__))
if ("community" in selfPath):
projPath = selfPath[:selfPath.find("community")]
else:
projPath = selfPath[:selfPath.find("tests")]
for root, dirs, files in os.walk(projPath):
if ("taosd" in files):
rootRealPath = os.path.dirname(os.path.realpath(root))
if ("packaging" not in rootRealPath):
buildPath = root[:len(root) - len("/debug/build/bin")]
break
return buildPath
def getExeToolsDir(self) -> str:
self.debugdir = self.getBuildPath() + "/debug/build/bin"
return self.debugdir
def getCfgDir(self) -> str:
self.config = self.getBuildPath() + "/sim/dnode1/cfg"
return self.config
# taodemo insert file config
def dbinfocfg(self) -> dict:
return {
"name": self.dbname,
"drop": self.drop,
"replica": 1,
"days": 10,
"cache": 16,
"blocks": 8,
"precision": "ms",
"keep": self.keep,
"minRows": 100,
"maxRows": 4096,
"comp": 2,
"walLevel": 1,
"cachelast": 0,
"quorum": 1,
"fsync": 3000,
"update": 0
}
def type_check(func):
def wrapper(self, **kwargs):
num_types = ["int", "float", "bigint", "tinyint", "smallint", "double"]
str_types = ["binary", "nchar"]
for k ,v in kwargs.items():
if k.lower() not in num_types and k.lower() not in str_types:
return f"args {k} type error, not allowed"
elif not isinstance(v, (int, list, tuple)):
return f"value {v} type error, not allowed"
elif k.lower() in num_types and not isinstance(v, int):
return f"arg {v} takes 1 positional argument must be type int "
elif isinstance(v, (list,tuple)) and len(v) > 2:
return f"arg {v} takes from 1 to 2 positional arguments but more than 2 were given "
elif isinstance(v,(list,tuple)) and [ False for _ in v if not isinstance(_, int) ]:
return f"arg {v} takes from 1 to 2 positional arguments must be type int "
else:
pass
return func(self, **kwargs)
return wrapper
@type_check
def column_tag_count(self, **column_tag) -> list :
init_column_tag = []
for k, v in column_tag.items():
if re.search(k, "int, float, bigint, tinyint, smallint, double", re.IGNORECASE):
init_column_tag.append({"type": k, "count": v})
elif re.search(k, "binary, nchar", re.IGNORECASE):
if isinstance(v, int):
init_column_tag.append({"type": k, "count": v, "len":8})
elif len(v) == 1:
init_column_tag.append({"type": k, "count": v[0], "len": 8})
else:
init_column_tag.append({"type": k, "count": v[0], "len": v[1]})
return init_column_tag
def stbcfg(self, stb: str, child_tab_count: int, rows: int, prechildtab: str, columns: dict, tags: dict) -> dict:
return {
"name": stb,
"child_table_exists": "no",
"childtable_count": child_tab_count,
"childtable_prefix": prechildtab,
"auto_create_table": "no",
"batch_create_tbl_num": 10,
"data_source": "rand",
"insert_mode": "taosc",
"insert_rows": rows,
"childtable_limit": 0,
"childtable_offset": 0,
"rows_per_tbl": 1,
"max_sql_len": 65480,
"disorder_ratio": 0,
"disorder_range": 1000,
"timestamp_step": 10,
"start_timestamp": f"{datetime.datetime.now():%F %X}",
"sample_format": "csv",
"sample_file": "./sample.csv",
"tags_file": "",
"columns": self.column_tag_count(**columns),
"tags": self.column_tag_count(**tags)
}
def schemecfg(self,intcount=1,floatcount=0,bcount=0,tcount=0,scount=0,doublecount=0,binarycount=0,ncharcount=0):
return {
"INT": intcount,
"FLOAT": floatcount,
"BIGINT": bcount,
"TINYINT": tcount,
"SMALLINT": scount,
"DOUBLE": doublecount,
"BINARY": binarycount,
"NCHAR": ncharcount
}
def insertcfg(self,db: dict, stbs: list) -> dict:
return {
"filetype": "insert",
"cfgdir": self.config,
"host": self.host,
"port": 6030,
"user": self.user,
"password": self.password,
"thread_count": cpu_count(),
"thread_count_create_tbl": cpu_count(),
"result_file": "/tmp/insert_res.txt",
"confirm_parameter_prompt": "no",
"insert_interval": 0,
"num_of_records_per_req": 100,
"max_sql_len": 1024000,
"databases": [{
"dbinfo": db,
"super_tables": stbs
}]
}
def createinsertfile(self,db: dict, stbs: list) -> str:
date = datetime.datetime.now()
file_create_table = f"/tmp/insert_{date:%F-%H%M}.json"
with open(file_create_table, 'w') as f:
json.dump(self.insertcfg(db, stbs), f)
return file_create_table
# taosdemo query file config
def querysqls(self, sql: str) -> list:
return [{"sql":sql,"result":""}]
def querycfg(self, sql: str) -> dict:
return {
"filetype": "query",
"cfgdir": self.config,
"host": self.host,
"port": 6030,
"user": self.user,
"password": self.password,
"confirm_parameter_prompt": "yes",
"query_times": 10,
"query_mode": "taosc",
"databases": self.dbname,
"specified_table_query": {
"query_interval": 0,
"concurrent": cpu_count(),
"sqls": self.querysqls(sql)
}
}
def createqueryfile(self, sql: str):
date = datetime.datetime.now()
file_query_table = f"/tmp/query_{date:%F-%H%M}.json"
with open(file_query_table,"w") as f:
json.dump(self.querycfg(sql), f)
return file_query_table
# Execute taosdemo, and delete temporary files when finished
def taosdemotable(self, filepath: str, resultfile="/dev/null"):
taosdemopath = self.getBuildPath() + "/debug/build/bin"
with open(filepath,"r") as f:
filetype = json.load(f)["filetype"]
if filetype == "insert":
taosdemo_table_cmd = f"{taosdemopath}/taosdemo -f {filepath} > {resultfile} 2>&1"
else:
taosdemo_table_cmd = f"yes | {taosdemopath}/taosdemo -f {filepath} > {resultfile} 2>&1"
try:
_ = subprocess.check_output(taosdemo_table_cmd, shell=True).decode("utf-8")
except subprocess.CalledProcessError as e:
_ = e.output
def droptmpfile(self, filepath: str):
drop_file_cmd = f"[ -f {filepath} ] && rm -f {filepath}"
try:
_ = subprocess.check_output(drop_file_cmd, shell=True).decode("utf-8")
except subprocess.CalledProcessError as e:
_ = e.output
# TODO:需要完成TD-4153的数据插入和客户端请求的性能查询。
def td4153insert(self):
tdLog.printNoPrefix("========== start to create table and insert data ==========")
self.dbname = "td4153"
db = self.dbinfocfg()
stblist = []
columntype = self.schemecfg(intcount=1, ncharcount=100)
tagtype = self.schemecfg(intcount=1)
stbname = "stb1"
prechild = "t1"
stable = self.stbcfg(
stb=stbname,
prechildtab=prechild,
child_tab_count=2,
rows=10000,
columns=columntype,
tags=tagtype
)
stblist.append(stable)
insertfile = self.createinsertfile(db=db, stbs=stblist)
nmon_file = f"/tmp/insert_{datetime.datetime.now():%F-%H%M}.nmon"
cmd = f"nmon -s5 -F {nmon_file} -m /tmp/"
try:
_ = subprocess.check_output(cmd, shell=True).decode("utf-8")
except subprocess.CalledProcessError as e:
_ = e.output
self.taosdemotable(insertfile)
self.droptmpfile(insertfile)
self.droptmpfile("/tmp/insert_res.txt")
# In order to prevent too many performance files from being generated, the nmon file is deleted.
# and the delete statement can be cancelled during the actual test.
self.droptmpfile(nmon_file)
cmd = f"ps -ef|grep -w nmon| grep -v grep | awk '{{print $2}}'"
try:
time.sleep(10)
_ = subprocess.check_output(cmd,shell=True).decode("utf-8")
except BaseException as e:
raise e
def td4153query(self):
tdLog.printNoPrefix("========== start to query operation ==========")
sqls = {
"select_all": "select * from stb1",
"select_join": "select * from t10, t11 where t10.ts=t11.ts"
}
for type, sql in sqls.items():
result_file = f"/tmp/queryResult_{type}.log"
query_file = self.createqueryfile(sql)
try:
self.taosdemotable(query_file, resultfile=result_file)
except subprocess.CalledProcessError as e:
out_put = e.output
if result_file:
print(f"execute rows {type.split('_')[1]} sql, the sql is: {sql}")
max_sql_time_cmd = f'''
grep -o Spent.*s {result_file} |awk 'NR==1{{max=$2;next}}{{max=max>$2?max:$2}}END{{print "Max=",max,"s"}}'
'''
max_sql_time = subprocess.check_output(max_sql_time_cmd, shell=True).decode("UTF-8")
print(f"{type.split('_')[1]} rows sql time : {max_sql_time}")
min_sql_time_cmd = f'''
grep -o Spent.*s {result_file} |awk 'NR==1{{min=$2;next}}{{min=min<$2?min:$2}}END{{print "Min=",min,"s"}}'
'''
min_sql_time = subprocess.check_output(min_sql_time_cmd, shell=True).decode("UTF-8")
print(f"{type.split('_')[1]} rows sql time : {min_sql_time}")
avg_sql_time_cmd = f'''
grep -o Spent.*s {result_file} |awk '{{sum+=$2}}END{{print "Average=",sum/NR,"s"}}'
'''
avg_sql_time = subprocess.check_output(avg_sql_time_cmd, shell=True).decode("UTF-8")
print(f"{type.split('_')[1]} rows sql time : {avg_sql_time}")
self.droptmpfile(query_file)
self.droptmpfile(result_file)
drop_query_tmt_file_cmd = " find ./ -name 'querySystemInfo-*' -type f -exec rm {} \; "
try:
_ = subprocess.check_output(drop_query_tmt_file_cmd, shell=True).decode("utf-8")
except subprocess.CalledProcessError as e:
_ = e.output
pass
def td4153(self):
self.td4153insert()
self.td4153query()
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument(
'-r',
'--remove-cache',
action='store_true',
default=False,
help='clear cache before query (default: False)')
parser.add_argument(
'-d',
'--database-name',
action='store',
default='db',
type=str,
help='Database name to be created (default: db)')
parser.add_argument(
'-k',
'--keep-time',
action='store',
default=3650,
type=int,
help='Database keep parameters (default: 3650)')
args = parser.parse_args()
taosdemo = Taosdemo(args.remove_cache, args.database_name, args.keep_time)
# taosdemo.conn = taos.connect(
# taosdemo.host,
# taosdemo.user,
# taosdemo.password,
# taosdemo.config
# )
debugdir = taosdemo.getExeToolsDir()
cfgdir = taosdemo.getCfgDir()
cmd = f"{debugdir}/taosd -c {cfgdir} >/dev/null 2>&1 &"
try:
_ = subprocess.check_output(cmd, shell=True).decode("utf-8")
except subprocess.CalledProcessError as e:
_ = e.output
if taosdemo.clearCache:
# must be root permission
subprocess.check_output("echo 3 > /proc/sys/vm/drop_caches", shell=True).decode("utf-8")
taosdemo.td4153()
tests/pytest/tools/taosdemoPerformance.py
浏览文件 @ 9a4ccd2c
...@@ -24,7 +24,7 @@ class taosdemoPerformace: ...@@ -24,7 +24,7 @@ class taosdemoPerformace:
self.host = "127.0.0.1" self.host = "127.0.0.1"
self.user = "root" self.user = "root"
self.password = "taosdata" self.password = "taosdata"
self.config = "/etc/taosperf" self.config = "/etc/perf"
self.conn = taos.connect( self.conn = taos.connect(
self.host, self.host,
self.user, self.user,
...@@ -77,7 +77,7 @@ class taosdemoPerformace: ...@@ -77,7 +77,7 @@ class taosdemoPerformace:
insert_data = { insert_data = {
"filetype": "insert", "filetype": "insert",
"cfgdir": "/etc/taosperf", "cfgdir": "/etc/perf",
"host": "127.0.0.1", "host": "127.0.0.1",
"port": 6030, "port": 6030,
"user": "root", "user": "root",
......
tests/script/general/parser/select_with_tags.sim
浏览文件 @ 9a4ccd2c
...@@ -887,10 +887,16 @@ sql_error select tbname, t1 from select_tags_mt0 interval(1y); ...@@ -887,10 +887,16 @@ sql_error select tbname, t1 from select_tags_mt0 interval(1y);
#valid sql: select first(c1), last(c2), count(*) from select_tags_mt0 group by tbname, t1; #valid sql: select first(c1), last(c2), count(*) from select_tags_mt0 group by tbname, t1;
#valid sql: select first(c1), tbname, t1 from select_tags_mt0 group by t2; #valid sql: select first(c1), tbname, t1 from select_tags_mt0 group by t2;
print ==================================>TD-4231
sql_error select t1,tbname from select_tags_mt0 where c1<0
sql_error select t1,tbname from select_tags_mt0 where c1<0 and tbname in ('select_tags_tb12')
sql select tbname from select_tags_mt0 where tbname in ('select_tags_tb12');
sql_error select first(c1), last(c2), t1 from select_tags_mt0 group by tbname; sql_error select first(c1), last(c2), t1 from select_tags_mt0 group by tbname;
sql_error select first(c1), last(c2), tbname, t2 from select_tags_mt0 group by tbname; sql_error select first(c1), last(c2), tbname, t2 from select_tags_mt0 group by tbname;
sql_error select first(c1), count(*), t2, t1, tbname from select_tags_mt0 group by tbname; sql_error select first(c1), count(*), t2, t1, tbname from select_tags_mt0 group by tbname;
# this sql is valid: select first(c1), t2 from select_tags_mt0 group by tbname; #valid sql: select first(c1), t2 from select_tags_mt0 group by tbname;
#sql select first(ts), tbname from select_tags_mt0 group by tbname; #sql select first(ts), tbname from select_tags_mt0 group by tbname;
#sql select count(c1) from select_tags_mt0 where c1=99 group by tbname; #sql select count(c1) from select_tags_mt0 where c1=99 group by tbname;
......
tests/script/unique/arbitrator/dn3_mn1_vnode_nomaster.sim
浏览文件 @ 9a4ccd2c
...@@ -158,7 +158,7 @@ if $dnode4Vtatus != offline then ...@@ -158,7 +158,7 @@ if $dnode4Vtatus != offline then
sleep 2000 sleep 2000
goto wait_dnode4_vgroup_offline goto wait_dnode4_vgroup_offline
endi endi
if $dnode3Vtatus != master then if $dnode3Vtatus != unsynced then
sleep 2000 sleep 2000
goto wait_dnode4_vgroup_offline goto wait_dnode4_vgroup_offline
endi endi
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或