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

Merge branch 'develop' into hotfix/TD-2719

This commit is contained in:
dapan1121 2021-01-14 14:37:37 +08:00
parent c04a4f1712 52a524624d
commit 0f715a3f4c
125 changed files with 1888 additions and 812 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
.travis.yml
View File

@ -34,6 +34,7 @@ matrix:
- psmisc
- unixodbc
- unixodbc-dev
- mono-complete
before_script:
- export TZ=Asia/Harbin
@ -59,6 +60,18 @@ matrix:
pip3 install guppy3
pip3 install --user ${TRAVIS_BUILD_DIR}/src/connector/python/linux/python3/
cd ${TRAVIS_BUILD_DIR}/tests/examples/C#/taosdemo
mcs -out:taosdemo *.cs || travis_terminate $?
pkill -TERM -x taosd
fuser -k -n tcp 6030
sleep 1
${TRAVIS_BUILD_DIR}/debug/build/bin/taosd -c ${TRAVIS_BUILD_DIR}/debug/test/cfg > /dev/null &
sleep 5
mono taosdemo -Q DEFAULT -y || travis_terminate $?
pkill -KILL -x taosd
fuser -k -n tcp 6030
sleep 1
cd ${TRAVIS_BUILD_DIR}/tests
./test-all.sh smoke || travis_terminate $?
sleep 1
@ -74,6 +87,7 @@ matrix:
./valgrind-test.sh 2>&1 > mem-error-out.log
sleep 1
# Color setting
RED='\033[0;31m'
GREEN='\033[1;32m'

2
CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
IF (CMAKE_VERSION VERSION_LESS 3.0)
PROJECT(TDengine CXX)
SET(PROJECT_VERSION_MAJOR "${LIB_MAJOR_VERSION}")

2
deps/CMakeLists.txt vendored
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
ADD_SUBDIRECTORY(zlib-1.2.11)

2
deps/MQTT-C/CMakeLists.txt vendored
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
# MQTT-C build options
option(MQTT_C_OpenSSL_SUPPORT "Build MQTT-C with OpenSSL support?" OFF)

2
deps/MsvcLibX/CMakeLists.txt vendored
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
IF (TD_WINDOWS)

2
deps/iconv/CMakeLists.txt vendored
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
IF (TD_WINDOWS)

2
deps/pthread/CMakeLists.txt vendored
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
IF (TD_WINDOWS)

2
deps/regex/CMakeLists.txt vendored
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
IF (TD_WINDOWS)

2
deps/wepoll/CMakeLists.txt vendored
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
IF (TD_WINDOWS)

2
deps/zlib-1.2.11/CMakeLists.txt vendored
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
IF (TD_WINDOWS)

3
packaging/cfg/taos.cfg
View File

@ -72,9 +72,6 @@
# time interval of heart beat from shell to dnode, seconds
# shellActivityTimer 3
# time of keeping table meta data in cache, seconds
# tableMetaKeepTimer 7200
# minimum sliding window time, milli-second
# minSlidingTime 10

2
src/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
# Base compile

2
src/balance/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/mnode/inc)

2
src/client/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(inc)

4
src/client/inc/tscSubquery.h
View File

@ -43,9 +43,9 @@ TAOS_ROW doSetResultRowData(SSqlObj *pSql);
char *getArithmeticInputSrc(void *param, const char *name, int32_t colId);
void tscSpinLock(int32_t *lock);
void tscLockByThread(int64_t *lockedBy);
void tscSpinUnlock(int32_t *lock);
void tscUnlockByThread(int64_t *lockedBy);
#ifdef __cplusplus
}

9
src/client/inc/tsclient.h
View File

@ -300,8 +300,8 @@ typedef struct STscObj {
void * pTimer;
char user[TSDB_USER_LEN];
char pass[TSDB_KEY_LEN];
char acctId[TSDB_ACCT_LEN];
char db[TSDB_ACCT_LEN + TSDB_DB_NAME_LEN];
char acctId[TSDB_ACCT_ID_LEN];
char db[TSDB_ACCT_ID_LEN + TSDB_DB_NAME_LEN];
char sversion[TSDB_VERSION_LEN];
char writeAuth : 1;
char superAuth : 1;
@ -348,6 +348,11 @@ typedef struct SSqlObj {
SSubqueryState subState;
struct SSqlObj **pSubs;
int64_t metaRid;
int64_t svgroupRid;
int64_t squeryLock;
struct SSqlObj *prev, *next;
int64_t self;
} SSqlObj;

26
src/client/src/tscAsync.c
View File

@ -402,8 +402,10 @@ void tscAsyncResultOnError(SSqlObj *pSql) {
int tscSendMsgToServer(SSqlObj *pSql);
void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
SSqlObj *pSql = (SSqlObj *)param;
if (pSql == NULL || pSql->signature != pSql) return;
SSqlObj* pSql = (SSqlObj*)taosAcquireRef(tscObjRef, (int64_t)param);
if (pSql == NULL) return;
assert(pSql->signature == pSql && (int64_t)param == pSql->self);
SSqlCmd *pCmd = &pSql->cmd;
SSqlRes *pRes = &pSql->res;
@ -428,7 +430,8 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
code = tscGetTableMeta(pSql, pTableMetaInfo);
assert(code == TSDB_CODE_TSC_ACTION_IN_PROGRESS || code == TSDB_CODE_SUCCESS);
if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
taosReleaseRef(tscObjRef, pSql->self);
return;
}
@ -436,6 +439,7 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
// tscProcessSql can add error into async res
tscProcessSql(pSql);
taosReleaseRef(tscObjRef, pSql->self);
return;
} else { // continue to process normal async query
if (pCmd->parseFinished) {
@ -446,6 +450,7 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
assert(code == TSDB_CODE_TSC_ACTION_IN_PROGRESS || code == TSDB_CODE_SUCCESS);
if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
taosReleaseRef(tscObjRef, pSql->self);
return;
}
@ -458,6 +463,7 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
code = tsParseSql(pSql, true);
if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
taosReleaseRef(tscObjRef, pSql->self);
return;
} else if (code != TSDB_CODE_SUCCESS) {
goto _error;
@ -468,12 +474,14 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
tscProcessSql(pSql);
}
taosReleaseRef(tscObjRef, pSql->self);
return;
} else {
tscDebug("%p continue parse sql after get table meta", pSql);
code = tsParseSql(pSql, false);
if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
taosReleaseRef(tscObjRef, pSql->self);
return;
} else if (code != TSDB_CODE_SUCCESS) {
goto _error;
@ -483,12 +491,14 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
STableMetaInfo* pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
code = tscGetTableMeta(pSql, pTableMetaInfo);
if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
taosReleaseRef(tscObjRef, pSql->self);
return;
} else {
assert(code == TSDB_CODE_SUCCESS);
}
(*pSql->fp)(pSql->param, pSql, code);
taosReleaseRef(tscObjRef, pSql->self);
return;
}
@ -501,6 +511,7 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
code = tscGetTableMeta(pSql, pTableMetaInfo);
if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
taosReleaseRef(tscObjRef, pSql->self);
return;
} else if (code != TSDB_CODE_SUCCESS) {
goto _error;
@ -509,6 +520,7 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
if (UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) {
code = tscGetSTableVgroupInfo(pSql, pCmd->clauseIndex);
if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
taosReleaseRef(tscObjRef, pSql->self);
return;
} else if (code != TSDB_CODE_SUCCESS) {
goto _error;
@ -521,10 +533,16 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
}
(*pSql->fp)(pSql->param, pSql, code);
taosReleaseRef(tscObjRef, pSql->self);
return;
}
tscDoQuery(pSql);
taosReleaseRef(tscObjRef, pSql->self);
return;
_error:
@ -532,4 +550,6 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
pSql->res.code = code;
tscAsyncResultOnError(pSql);
}
taosReleaseRef(tscObjRef, pSql->self);
}

8
src/client/src/tscParseInsert.c
View File

@ -762,6 +762,8 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql) {
char *sql = *sqlstr;
pSql->cmd.autoCreated = false;
// get the token of specified table
index = 0;
tableToken = tStrGetToken(sql, &index, false, 0, NULL);
@ -945,11 +947,15 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql) {
SKVRow row = tdGetKVRowFromBuilder(&kvRowBuilder);
tdDestroyKVRowBuilder(&kvRowBuilder);
if (row == NULL) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
return tscInvalidSQLErrMsg(pCmd->payload, "tag value expected", NULL);
}
tdSortKVRowByColIdx(row);
pCmd->tagData.dataLen = kvRowLen(row);
if (pCmd->tagData.dataLen <= 0){
return tscInvalidSQLErrMsg(pCmd->payload, "tag value expected", NULL);
}
char* pTag = realloc(pCmd->tagData.data, pCmd->tagData.dataLen);
if (pTag == NULL) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;

7
src/client/src/tscSQLParser.c
View File

@ -221,7 +221,7 @@ static int32_t handlePassword(SSqlCmd* pCmd, SStrToken* pPwd) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg1);
}
if (pPwd->n >= TSDB_PASSWORD_LEN) {
if (pPwd->n >= TSDB_KEY_LEN) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg2);
}
@ -1242,7 +1242,7 @@ int32_t setObjFullName(char* fullName, const char* account, SStrToken* pDB, SStr
/* db name is not specified, the tableName dose not include db name */
if (pDB != NULL) {
if (pDB->n >= TSDB_ACCT_LEN + TSDB_DB_NAME_LEN || pDB->n == 0) {
if (pDB->n >= TSDB_ACCT_ID_LEN + TSDB_DB_NAME_LEN || pDB->n == 0) {
return TSDB_CODE_TSC_INVALID_SQL;
}
@ -6101,7 +6101,7 @@ void tscPrintSelectClause(SSqlObj* pSql, int32_t subClauseIndex) {
int32_t tmpLen = 0;
tmpLen =
sprintf(tmpBuf, "%s(uid:%" PRId64 ", %d)", aAggs[pExpr->functionId].aName, pExpr->uid, pExpr->colInfo.colId);
if (tmpLen + offset > totalBufSize) break;
if (tmpLen + offset >= totalBufSize - 1) break;
offset += sprintf(str + offset, "%s", tmpBuf);
@ -6110,6 +6110,7 @@ void tscPrintSelectClause(SSqlObj* pSql, int32_t subClauseIndex) {
}
}
assert(offset < totalBufSize);
str[offset] = ']';
tscDebug("%p select clause:%s", pSql, str);
}

37
src/client/src/tscServer.c
View File

@ -423,7 +423,7 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
if (shouldFree) { // in case of table-meta/vgrouplist query, automatically free it
taosRemoveRef(tscObjRef, pSql->self);
tscDebug("%p sqlObj is automatically freed", pSql);
tscDebug("%p sqlObj is automatically freed", pSql);
}
rpcFreeCont(rpcMsg->pCont);
@ -1975,9 +1975,6 @@ int tscProcessMultiMeterMetaRsp(SSqlObj *pSql) {
//
// rsp += tagLen;
// int32_t size = (int32_t)(rsp - ((char *)pMeta)); // Consistent with STableMeta in cache
//
// pMeta->index = 0;
// (void)taosCachePut(tscMetaCache, pMeta->tableId, (char *)pMeta, size, tsTableMetaKeepTimer);
// }
}
@ -1990,16 +1987,20 @@ int tscProcessMultiMeterMetaRsp(SSqlObj *pSql) {
}
int tscProcessSTableVgroupRsp(SSqlObj *pSql) {
// master sqlObj locates in param
SSqlObj* parent = (SSqlObj*)taosAcquireRef(tscObjRef, (int64_t)pSql->param);
if(parent == NULL) {
return pSql->res.code;
}
assert(parent->signature == parent && (int64_t)pSql->param == parent->self);
SSqlRes* pRes = &pSql->res;
// NOTE: the order of several table must be preserved.
SSTableVgroupRspMsg *pStableVgroup = (SSTableVgroupRspMsg *)pRes->pRsp;
pStableVgroup->numOfTables = htonl(pStableVgroup->numOfTables);
char *pMsg = pRes->pRsp + sizeof(SSTableVgroupRspMsg);
// master sqlObj locates in param
SSqlObj* parent = pSql->param;
assert(parent != NULL);
SSqlCmd* pCmd = &parent->cmd;
for(int32_t i = 0; i < pStableVgroup->numOfTables; ++i) {
@ -2033,6 +2034,8 @@ int tscProcessSTableVgroupRsp(SSqlObj *pSql) {
pMsg += size;
}
taosReleaseRef(tscObjRef, parent->self);
return pSql->res.code;
}
@ -2326,11 +2329,15 @@ static int32_t getTableMetaFromMnode(SSqlObj *pSql, STableMetaInfo *pTableMetaIn
tscDebug("%p new pSqlObj:%p to get tableMeta, auto create:%d", pSql, pNew, pNew->cmd.autoCreated);
pNew->fp = tscTableMetaCallBack;
pNew->param = pSql;
registerSqlObj(pNew);
pNew->fp = tscTableMetaCallBack;
pNew->param = (void *)pSql->self;
tscDebug("%p metaRid from %" PRId64 " to %" PRId64 , pSql, pSql->metaRid, pNew->self);
pSql->metaRid = pNew->self;
int32_t code = tscProcessSql(pNew);
if (code == TSDB_CODE_SUCCESS) {
code = TSDB_CODE_TSC_ACTION_IN_PROGRESS; // notify application that current process needs to be terminated
@ -2346,6 +2353,7 @@ int32_t tscGetTableMeta(SSqlObj *pSql, STableMetaInfo *pTableMetaInfo) {
uint32_t size = tscGetTableMetaMaxSize();
pTableMetaInfo->pTableMeta = calloc(1, size);
pTableMetaInfo->pTableMeta->tableType = -1;
pTableMetaInfo->pTableMeta->tableInfo.numOfColumns = -1;
int32_t len = (int32_t) strlen(pTableMetaInfo->name);
@ -2445,10 +2453,15 @@ int tscGetSTableVgroupInfo(SSqlObj *pSql, int32_t clauseIndex) {
pNewQueryInfo->numOfTables = pQueryInfo->numOfTables;
registerSqlObj(pNew);
tscDebug("%p svgroupRid from %" PRId64 " to %" PRId64 , pSql, pSql->svgroupRid, pNew->self);
pSql->svgroupRid = pNew->self;
tscDebug("%p new sqlObj:%p to get vgroupInfo, numOfTables:%d", pSql, pNew, pNewQueryInfo->numOfTables);
pNew->fp = tscTableMetaCallBack;
pNew->param = pSql;
pNew->param = (void *)pSql->self;
code = tscProcessSql(pNew);
if (code == TSDB_CODE_SUCCESS) {
code = TSDB_CODE_TSC_ACTION_IN_PROGRESS;

16
src/client/src/tscSql.c
View File

@ -47,7 +47,7 @@ static bool validUserName(const char* user) {
}
static bool validPassword(const char* passwd) {
return validImpl(passwd, TSDB_PASSWORD_LEN - 1);
return validImpl(passwd, TSDB_KEY_LEN - 1);
}
static SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pass, const char *auth, const char *db,
@ -238,11 +238,11 @@ TAOS *taos_connect_c(const char *ip, uint8_t ipLen, const char *user, uint8_t us
uint8_t passLen, const char *db, uint8_t dbLen, uint16_t port) {
char ipBuf[TSDB_EP_LEN] = {0};
char userBuf[TSDB_USER_LEN] = {0};
char passBuf[TSDB_PASSWORD_LEN] = {0};
char passBuf[TSDB_KEY_LEN] = {0};
char dbBuf[TSDB_DB_NAME_LEN] = {0};
strncpy(ipBuf, ip, MIN(TSDB_EP_LEN - 1, ipLen));
strncpy(userBuf, user, MIN(TSDB_USER_LEN - 1, userLen));
strncpy(passBuf, pass, MIN(TSDB_PASSWORD_LEN - 1, passLen));
strncpy(passBuf, pass, MIN(TSDB_KEY_LEN - 1, passLen));
strncpy(dbBuf, db, MIN(TSDB_DB_NAME_LEN - 1, dbLen));
return taos_connect(ipBuf, userBuf, passBuf, dbBuf, port);
}
@ -694,6 +694,8 @@ static void tscKillSTableQuery(SSqlObj *pSql) {
// set the master sqlObj flag to cancel query
pSql->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
tscLockByThread(&pSql->squeryLock);
for (int i = 0; i < pSql->subState.numOfSub; ++i) {
// NOTE: pSub may have been released already here
SSqlObj *pSub = pSql->pSubs[i];
@ -713,6 +715,12 @@ static void tscKillSTableQuery(SSqlObj *pSql) {
taosReleaseRef(tscObjRef, pSubObj->self);
}
if (pSql->subState.numOfSub <= 0) {
tscAsyncResultOnError(pSql);
}
tscUnlockByThread(&pSql->squeryLock);
tscDebug("%p super table query cancelled", pSql);
}
@ -978,7 +986,7 @@ static int tscParseTblNameList(SSqlObj *pSql, const char *tblNameList, int32_t t
return code;
}
if (++pCmd->count > TSDB_MULTI_METERMETA_MAX_NUM) {
if (++pCmd->count > TSDB_MULTI_TABLEMETA_MAX_NUM) {
code = TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH;
sprintf(pCmd->payload, "tables over the max number");
return code;

6
src/client/src/tscStream.c
View File

@ -551,7 +551,6 @@ static void tscCreateStream(void *param, TAOS_RES *res, int code) {
int64_t starttime = tscGetLaunchTimestamp(pStream);
pCmd->command = TSDB_SQL_SELECT;
registerSqlObj(pSql);
tscAddIntoStreamList(pStream);
taosTmrReset(tscProcessStreamTimer, (int32_t)starttime, pStream, tscTmr, &pStream->pTimer);
@ -610,12 +609,15 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
pSql->fp = tscCreateStream;
pSql->fetchFp = tscCreateStream;
registerSqlObj(pSql);
int32_t code = tsParseSql(pSql, true);
if (code == TSDB_CODE_SUCCESS) {
tscCreateStream(pStream, pSql, code);
} else if (code != TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
tscError("%p open stream failed, sql:%s, code:%s", pSql, sqlstr, tstrerror(pRes->code));
tscFreeSqlObj(pSql);
taosReleaseRef(tscObjRef, pSql->self);
free(pStream);
return NULL;
}

27
src/client/src/tscSubquery.c
View File

@ -1429,15 +1429,18 @@ void tscJoinQueryCallback(void* param, TAOS_RES* tres, int code) {
SJoinSupporter* pSupporter = (SJoinSupporter*)param;
SSqlObj* pParentSql = pSupporter->pObj;
// There is only one subquery and table for each subquery.
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
assert(pQueryInfo->numOfTables == 1 && pSql->cmd.numOfClause == 1);
// retrieve actual query results from vnode during the second stage join subquery
if (pParentSql->res.code != TSDB_CODE_SUCCESS) {
tscError("%p abort query due to other subquery failure. code:%d, global code:%d", pSql, code, pParentSql->res.code);
quitAllSubquery(pSql, pParentSql, pSupporter);
tscAsyncResultOnError(pParentSql);
return;
@ -1472,9 +1475,6 @@ void tscJoinQueryCallback(void* param, TAOS_RES* tres, int code) {
return;
}
STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
// In case of consequence query from other vnode, do not wait for other query response here.
if (!(pTableMetaInfo->vgroupIndex > 0 && tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0))) {
if (!subAndCheckDone(pSql, &pParentSql->subState, pSupporter->subqueryIndex)) {
@ -1736,6 +1736,25 @@ static void doCleanupSubqueries(SSqlObj *pSql, int32_t numOfSubs) {
}
}
void tscLockByThread(int64_t *lockedBy) {
int64_t tid = taosGetSelfPthreadId();
int i = 0;
while (atomic_val_compare_exchange_64(lockedBy, 0, tid) != 0) {
if (++i % 100 == 0) {
sched_yield();
}
}
}
void tscUnlockByThread(int64_t *lockedBy) {
int64_t tid = taosGetSelfPthreadId();
if (atomic_val_compare_exchange_64(lockedBy, tid, 0) != tid) {
assert(false);
}
}
int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) {
SSqlRes *pRes = &pSql->res;
SSqlCmd *pCmd = &pSql->cmd;

20
src/client/src/tscUtil.c
View File

@ -474,6 +474,18 @@ void tscFreeRegisteredSqlObj(void *pSql) {
}
void tscFreeMetaSqlObj(int64_t *rid){
if (RID_VALID(*rid)) {
SSqlObj* pSql = (SSqlObj*)taosAcquireRef(tscObjRef, *rid);
if (pSql) {
taosRemoveRef(tscObjRef, *rid);
taosReleaseRef(tscObjRef, *rid);
}
*rid = 0;
}
}
void tscFreeSqlObj(SSqlObj* pSql) {
if (pSql == NULL || pSql->signature != pSql) {
return;
@ -483,6 +495,9 @@ void tscFreeSqlObj(SSqlObj* pSql) {
pSql->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
tscFreeMetaSqlObj(&pSql->metaRid);
tscFreeMetaSqlObj(&pSql->svgroupRid);
tscFreeSubobj(pSql);
SSqlCmd* pCmd = &pSql->cmd;
@ -511,6 +526,7 @@ void tscFreeSqlObj(SSqlObj* pSql) {
pCmd->allocSize = 0;
tsem_destroy(&pSql->rspSem);
memset(pSql, 0, sizeof(*pSql));
free(pSql);
}
@ -2199,7 +2215,9 @@ void tscDoQuery(SSqlObj* pSql) {
tscProcessSql(pSql);
} else { // secondary stage join query.
if (tscIsTwoStageSTableQuery(pQueryInfo, 0)) { // super table query
tscLockByThread(&pSql->squeryLock);
tscHandleMasterSTableQuery(pSql);
tscUnlockByThread(&pSql->squeryLock);
} else {
tscProcessSql(pSql);
}
@ -2208,7 +2226,9 @@ void tscDoQuery(SSqlObj* pSql) {
return;
} else if (tscIsTwoStageSTableQuery(pQueryInfo, 0)) { // super table query
tscLockByThread(&pSql->squeryLock);
tscHandleMasterSTableQuery(pSql);
tscUnlockByThread(&pSql->squeryLock);
return;
}

2
src/client/tests/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
FIND_PATH(HEADER_GTEST_INCLUDE_DIR gtest.h /usr/include/gtest /usr/local/include/gtest)

4
src/client/tests/timeParseTest.cpp
View File

@ -162,6 +162,10 @@ TEST(testCase, parse_time) {
taosParseTime(t13, &time, strlen(t13), TSDB_TIME_PRECISION_MILLI, 0);
EXPECT_EQ(time, -28800 * MILLISECOND_PER_SECOND);
char* t = "2021-01-08T02:11:40.000+00:00";
taosParseTime(t, &time, strlen(t), TSDB_TIME_PRECISION_MILLI, 0);
printf("%ld\n", time);
}

2
src/common/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(inc)

1
src/common/inc/tglobal.h
View File

@ -63,7 +63,6 @@ extern int32_t tsRetrieveBlockingModel;// retrieve threads will be blocked
extern int8_t tsKeepOriginalColumnName;
// client
extern int32_t tsTableMetaKeepTimer;
extern int32_t tsMaxSQLStringLen;
extern int8_t tsTscEnableRecordSql;
extern int32_t tsMaxNumOfOrderedResults;

11
src/common/src/tglobal.c
View File

@ -72,7 +72,6 @@ char tsTempDir[TSDB_FILENAME_LEN] = "/tmp/";
int32_t tsCompressMsgSize = -1;
// client
int32_t tsTableMetaKeepTimer = 7200; // second
int32_t tsMaxSQLStringLen = TSDB_MAX_SQL_LEN;
int8_t tsTscEnableRecordSql = 0;
@ -596,16 +595,6 @@ static void doInitGlobalConfig(void) {
cfg.unitType = TAOS_CFG_UTYPE_SECOND;
taosInitConfigOption(cfg);
cfg.option = "tableMetaKeepTimer";
cfg.ptr = &tsTableMetaKeepTimer;
cfg.valType = TAOS_CFG_VTYPE_INT32;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_CLIENT;
cfg.minValue = 1;
cfg.maxValue = 8640000;
cfg.ptrLength = 0;
cfg.unitType = TAOS_CFG_UTYPE_SECOND;
taosInitConfigOption(cfg);
cfg.option = "minSlidingTime";
cfg.ptr = &tsMinSlidingTime;
cfg.valType = TAOS_CFG_VTYPE_INT32;

2
src/connector/jdbc/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)

2
src/connector/odbc/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
IF (TD_LINUX_64)

2
src/connector/odbc/src/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
IF (TD_LINUX_64)

2
src/cq/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(inc)

2
src/cq/src/cqMain.c
View File

@ -43,7 +43,7 @@ typedef struct {
int32_t master;
int32_t num; // number of continuous streams
char user[TSDB_USER_LEN];
char pass[TSDB_PASSWORD_LEN];
char pass[TSDB_KEY_LEN];
char db[TSDB_DB_NAME_LEN];
FCqWrite cqWrite;
struct SCqObj *pHead;

2
src/cq/test/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
LIST(APPEND CQTEST_SRC ./cqtest.c)

2
src/dnode/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/query/inc)

8
src/dnode/src/dnodeVWrite.c
View File

@ -202,12 +202,12 @@ static void *dnodeProcessVWriteQueue(void *wparam) {
for (int32_t i = 0; i < numOfMsgs; ++i) {
taosGetQitem(pWorker->qall, &qtype, (void **)&pWrite);
dTrace("msg:%p, app:%p type:%s will be processed in vwrite queue, qtype:%s hver:%" PRIu64, pWrite,
pWrite->rpcMsg.ahandle, taosMsg[pWrite->pHead->msgType], qtypeStr[qtype], pWrite->pHead->version);
pWrite->rpcMsg.ahandle, taosMsg[pWrite->pHead.msgType], qtypeStr[qtype], pWrite->pHead.version);
pWrite->code = vnodeProcessWrite(pVnode, pWrite->pHead, qtype, pWrite);
pWrite->code = vnodeProcessWrite(pVnode, &pWrite->pHead, qtype, pWrite);
if (pWrite->code <= 0) pWrite->processedCount = 1;
if (pWrite->code > 0) pWrite->code = 0;
if (pWrite->code == 0 && pWrite->pHead->msgType != TSDB_MSG_TYPE_SUBMIT) forceFsync = true;
if (pWrite->code == 0 && pWrite->pHead.msgType != TSDB_MSG_TYPE_SUBMIT) forceFsync = true;
dTrace("msg:%p is processed in vwrite queue, code:0x%x", pWrite, pWrite->code);
}
@ -222,7 +222,7 @@ static void *dnodeProcessVWriteQueue(void *wparam) {
dnodeSendRpcVWriteRsp(pVnode, pWrite, pWrite->code);
} else {
if (qtype == TAOS_QTYPE_FWD) {
vnodeConfirmForward(pVnode, pWrite->pHead->version, 0);
vnodeConfirmForward(pVnode, pWrite->pHead.version, 0);
}
if (pWrite->rspRet.rsp) {
rpcFreeCont(pWrite->rspRet.rsp);

18
src/inc/taosdef.h
View File

@ -256,10 +256,8 @@ int32_t tStrToInteger(const char* z, int16_t type, int32_t n, int64_t* value, bo
#define TSDB_USER_LEN TSDB_UNI_LEN
// ACCOUNT is a 32 bit positive integer
// this is the length of its string representation
// including the terminator zero
#define TSDB_ACCT_LEN 11
#define TSDB_PASSWORD_LEN TSDB_UNI_LEN
// this is the length of its string representation, including the terminator zero
#define TSDB_ACCT_ID_LEN 11
#define TSDB_MAX_COLUMNS 1024
#define TSDB_MIN_COLUMNS 2 //PRIMARY COLUMN(timestamp) + other columns
@ -267,7 +265,7 @@ int32_t tStrToInteger(const char* z, int16_t type, int32_t n, int64_t* value, bo
#define TSDB_NODE_NAME_LEN 64
#define TSDB_TABLE_NAME_LEN 193 // it is a null-terminated string
#define TSDB_DB_NAME_LEN 33
#define TSDB_TABLE_FNAME_LEN (TSDB_ACCT_LEN + TSDB_DB_NAME_LEN + TSDB_TABLE_NAME_LEN)
#define TSDB_TABLE_FNAME_LEN (TSDB_ACCT_ID_LEN + TSDB_DB_NAME_LEN + TSDB_TABLE_NAME_LEN)
#define TSDB_COL_NAME_LEN 65
#define TSDB_MAX_SAVED_SQL_LEN TSDB_MAX_COLUMNS * 64
#define TSDB_MAX_SQL_LEN TSDB_PAYLOAD_SIZE
@ -293,11 +291,6 @@ int32_t tStrToInteger(const char* z, int16_t type, int32_t n, int64_t* value, bo
#define TSDB_EP_LEN (TSDB_FQDN_LEN+6)
#define TSDB_IPv4ADDR_LEN 16
#define TSDB_FILENAME_LEN 128
#define TSDB_METER_VNODE_BITS 20
#define TSDB_METER_SID_MASK 0xFFFFF
#define TSDB_SHELL_VNODE_BITS 24
#define TSDB_SHELL_SID_MASK 0xFF
#define TSDB_HTTP_TOKEN_LEN 20
#define TSDB_SHOW_SQL_LEN 512
#define TSDB_SLOW_QUERY_SQL_LEN 512
@ -311,9 +304,6 @@ int32_t tStrToInteger(const char* z, int16_t type, int32_t n, int64_t* value, bo
#define TSDB_MQTT_TOPIC_LEN 64
#define TSDB_MQTT_CLIENT_ID_LEN 32
#define TSDB_METER_STATE_OFFLINE 0
#define TSDB_METER_STATE_ONLLINE 1
#define TSDB_DEFAULT_PKT_SIZE 65480 //same as RPC_MAX_UDP_SIZE
#define TSDB_PAYLOAD_SIZE TSDB_DEFAULT_PKT_SIZE
@ -333,7 +323,7 @@ int32_t tStrToInteger(const char* z, int16_t type, int32_t n, int64_t* value, bo
#define TSDB_TBNAME_COLUMN_INDEX (-1)
#define TSDB_UD_COLUMN_INDEX (-100)
#define TSDB_MULTI_METERMETA_MAX_NUM 100000 // maximum batch size allowed to load metermeta
#define TSDB_MULTI_TABLEMETA_MAX_NUM 100000 // maximum batch size allowed to load table meta
#define TSDB_MIN_CACHE_BLOCK_SIZE 1
#define TSDB_MAX_CACHE_BLOCK_SIZE 128 // 128MB for each vnode

12
src/inc/taosmsg.h
View File

@ -268,7 +268,7 @@ typedef struct {
typedef struct {
int32_t len; // one create table message
char tableId[TSDB_TABLE_FNAME_LEN];
char db[TSDB_ACCT_LEN + TSDB_DB_NAME_LEN];
char db[TSDB_ACCT_ID_LEN + TSDB_DB_NAME_LEN];
int8_t igExists;
int8_t getMeta;
int16_t numOfTags;
@ -290,7 +290,7 @@ typedef struct {
typedef struct {
char tableId[TSDB_TABLE_FNAME_LEN];
char db[TSDB_ACCT_LEN + TSDB_DB_NAME_LEN];
char db[TSDB_ACCT_ID_LEN + TSDB_DB_NAME_LEN];
int16_t type; /* operation type */
int16_t numOfCols; /* number of schema */
int32_t tagValLen;
@ -322,7 +322,7 @@ typedef struct {
} SConnectMsg;
typedef struct {
char acctId[TSDB_ACCT_LEN];
char acctId[TSDB_ACCT_ID_LEN];
char serverVersion[TSDB_VERSION_LEN];
char clusterId[TSDB_CLUSTER_ID_LEN];
int8_t writeAuth;
@ -534,7 +534,7 @@ typedef struct {
} SVnodeLoad;
typedef struct {
char db[TSDB_ACCT_LEN + TSDB_DB_NAME_LEN];
char db[TSDB_ACCT_ID_LEN + TSDB_DB_NAME_LEN];
int32_t cacheBlockSize; //MB
int32_t totalBlocks;
int32_t maxTables;
@ -682,7 +682,7 @@ typedef struct {
} SVnodeDesc;
typedef struct {
char db[TSDB_ACCT_LEN + TSDB_DB_NAME_LEN];
char db[TSDB_ACCT_ID_LEN + TSDB_DB_NAME_LEN];
SVnodeCfg cfg;
SVnodeDesc nodes[TSDB_MAX_REPLICA];
} SCreateVnodeMsg, SAlterVnodeMsg;
@ -761,7 +761,7 @@ typedef struct {
*/
typedef struct {
int8_t type;
char db[TSDB_ACCT_LEN + TSDB_DB_NAME_LEN];
char db[TSDB_ACCT_ID_LEN + TSDB_DB_NAME_LEN];
uint16_t payloadLen;
char payload[];
} SShowMsg;

2
src/inc/tcq.h
View File

@ -26,7 +26,7 @@ typedef int32_t (*FCqWrite)(int32_t vgId, void *pHead, int32_t qtype, void *pMsg
typedef struct {
int32_t vgId;
char user[TSDB_USER_LEN];
char pass[TSDB_PASSWORD_LEN];
char pass[TSDB_KEY_LEN];
char db[TSDB_DB_NAME_LEN];
FCqWrite cqWrite;
} SCqCfg;

2
src/inc/vnode.h
View File

@ -49,7 +49,7 @@ typedef struct {
SRpcMsg rpcMsg;
SRspRet rspRet;
char reserveForSync[24];
SWalHead pHead[];
SWalHead pHead;
} SVWriteMsg;
// vnodeStatus

2
src/kit/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
ADD_SUBDIRECTORY(shell)

2
src/kit/shell/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/client/inc)

2
src/kit/taosdemo/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/client/inc)

2
src/kit/taosdemox/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/client/inc)

2
src/kit/taosdump/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/client/inc)

2
src/mnode/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
IF (TD_LINUX)

4
src/mnode/inc/mnodeDef.h
View File

@ -138,7 +138,7 @@ typedef struct SVgObj {
int64_t createdTime;
int32_t lbDnodeId;
int32_t lbTime;
char dbName[TSDB_ACCT_LEN + TSDB_DB_NAME_LEN];
char dbName[TSDB_ACCT_ID_LEN + TSDB_DB_NAME_LEN];
int8_t inUse;
int8_t accessState;
int8_t status;
@ -179,7 +179,7 @@ typedef struct {
} SDbCfg;
typedef struct SDbObj {
char name[TSDB_ACCT_LEN + TSDB_DB_NAME_LEN];
char name[TSDB_ACCT_ID_LEN + TSDB_DB_NAME_LEN];
int8_t reserved0[4];
char acct[TSDB_USER_LEN];
int64_t createdTime;

2
src/mnode/src/mnodeVgroup.c
View File

@ -565,7 +565,7 @@ int32_t mnodeCreateVgroup(SMnodeMsg *pMsg) {
SDbObj *pDb = pMsg->pDb;
SVgObj *pVgroup = (SVgObj *)calloc(1, sizeof(SVgObj));
tstrncpy(pVgroup->dbName, pDb->name, TSDB_ACCT_LEN + TSDB_DB_NAME_LEN);
tstrncpy(pVgroup->dbName, pDb->name, TSDB_ACCT_ID_LEN + TSDB_DB_NAME_LEN);
pVgroup->numOfVnodes = pDb->cfg.replications;
pVgroup->createdTime = taosGetTimestampMs();
pVgroup->accessState = TSDB_VN_ALL_ACCCESS;

4
src/os/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
IF (TD_LINUX)
@ -10,3 +10,5 @@ ELSEIF (TD_WINDOWS)
ENDIF ()
ADD_SUBDIRECTORY(src/detail)
ADD_SUBDIRECTORY(tests)

2
src/os/src/alpine/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
AUX_SOURCE_DIRECTORY(. SRC)

2
src/os/src/darwin/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
AUX_SOURCE_DIRECTORY(. SRC)

2
src/os/src/detail/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(.)

23
src/os/src/detail/osTime.c
View File

@ -486,7 +486,7 @@ int64_t taosTimeTruncate(int64_t t, const SInterval* pInterval, int32_t precisio
start = (delta / pInterval->sliding + factor) * pInterval->sliding;
if (pInterval->intervalUnit == 'd' || pInterval->intervalUnit == 'w') {
/*
/*
* here we revised the start time of day according to the local time zone,
* but in case of DST, the start time of one day need to be dynamically decided.
*/
@ -501,9 +501,24 @@ int64_t taosTimeTruncate(int64_t t, const SInterval* pInterval, int32_t precisio
start += (int64_t)(timezone * TSDB_TICK_PER_SECOND(precision));
}
int64_t end = start + pInterval->interval - 1;
if (end < t) {
start += pInterval->sliding;
int64_t end = 0;
// not enough time range
if (INT64_MAX - start > pInterval->interval - 1) {
end = start + pInterval->interval - 1;
while(end < t && ((start + pInterval->sliding) <= INT64_MAX)) { // move forward to the correct time window
start += pInterval->sliding;
if (INT64_MAX - start > pInterval->interval - 1) {
end = start + pInterval->interval - 1;
} else {
end = INT64_MAX;
break;
}
}
} else {
end = INT64_MAX;
}
}

2
src/os/src/linux/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
AUX_SOURCE_DIRECTORY(. SRC)

2
src/os/src/windows/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
AUX_SOURCE_DIRECTORY(. SRC)

15
src/os/tests/CMakeLists.txt Normal file
View File

@ -0,0 +1,15 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
PROJECT(TDengine)
FIND_PATH(HEADER_GTEST_INCLUDE_DIR gtest.h /usr/include/gtest /usr/local/include/gtest)
FIND_LIBRARY(LIB_GTEST_STATIC_DIR libgtest.a /usr/lib/ /usr/local/lib)
IF (HEADER_GTEST_INCLUDE_DIR AND LIB_GTEST_STATIC_DIR)
MESSAGE(STATUS "gTest library found, build unit test")
INCLUDE_DIRECTORIES(${HEADER_GTEST_INCLUDE_DIR})
AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR} SOURCE_LIST)
ADD_EXECUTABLE(osTest ${SOURCE_LIST})
TARGET_LINK_LIBRARIES(osTest taos osdetail tutil common gtest pthread)
ENDIF()

34
src/os/tests/test.cpp Normal file
View File

@ -0,0 +1,34 @@
#include "os.h"
#include <gtest/gtest.h>
#include <cassert>
#include <iostream>
#include "taos.h"
#include "tstoken.h"
#include "tutil.h"
int main(int argc, char** argv) {
testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
// test function in os module
TEST(testCase, parse_time) {
taos_options(TSDB_OPTION_TIMEZONE, "GMT-8");
deltaToUtcInitOnce();
// window: 1500000001000, 1500002000000
// pQuery->interval: interval: 86400000, sliding:3600000
int64_t key = 1500000001000;
SInterval interval = {0};
interval.interval = 86400000;
interval.intervalUnit = 'd';
interval.sliding = 3600000;
interval.slidingUnit = 'h';
int64_t s = taosTimeTruncate(key, &interval, TSDB_TIME_PRECISION_MILLI);
ASSERT_TRUE(s + interval.interval >= key);
}

2
src/plugins/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
ADD_SUBDIRECTORY(monitor)

2
src/plugins/http/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/zlib-1.2.11/inc)

4
src/plugins/http/inc/httpInt.h
View File

@ -39,7 +39,7 @@
#define HTTP_GC_TARGET_SIZE 512
#define HTTP_WRITE_RETRY_TIMES 500
#define HTTP_WRITE_WAIT_TIME_MS 5
#define HTTP_SESSION_ID_LEN (TSDB_USER_LEN + TSDB_PASSWORD_LEN)
#define HTTP_SESSION_ID_LEN (TSDB_USER_LEN + TSDB_KEY_LEN)
typedef enum HttpReqType {
HTTP_REQTYPE_OTHERS = 0,
@ -147,7 +147,7 @@ typedef struct HttpContext {
uint8_t parsed;
char ipstr[22];
char user[TSDB_USER_LEN]; // parsed from auth token or login message
char pass[TSDB_PASSWORD_LEN];
char pass[TSDB_KEY_LEN];
TAOS * taos;
void * ppContext;
HttpSession *session;

8
src/plugins/http/src/httpAuth.c
View File

@ -51,7 +51,7 @@ int32_t httpParseBasicAuthToken(HttpContext *pContext, char *token, int32_t len)
char *password = user + 1;
int32_t pass_len = (int32_t)((base64 + outlen) - password);
if (pass_len < 1 || pass_len >= TSDB_PASSWORD_LEN) {
if (pass_len < 1 || pass_len >= TSDB_KEY_LEN) {
httpError("context:%p, fd:%d, basic token:%s parse password error", pContext, pContext->fd, token);
free(base64);
return -1;
@ -73,7 +73,7 @@ int32_t httpParseTaosdAuthToken(HttpContext *pContext, char *token, int32_t len)
if (base64) free(base64);
return 01;
}
if (outlen != (TSDB_USER_LEN + TSDB_PASSWORD_LEN)) {
if (outlen != (TSDB_USER_LEN + TSDB_KEY_LEN)) {
httpError("context:%p, fd:%d, taosd token:%s length error", pContext, pContext->fd, token);
free(base64);
return -1;
@ -103,8 +103,8 @@ int32_t httpGenTaosdAuthToken(HttpContext *pContext, char *token, int32_t maxLen
size = sizeof(pContext->pass);
tstrncpy(buffer + sizeof(pContext->user), pContext->pass, size);
char *encrypt = taosDesEncode(KEY_DES_4, buffer, TSDB_USER_LEN + TSDB_PASSWORD_LEN);
char *base64 = base64_encode((const unsigned char *)encrypt, TSDB_USER_LEN + TSDB_PASSWORD_LEN);
char *encrypt = taosDesEncode(KEY_DES_4, buffer, TSDB_USER_LEN + TSDB_KEY_LEN);
char *base64 = base64_encode((const unsigned char *)encrypt, TSDB_USER_LEN + TSDB_KEY_LEN);
size_t len = strlen(base64);
tstrncpy(token, base64, len + 1);

4
src/plugins/http/src/httpGcHandle.c
View File

@ -59,11 +59,11 @@ bool gcGetUserFromUrl(HttpContext* pContext) {
bool gcGetPassFromUrl(HttpContext* pContext) {
HttpParser* pParser = pContext->parser;
if (pParser->path[GC_PASS_URL_POS].pos >= TSDB_PASSWORD_LEN || pParser->path[GC_PASS_URL_POS].pos <= 0) {
if (pParser->path[GC_PASS_URL_POS].pos >= TSDB_KEY_LEN || pParser->path[GC_PASS_URL_POS].pos <= 0) {
return false;
}
tstrncpy(pContext->pass, pParser->path[GC_PASS_URL_POS].str, TSDB_PASSWORD_LEN);
tstrncpy(pContext->pass, pParser->path[GC_PASS_URL_POS].str, TSDB_KEY_LEN);
return true;
}

4
src/plugins/http/src/httpRestHandle.c
View File

@ -72,11 +72,11 @@ bool restGetUserFromUrl(HttpContext* pContext) {
bool restGetPassFromUrl(HttpContext* pContext) {
HttpParser* pParser = pContext->parser;
if (pParser->path[REST_PASS_URL_POS].pos >= TSDB_PASSWORD_LEN || pParser->path[REST_PASS_URL_POS].pos <= 0) {
if (pParser->path[REST_PASS_URL_POS].pos >= TSDB_KEY_LEN || pParser->path[REST_PASS_URL_POS].pos <= 0) {
return false;
}
tstrncpy(pContext->pass, pParser->path[REST_PASS_URL_POS].str, TSDB_PASSWORD_LEN);
tstrncpy(pContext->pass, pParser->path[REST_PASS_URL_POS].str, TSDB_KEY_LEN);
return true;
}

2
src/plugins/http/src/httpTgHandle.c
View File

@ -324,7 +324,7 @@ bool tgGetUserFromUrl(HttpContext *pContext) {
bool tgGetPassFromUrl(HttpContext *pContext) {
HttpParser *pParser = pContext->parser;
if (pParser->path[TG_PASS_URL_POS].pos >= TSDB_PASSWORD_LEN || pParser->path[TG_PASS_URL_POS].pos <= 0) {
if (pParser->path[TG_PASS_URL_POS].pos >= TSDB_KEY_LEN || pParser->path[TG_PASS_URL_POS].pos <= 0) {
return false;
}

2
src/plugins/monitor/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(inc)

2
src/plugins/mqtt/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(inc)

2
src/query/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/tsdb/inc)

38
src/query/inc/qPercentile.h
View File

@ -16,32 +16,36 @@
#ifndef TDENGINE_QPERCENTILE_H
#define TDENGINE_QPERCENTILE_H
#ifdef __cplusplus
extern "C" {
#endif
#include "qExtbuffer.h"
#include "qResultbuf.h"
#include "qTsbuf.h"
typedef struct MinMaxEntry {
union {
double dMinVal;
int32_t iMinVal;
int64_t i64MinVal;
double dMinVal;
int64_t i64MinVal;
uint64_t u64MinVal;
};
union {
double dMaxVal;
int32_t iMaxVal;
int64_t i64MaxVal;
int64_t u64MaxVal;
};
} MinMaxEntry;
typedef struct {
int32_t size;
int32_t pageId;
int32_t size;
int32_t pageId;
tFilePage *data;
} SSlotInfo;
typedef struct tMemBucketSlot {
SSlotInfo info;
MinMaxEntry range;
SSlotInfo info;
MinMaxEntry range;
} tMemBucketSlot;
struct tMemBucket;
@ -52,16 +56,16 @@ typedef struct tMemBucket {
int16_t type;
int16_t bytes;
int32_t total;
int32_t elemPerPage; // number of elements for each object
int32_t maxCapacity; // maximum allowed number of elements that can be sort directly to get the result
int32_t bufPageSize; // disk page size
MinMaxEntry range; // value range
int32_t times; // count that has been checked for deciding the correct data value buckets.
int32_t elemPerPage; // number of elements for each object
int32_t maxCapacity; // maximum allowed number of elements that can be sort directly to get the result
int32_t bufPageSize; // disk page size
MinMaxEntry range; // value range
int32_t times; // count that has been checked for deciding the correct data value buckets.
__compar_fn_t comparFn;
tMemBucketSlot *pSlots;
tMemBucketSlot * pSlots;
SDiskbasedResultBuf *pBuffer;
__perc_hash_func_t hashFunc;
__perc_hash_func_t hashFunc;
} tMemBucket;
tMemBucket *tMemBucketCreate(int16_t nElemSize, int16_t dataType, double minval, double maxval);
@ -73,3 +77,7 @@ int32_t tMemBucketPut(tMemBucket *pBucket, const void *data, size_t size);
double getPercentile(tMemBucket *pMemBucket, double percent);
#endif // TDENGINE_QPERCENTILE_H
#ifdef __cplusplus
}
#endif

2
src/query/src/qAggMain.c
View File

@ -2545,7 +2545,7 @@ static void percentile_next_step(SQLFunctionCtx *pCtx) {
if (pInfo->numOfElems == 0) {
pResInfo->complete = true;
} else {
pInfo->pMemBucket = tMemBucketCreate(pCtx->inputBytes, pCtx->inputType, GET_DOUBLE_VAL(&pInfo->minval), GET_DOUBLE_VAL(&pInfo->maxval));
pInfo->pMemBucket = tMemBucketCreate(pCtx->inputBytes, pCtx->inputType, pInfo->minval, pInfo->maxval);
}
pInfo->stage += 1;

567
src/query/src/qPercentile.c
View File

@ -12,14 +12,14 @@
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "os.h"
#include "qPercentile.h"
#include "qResultbuf.h"
#include "os.h"
#include "queryLog.h"
#include "taosdef.h"
#include "tulog.h"
#include "tcompare.h"
#include "ttype.h"
#define DEFAULT_NUM_OF_SLOT 1024
@ -48,25 +48,15 @@ static tFilePage *loadDataFromFilePage(tMemBucket *pMemBucket, int32_t slotIdx)
}
static void resetBoundingBox(MinMaxEntry* range, int32_t type) {
switch (type) {
case TSDB_DATA_TYPE_BIGINT: {
range->i64MaxVal = INT64_MIN;
range->i64MinVal = INT64_MAX;
break;
};
case TSDB_DATA_TYPE_INT:
case TSDB_DATA_TYPE_SMALLINT:
case TSDB_DATA_TYPE_TINYINT: {
range->iMaxVal = INT32_MIN;
range->iMinVal = INT32_MAX;
break;
};
case TSDB_DATA_TYPE_DOUBLE:
case TSDB_DATA_TYPE_FLOAT: {
range->dMaxVal = -DBL_MAX;
range->dMinVal = DBL_MAX;
break;
}
if (IS_SIGNED_NUMERIC_TYPE(type)) {
range->i64MaxVal = INT64_MIN;
range->i64MinVal = INT64_MAX;
} else if (IS_UNSIGNED_NUMERIC_TYPE(type)) {
range->u64MaxVal = 0;
range->u64MinVal = UINT64_MAX;
} else {
range->dMaxVal = -DBL_MAX;
range->dMinVal = DBL_MAX;
}
}
@ -75,23 +65,15 @@ static int32_t setBoundingBox(MinMaxEntry* range, int16_t type, double minval, d
return -1;
}
switch(type) {
case TSDB_DATA_TYPE_TINYINT:
case TSDB_DATA_TYPE_SMALLINT:
case TSDB_DATA_TYPE_INT:
range->iMinVal = (int32_t) minval;
range->iMaxVal = (int32_t) maxval;
break;
case TSDB_DATA_TYPE_BIGINT:
range->i64MinVal = (int64_t) minval;
range->i64MaxVal = (int64_t) maxval;
break;
case TSDB_DATA_TYPE_FLOAT:
case TSDB_DATA_TYPE_DOUBLE:
range->dMinVal = minval;
range->dMaxVal = maxval;
break;
if (IS_SIGNED_NUMERIC_TYPE(type)) {
range->i64MinVal = (int64_t) minval;
range->i64MaxVal = (int64_t) maxval;
} else if (IS_UNSIGNED_NUMERIC_TYPE(type)){
range->u64MinVal = (uint64_t) minval;
range->u64MaxVal = (uint64_t) maxval;
} else {
range->dMinVal = minval;
range->dMaxVal = maxval;
}
return 0;
@ -120,117 +102,56 @@ double findOnlyResult(tMemBucket *pMemBucket) {
tFilePage* pPage = getResBufPage(pMemBucket->pBuffer, pgInfo->pageId);
assert(pPage->num == 1);
switch (pMemBucket->type) {
case TSDB_DATA_TYPE_INT:
return *(int32_t *)pPage->data;
case TSDB_DATA_TYPE_SMALLINT:
return *(int16_t *)pPage->data;
case TSDB_DATA_TYPE_TINYINT:
return *(int8_t *)pPage->data;
case TSDB_DATA_TYPE_BIGINT:
return (double)(*(int64_t *)pPage->data);
case TSDB_DATA_TYPE_DOUBLE: {
double dv = GET_DOUBLE_VAL(pPage->data);
return dv;
}
case TSDB_DATA_TYPE_FLOAT: {
float fv = GET_FLOAT_VAL(pPage->data);
return fv;
}
default:
return 0;
}
double v = 0;
GET_TYPED_DATA(v, double, pMemBucket->type, pPage->data);
return v;
}
return 0;
}
int32_t tBucketBigIntHash(tMemBucket *pBucket, const void *value) {
int64_t v = *(int64_t *)value;
int32_t tBucketIntHash(tMemBucket *pBucket, const void *value) {
int64_t v = 0;
GET_TYPED_DATA(v, int64_t, pBucket->type, value);
int32_t index = -1;
int32_t halfSlot = pBucket->numOfSlots >> 1;
// int32_t bits = 32;//bitsOfNumber(pBucket->numOfSlots) - 1;
if (pBucket->range.i64MaxVal == INT64_MIN) {
if (v >= 0) {
index = (v >> (64 - 9)) + halfSlot;
} else { // v<0
index = ((-v) >> (64 - 9));
index = -index + (halfSlot - 1);
}
return index;
// divide the value range into 1024 buckets
uint64_t span = pBucket->range.i64MaxVal - pBucket->range.i64MinVal;
if (span < pBucket->numOfSlots) {
int64_t delta = v - pBucket->range.i64MinVal;
index = (delta % pBucket->numOfSlots);
} else {
// out of range
if (v < pBucket->range.i64MinVal || v > pBucket->range.i64MaxVal) {
return -1;
double slotSpan = (double)span / pBucket->numOfSlots;
index = (int32_t)((v - pBucket->range.i64MinVal) / slotSpan);
if (v == pBucket->range.i64MaxVal) {
index -= 1;
}
// todo hash for bigint and float and double
int64_t span = pBucket->range.i64MaxVal - pBucket->range.i64MinVal;
if (span < pBucket->numOfSlots) {
int32_t delta = (int32_t)(v - pBucket->range.i64MinVal);
index = delta % pBucket->numOfSlots;
} else {
double slotSpan = (double)span / pBucket->numOfSlots;
index = (int32_t)((v - pBucket->range.i64MinVal) / slotSpan);
if (v == pBucket->range.i64MaxVal) {
index -= 1;
}
}
return index;
}
assert(v >= pBucket->range.i64MinVal && v <= pBucket->range.i64MaxVal && index >= 0 && index < pBucket->numOfSlots);
return index;
}
// todo refactor to more generic
int32_t tBucketIntHash(tMemBucket *pBucket, const void *value) {
int32_t v = 0;
switch(pBucket->type) {
case TSDB_DATA_TYPE_SMALLINT: v = *(int16_t*) value; break;
case TSDB_DATA_TYPE_TINYINT: v = *(int8_t*) value; break;
default: v = *(int32_t*) value;break;
}
int32_t tBucketUintHash(tMemBucket *pBucket, const void *value) {
int64_t v = 0;
GET_TYPED_DATA(v, uint64_t, pBucket->type, value);
int32_t index = -1;
if (pBucket->range.iMaxVal == INT32_MIN) {
/*
* taking negative integer into consideration,
* there is only half of pBucket->segs available for non-negative integer
*/
int32_t halfSlot = pBucket->numOfSlots >> 1;
int32_t bits = 32;//bitsOfNumber(pBucket->numOfSlots) - 1;
if (v >= 0) {
index = (v >> (bits - 9)) + halfSlot;
} else { // v < 0
index = ((-v) >> (32 - 9));
index = -index + (halfSlot - 1);
}
return index;
// divide the value range into 1024 buckets
uint64_t span = pBucket->range.u64MaxVal - pBucket->range.u64MinVal;
if (span < pBucket->numOfSlots) {
int64_t delta = v - pBucket->range.u64MinVal;
index = (int32_t) (delta % pBucket->numOfSlots);
} else {
// out of range
if (v < pBucket->range.iMinVal || v > pBucket->range.iMaxVal) {
return -1;
double slotSpan = (double)span / pBucket->numOfSlots;
index = (int32_t)((v - pBucket->range.u64MinVal) / slotSpan);
if (v == pBucket->range.u64MaxVal) {
index -= 1;
}
// divide a range of [iMinVal, iMaxVal] into 1024 buckets
int32_t span = pBucket->range.iMaxVal - pBucket->range.iMinVal;
if (span < pBucket->numOfSlots) {
int32_t delta = v - pBucket->range.iMinVal;
index = (delta % pBucket->numOfSlots);
} else {
double slotSpan = (double)span / pBucket->numOfSlots;
index = (int32_t)((v - pBucket->range.iMinVal) / slotSpan);
if (v == pBucket->range.iMaxVal) {
index -= 1;
}
}
return index;
}
assert(v >= pBucket->range.u64MinVal && v <= pBucket->range.i64MaxVal && index >= 0 && index < pBucket->numOfSlots);
return index;
}
int32_t tBucketDoubleHash(tMemBucket *pBucket, const void *value) {
@ -243,62 +164,30 @@ int32_t tBucketDoubleHash(tMemBucket *pBucket, const void *value) {
int32_t index = -1;
if (pBucket->range.dMinVal == DBL_MAX) {
/*
* taking negative integer into consideration,
* there is only half of pBucket->segs available for non-negative integer
*/
double x = DBL_MAX / (pBucket->numOfSlots >> 1);
double posx = (v + DBL_MAX) / x;
return ((int32_t)posx) % pBucket->numOfSlots;
// divide a range of [dMinVal, dMaxVal] into 1024 buckets
double span = pBucket->range.dMaxVal - pBucket->range.dMinVal;
if (span < pBucket->numOfSlots) {
int32_t delta = (int32_t)(v - pBucket->range.dMinVal);
index = (delta % pBucket->numOfSlots);
} else {
// out of range
if (v < pBucket->range.dMinVal || v > pBucket->range.dMaxVal) {
return -1;
double slotSpan = span / pBucket->numOfSlots;
index = (int32_t)((v - pBucket->range.dMinVal) / slotSpan);
if (v == pBucket->range.dMaxVal) {
index -= 1;
}
// divide a range of [dMinVal, dMaxVal] into 1024 buckets
double span = pBucket->range.dMaxVal - pBucket->range.dMinVal;
if (span < pBucket->numOfSlots) {
int32_t delta = (int32_t)(v - pBucket->range.dMinVal);
index = (delta % pBucket->numOfSlots);
} else {
double slotSpan = span / pBucket->numOfSlots;
index = (int32_t)((v - pBucket->range.dMinVal) / slotSpan);
if (v == pBucket->range.dMaxVal) {
index -= 1;
}
}
if (index < 0 || index > pBucket->numOfSlots) {
uError("error in hash process. slot id: %d", index);
}
return index;
}
assert(v >= pBucket->range.dMinVal && v <= pBucket->range.dMaxVal && index >= 0 && index < pBucket->numOfSlots);
return index;
}
static __perc_hash_func_t getHashFunc(int32_t type) {
switch (type) {
case TSDB_DATA_TYPE_INT:
case TSDB_DATA_TYPE_SMALLINT:
case TSDB_DATA_TYPE_TINYINT: {
return tBucketIntHash;
};
case TSDB_DATA_TYPE_DOUBLE:
case TSDB_DATA_TYPE_FLOAT: {
return tBucketDoubleHash;
};
case TSDB_DATA_TYPE_BIGINT: {
return tBucketBigIntHash;
};
default: {
return NULL;
}
if (IS_SIGNED_NUMERIC_TYPE(type)) {
return tBucketIntHash;
} else if (IS_UNSIGNED_NUMERIC_TYPE(type)) {
return tBucketUintHash;
} else {
return tBucketDoubleHash;
}
}
@ -328,7 +217,7 @@ tMemBucket *tMemBucketCreate(int16_t nElemSize, int16_t dataType, double minval,
pBucket->maxCapacity = 200000;
if (setBoundingBox(&pBucket->range, pBucket->type, minval, maxval) != 0) {
uError("MemBucket:%p, invalid value range: %f-%f", pBucket, minval, maxval);
qError("MemBucket:%p, invalid value range: %f-%f", pBucket, minval, maxval);
free(pBucket);
return NULL;
}
@ -338,7 +227,7 @@ tMemBucket *tMemBucketCreate(int16_t nElemSize, int16_t dataType, double minval,
pBucket->hashFunc = getHashFunc(pBucket->type);
if (pBucket->hashFunc == NULL) {
uError("MemBucket:%p, not support data type %d, failed", pBucket, pBucket->type);
qError("MemBucket:%p, not support data type %d, failed", pBucket, pBucket->type);
free(pBucket);
return NULL;
}
@ -357,7 +246,7 @@ tMemBucket *tMemBucketCreate(int16_t nElemSize, int16_t dataType, double minval,
return NULL;
}
uDebug("MemBucket:%p, elem size:%d", pBucket, pBucket->bytes);
qDebug("MemBucket:%p, elem size:%d", pBucket, pBucket->bytes);
return pBucket;
}
@ -372,77 +261,41 @@ void tMemBucketDestroy(tMemBucket *pBucket) {
}
void tMemBucketUpdateBoundingBox(MinMaxEntry *r, const char *data, int32_t dataType) {
switch (dataType) {
case TSDB_DATA_TYPE_INT: {
int32_t val = *(int32_t *)data;
if (r->iMinVal > val) {
r->iMinVal = val;
}
if (IS_SIGNED_NUMERIC_TYPE(dataType)) {
int64_t v = 0;
GET_TYPED_DATA(v, int64_t, dataType, data);
if (r->iMaxVal < val) {
r->iMaxVal = val;
}
break;
};
case TSDB_DATA_TYPE_BIGINT: {
int64_t val = *(int64_t *)data;
if (r->i64MinVal > val) {
r->i64MinVal = val;
}
if (r->i64MinVal > v) {
r->i64MinVal = v;
}
if (r->i64MaxVal < val) {
r->i64MaxVal = val;
}
break;
};
case TSDB_DATA_TYPE_SMALLINT: {
int32_t val = *(int16_t *)data;
if (r->iMinVal > val) {
r->iMinVal = val;
}
if (r->i64MaxVal < v) {
r->i64MaxVal = v;
}
} else if (IS_UNSIGNED_NUMERIC_TYPE(dataType)) {
uint64_t v = 0;
GET_TYPED_DATA(v, uint64_t, dataType, data);
if (r->iMaxVal < val) {
r->iMaxVal = val;
}
break;
};
case TSDB_DATA_TYPE_TINYINT: {
int32_t val = *(int8_t *)data;
if (r->iMinVal > val) {
r->iMinVal = val;
}
if (r->i64MinVal > v) {
r->i64MinVal = v;
}
if (r->iMaxVal < val) {
r->iMaxVal = val;
}
if (r->i64MaxVal < v) {
r->i64MaxVal = v;
}
} else if (IS_FLOAT_TYPE(dataType)) {
double v = 0;
GET_TYPED_DATA(v, double, dataType, data);
break;
};
case TSDB_DATA_TYPE_DOUBLE: {
// double val = *(double *)data;
double val = GET_DOUBLE_VAL(data);
if (r->dMinVal > val) {
r->dMinVal = val;
}
if (r->dMinVal > v) {
r->dMinVal = v;
}
if (r->dMaxVal < val) {
r->dMaxVal = val;
}
break;
};
case TSDB_DATA_TYPE_FLOAT: {
double val = GET_FLOAT_VAL(data);
if (r->dMinVal > val) {
r->dMinVal = val;
}
if (r->dMaxVal < val) {
r->dMaxVal = val;
}
break;
};
default: { assert(false); }
if (r->dMaxVal < v) {
r->dMaxVal = v;
}
} else {
assert(0);
}
}
@ -452,16 +305,13 @@ void tMemBucketUpdateBoundingBox(MinMaxEntry *r, const char *data, int32_t dataT
int32_t tMemBucketPut(tMemBucket *pBucket, const void *data, size_t size) {
assert(pBucket != NULL && data != NULL && size > 0);
pBucket->total += (int32_t)size;
int32_t count = 0;
int32_t bytes = pBucket->bytes;
for (int32_t i = 0; i < size; ++i) {
char *d = (char *) data + i * bytes;
count += 1;
int32_t index = (pBucket->hashFunc)(pBucket, d);
if (index == -1) { // the value is out of range, do not add it into bucket
return -1;
}
tMemBucketSlot *pSlot = &pBucket->pSlots[index];
tMemBucketUpdateBoundingBox(&pSlot->range, d, pBucket->type);
@ -489,64 +339,11 @@ int32_t tMemBucketPut(tMemBucket *pBucket, const void *data, size_t size) {
pSlot->info.size += 1;
}
pBucket->total += count;
return 0;
}
////////////////////////////////////////////////////////////////////////////////////////////
static UNUSED_FUNC void findMaxMinValue(tMemBucket *pMemBucket, double *maxVal, double *minVal) {
*minVal = DBL_MAX;
*maxVal = -DBL_MAX;
for (int32_t i = 0; i < pMemBucket->numOfSlots; ++i) {
tMemBucketSlot *pSlot = &pMemBucket->pSlots[i];
if (pSlot->info.size == 0) {
continue;
}
switch (pMemBucket->type) {
case TSDB_DATA_TYPE_INT:
case TSDB_DATA_TYPE_SMALLINT:
case TSDB_DATA_TYPE_TINYINT: {
double minv = pSlot->range.iMinVal;
double maxv = pSlot->range.iMaxVal;
if (*minVal > minv) {
*minVal = minv;
}
if (*maxVal < maxv) {
*maxVal = maxv;
}
break;
}
case TSDB_DATA_TYPE_DOUBLE:
case TSDB_DATA_TYPE_FLOAT: {
double minv = pSlot->range.dMinVal;
double maxv = pSlot->range.dMaxVal;
if (*minVal > minv) {
*minVal = minv;
}
if (*maxVal < maxv) {
*maxVal = maxv;
}
break;
}
case TSDB_DATA_TYPE_BIGINT: {
double minv = (double)pSlot->range.i64MinVal;
double maxv = (double)pSlot->range.i64MaxVal;
if (*minVal > minv) {
*minVal = minv;
}
if (*maxVal < maxv) {
*maxVal = maxv;
}
break;
}
}
}
}
/*
*
* now, we need to find the minimum value of the next slot for
@ -565,7 +362,6 @@ static MinMaxEntry getMinMaxEntryOfNextSlotWithData(tMemBucket *pMemBucket, int3
}
static bool isIdenticalData(tMemBucket *pMemBucket, int32_t index);
char *getFirstElemOfMemBuffer(tMemBucketSlot *pSeg, int32_t slotIdx, tFilePage *pPage);
static double getIdenticalDataVal(tMemBucket* pMemBucket, int32_t slotIndex) {
assert(isIdenticalData(pMemBucket, slotIndex));
@ -573,24 +369,12 @@ static double getIdenticalDataVal(tMemBucket* pMemBucket, int32_t slotIndex) {
tMemBucketSlot *pSlot = &pMemBucket->pSlots[slotIndex];
double finalResult = 0.0;
switch (pMemBucket->type) {
case TSDB_DATA_TYPE_SMALLINT:
case TSDB_DATA_TYPE_TINYINT:
case TSDB_DATA_TYPE_INT: {
finalResult = pSlot->range.iMinVal;
break;
}
case TSDB_DATA_TYPE_FLOAT:
case TSDB_DATA_TYPE_DOUBLE: {
finalResult = pSlot->range.dMinVal;
break;
};
case TSDB_DATA_TYPE_BIGINT: {
finalResult = (double)pSlot->range.i64MinVal;
break;
}
if (IS_SIGNED_NUMERIC_TYPE(pMemBucket->type)) {
finalResult = (double) pSlot->range.i64MinVal;
} else if (IS_UNSIGNED_NUMERIC_TYPE(pMemBucket->type)) {
finalResult = (double) pSlot->range.u64MinVal;
} else {
finalResult = (double) pSlot->range.dMinVal;
}
return finalResult;
@ -616,26 +400,16 @@ double getPercentileImpl(tMemBucket *pMemBucket, int32_t count, double fraction)
double maxOfThisSlot = 0;
double minOfNextSlot = 0;
switch (pMemBucket->type) {
case TSDB_DATA_TYPE_INT:
case TSDB_DATA_TYPE_SMALLINT:
case TSDB_DATA_TYPE_TINYINT: {
maxOfThisSlot = pSlot->range.iMaxVal;
minOfNextSlot = next.iMinVal;
break;
};
case TSDB_DATA_TYPE_FLOAT:
case TSDB_DATA_TYPE_DOUBLE: {
maxOfThisSlot = pSlot->range.dMaxVal;
minOfNextSlot = next.dMinVal;
break;
};
case TSDB_DATA_TYPE_BIGINT: {
maxOfThisSlot = (double)pSlot->range.i64MaxVal;
minOfNextSlot = (double)next.i64MinVal;
break;
}
};
if (IS_SIGNED_NUMERIC_TYPE(pMemBucket->type)) {
maxOfThisSlot = (double) pSlot->range.i64MaxVal;
minOfNextSlot = (double) next.i64MinVal;
} else if (IS_UNSIGNED_NUMERIC_TYPE(pMemBucket->type)) {
maxOfThisSlot = (double) pSlot->range.u64MaxVal;
minOfNextSlot = (double) next.u64MinVal;
} else {
maxOfThisSlot = (double) pSlot->range.dMaxVal;
minOfNextSlot = (double) next.dMinVal;
}
assert(minOfNextSlot > maxOfThisSlot);
@ -652,38 +426,8 @@ double getPercentileImpl(tMemBucket *pMemBucket, int32_t count, double fraction)
char *nextVal = thisVal + pMemBucket->bytes;
double td = 1.0, nd = 1.0;
switch (pMemBucket->type) {
case TSDB_DATA_TYPE_SMALLINT: {
td = *(int16_t *)thisVal;
nd = *(int16_t *)nextVal;
break;
}
case TSDB_DATA_TYPE_TINYINT: {
td = *(int8_t *)thisVal;
nd = *(int8_t *)nextVal;
break;
}
case TSDB_DATA_TYPE_INT: {
td = *(int32_t *)thisVal;
nd = *(int32_t *)nextVal;
break;
};
case TSDB_DATA_TYPE_FLOAT: {
td = GET_FLOAT_VAL(thisVal);
nd = GET_FLOAT_VAL(nextVal);
break;
}
case TSDB_DATA_TYPE_DOUBLE: {
td = GET_DOUBLE_VAL(thisVal);
nd = GET_DOUBLE_VAL(nextVal);
break;
}
case TSDB_DATA_TYPE_BIGINT: {
td = (double)*(int64_t *)thisVal;
nd = (double)*(int64_t *)nextVal;
break;
}
}
GET_TYPED_DATA(td, double, pMemBucket->type, thisVal);
GET_TYPED_DATA(nd, double, pMemBucket->type, nextVal);
double val = (1 - fraction) * td + fraction * nd;
tfree(buffer);
@ -696,7 +440,7 @@ double getPercentileImpl(tMemBucket *pMemBucket, int32_t count, double fraction)
// try next round
pMemBucket->times += 1;
uDebug("MemBucket:%p, start next round data bucketing, time:%d", pMemBucket, pMemBucket->times);
qDebug("MemBucket:%p, start next round data bucketing, time:%d", pMemBucket, pMemBucket->times);
pMemBucket->range = pSlot->range;
pMemBucket->total = 0;
@ -741,20 +485,14 @@ double getPercentile(tMemBucket *pMemBucket, double percent) {
if (fabs(percent - 100.0) < DBL_EPSILON || (percent < DBL_EPSILON)) {
MinMaxEntry* pRange = &pMemBucket->range;
switch(pMemBucket->type) {
case TSDB_DATA_TYPE_TINYINT:
case TSDB_DATA_TYPE_SMALLINT:
case TSDB_DATA_TYPE_INT:
return fabs(percent - 100) < DBL_EPSILON? pRange->iMaxVal:pRange->iMinVal;
case TSDB_DATA_TYPE_BIGINT: {
double v = (double)(fabs(percent - 100) < DBL_EPSILON ? pRange->i64MaxVal : pRange->i64MinVal);
return v;
}
case TSDB_DATA_TYPE_FLOAT:
case TSDB_DATA_TYPE_DOUBLE:
return fabs(percent - 100) < DBL_EPSILON? pRange->dMaxVal:pRange->dMinVal;
default:
return -1;
if (IS_SIGNED_NUMERIC_TYPE(pMemBucket->type)) {
double v = (double)(fabs(percent - 100) < DBL_EPSILON ? pRange->i64MaxVal : pRange->i64MinVal);
return v;
} else if (IS_UNSIGNED_NUMERIC_TYPE(pMemBucket->type)) {
double v = (double)(fabs(percent - 100) < DBL_EPSILON ? pRange->u64MaxVal : pRange->u64MinVal);
return v;
} else {
return fabs(percent - 100) < DBL_EPSILON? pRange->dMaxVal:pRange->dMinVal;
}
}
@ -771,40 +509,9 @@ double getPercentile(tMemBucket *pMemBucket, double percent) {
bool isIdenticalData(tMemBucket *pMemBucket, int32_t index) {
tMemBucketSlot *pSeg = &pMemBucket->pSlots[index];
if (pMemBucket->type == TSDB_DATA_TYPE_INT || pMemBucket->type == TSDB_DATA_TYPE_BIGINT ||
pMemBucket->type == TSDB_DATA_TYPE_SMALLINT || pMemBucket->type == TSDB_DATA_TYPE_TINYINT) {
if (IS_FLOAT_TYPE(pMemBucket->type)) {
return fabs(pSeg->range.dMaxVal - pSeg->range.dMinVal) < DBL_EPSILON;
} else {
return pSeg->range.i64MinVal == pSeg->range.i64MaxVal;
}
if (pMemBucket->type == TSDB_DATA_TYPE_FLOAT || pMemBucket->type == TSDB_DATA_TYPE_DOUBLE) {
return fabs(pSeg->range.dMaxVal - pSeg->range.dMinVal) < DBL_EPSILON;
}
return false;
}
/*
* get the first element of one slot into memory.
* if no data of current slot in memory, load it from disk
*/
char *getFirstElemOfMemBuffer(tMemBucketSlot *pSeg, int32_t slotIdx, tFilePage *pPage) {
// STSBuf *pMemBuffer = pSeg->pBuffer[slotIdx];
char *thisVal = NULL;
// if (pSeg->pBuffer[slotIdx]->numOfTotal != 0) {
//// thisVal = pSeg->pBuffer[slotIdx]->pHead->item.data;
// } else {
// /*
// * no data in memory, load one page into memory
// */
// tFlushoutInfo *pFlushInfo = &pMemBuffer->fileMeta.flushoutData.pFlushoutInfo[0];
// assert(pFlushInfo->numOfPages == pMemBuffer->fileMeta.nFileSize);
// int32_t ret;
// ret = fseek(pMemBuffer->file, pFlushInfo->startPageId * pMemBuffer->pageSize, SEEK_SET);
// UNUSED(ret);
// size_t sz = fread(pPage, pMemBuffer->pageSize, 1, pMemBuffer->file);
// UNUSED(sz);
// thisVal = pPage->data;
// }
return thisVal;
}

2
src/query/tests/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
FIND_PATH(HEADER_GTEST_INCLUDE_DIR gtest.h /usr/include/gtest /usr/local/include/gtest)

254
src/query/tests/percentileTest.cpp Normal file
View File

@ -0,0 +1,254 @@
#include <gtest/gtest.h>
#include <iostream>
#include "qResultbuf.h"
#include "taos.h"
#include "taosdef.h"
#include "qPercentile.h"
namespace {
tMemBucket *createBigIntDataBucket(int32_t start, int32_t end) {
tMemBucket *pBucket = tMemBucketCreate(sizeof(int64_t), TSDB_DATA_TYPE_BIGINT, start, end);
for (int32_t i = start; i <= end; ++i) {
int64_t val = i;
tMemBucketPut(pBucket, &val, 1);
}
return pBucket;
}
tMemBucket *createIntDataBucket(int32_t start, int32_t end) {
tMemBucket *pBucket = tMemBucketCreate(sizeof(int32_t), TSDB_DATA_TYPE_INT, start, end);
for (int32_t i = start; i <= end; ++i) {
int32_t val = i;
tMemBucketPut(pBucket, &val, 1);
}
return pBucket;
}
tMemBucket *createDoubleDataBucket(int32_t start, int32_t end) {
tMemBucket *pBucket = tMemBucketCreate(sizeof(double), TSDB_DATA_TYPE_DOUBLE, start, end);
for (int32_t i = start; i <= end; ++i) {
double val = i;
int32_t ret = tMemBucketPut(pBucket, &val, 1);
if (ret != 0) {
printf("value out of range:%f", val);
}
}
return pBucket;
}
tMemBucket *createUnsignedDataBucket(int32_t start, int32_t end, int32_t type) {
tMemBucket *pBucket = tMemBucketCreate(tDataTypeDesc[type].nSize, type, start, end);
for (int32_t i = start; i <= end; ++i) {
uint64_t k = i;
int32_t ret = tMemBucketPut(pBucket, &k, 1);
if (ret != 0) {
printf("value out of range:%f", k);
}
}
return pBucket;
}
void intDataTest() {
printf("running %s\n", __FUNCTION__);
tMemBucket *pBucket = NULL;
double result = 0.;
pBucket = createIntDataBucket(0, 0);
result = getPercentile(pBucket, 0);
ASSERT_DOUBLE_EQ(result, 0);
tMemBucketDestroy(pBucket);
pBucket = createIntDataBucket(0, 1);
result = getPercentile(pBucket, 100);
ASSERT_DOUBLE_EQ(result, 1);
result = getPercentile(pBucket, 0);
ASSERT_DOUBLE_EQ(result, 0);
tMemBucketDestroy(pBucket);
pBucket = createIntDataBucket(-1, 1);
result = getPercentile(pBucket, 50);
ASSERT_DOUBLE_EQ(result, 0);
result = getPercentile(pBucket, 0);
ASSERT_DOUBLE_EQ(result, -1);
result = getPercentile(pBucket, 75);
ASSERT_DOUBLE_EQ(result, 0.5);
result = getPercentile(pBucket, 100);
ASSERT_DOUBLE_EQ(result, 1);
tMemBucketDestroy(pBucket);
pBucket = createIntDataBucket(0, 99999);
result = getPercentile(pBucket, 50);
ASSERT_DOUBLE_EQ(result, 49999.5);
tMemBucketDestroy(pBucket);
}
void bigintDataTest() {
printf("running %s\n", __FUNCTION__);
tMemBucket *pBucket = NULL;
double result = 0.0;
pBucket = createBigIntDataBucket(-1000, 1000);
result = getPercentile(pBucket, 50);
ASSERT_DOUBLE_EQ(result, 0.);
tMemBucketDestroy(pBucket);
pBucket = createBigIntDataBucket(-10000, 10000);
result = getPercentile(pBucket, 100);
ASSERT_DOUBLE_EQ(result, 10000.0);
tMemBucketDestroy(pBucket);
pBucket = createBigIntDataBucket(-10000, 10000);
result = getPercentile(pBucket, 75);
ASSERT_DOUBLE_EQ(result, 5000.0);
tMemBucketDestroy(pBucket);
}
void doubleDataTest() {
printf("running %s\n", __FUNCTION__);
tMemBucket *pBucket = NULL;
double result = 0;
pBucket = createDoubleDataBucket(-10, 10);
result = getPercentile(pBucket, 0);
ASSERT_DOUBLE_EQ(result, -10.0);
printf("result is: %lf\n", result);
tMemBucketDestroy(pBucket);
pBucket = createDoubleDataBucket(-100000, 100000);
result = getPercentile(pBucket, 25);
ASSERT_DOUBLE_EQ(result, -50000);
printf("result is: %lf\n", result);
tMemBucketDestroy(pBucket);
pBucket = createDoubleDataBucket(-100000, 100000);
result = getPercentile(pBucket, 50);
ASSERT_DOUBLE_EQ(result, 0);
tMemBucketDestroy(pBucket);
pBucket = createDoubleDataBucket(-100000, 100000);
result = getPercentile(pBucket, 75);
ASSERT_DOUBLE_EQ(result, 50000);
tMemBucketDestroy(pBucket);
pBucket = createDoubleDataBucket(-100000, 100000);
result = getPercentile(pBucket, 100);
ASSERT_DOUBLE_EQ(result, 100000.0);
printf("result is: %lf\n", result);
tMemBucketDestroy(pBucket);
}
/*
* large data test, we employ 0.1billion double data to calculated the percentile
* which is 800MB data
*/
void largeDataTest() {
printf("running : %s\n", __FUNCTION__);
tMemBucket *pBucket = NULL;
double result = 0;
struct timeval tv;
gettimeofday(&tv, NULL);
int64_t start = tv.tv_sec;
printf("start time: %" PRId64 "\n", tv.tv_sec);
pBucket = createDoubleDataBucket(0, 100000000);
result = getPercentile(pBucket, 50);
ASSERT_DOUBLE_EQ(result, 50000000);
gettimeofday(&tv, NULL);
printf("total elapsed time: %" PRId64 " sec.", -start + tv.tv_sec);
printf("the result of %d is: %lf\n", 50, result);
tMemBucketDestroy(pBucket);
}
void qsortTest() {
printf("running : %s\n", __FUNCTION__);
SSchema field[1] = {
{TSDB_DATA_TYPE_INT, "k", sizeof(int32_t)},
};
const int32_t num = 2000;
int32_t *d = (int32_t *)malloc(sizeof(int32_t) * num);
for (int32_t i = 0; i < num; ++i) {
d[i] = i % 4;
}
const int32_t numOfOrderCols = 1;
int32_t orderColIdx = 0;
SColumnModel * pModel = createColumnModel(field, 1, 1000);
tOrderDescriptor *pDesc = tOrderDesCreate(&orderColIdx, numOfOrderCols, pModel, 1);
tColDataQSort(pDesc, num, 0, num - 1, (char *)d, 1);
for (int32_t i = 0; i < num; ++i) {
printf("%d\t", d[i]);
}
printf("\n");
destroyColumnModel(pModel);
}
void unsignedDataTest() {
printf("running %s\n", __FUNCTION__);
tMemBucket *pBucket = NULL;
double result = 0.0;
pBucket = createUnsignedDataBucket(0, 1000, TSDB_DATA_TYPE_UINT);
result = getPercentile(pBucket, 50);
ASSERT_DOUBLE_EQ(result, 500.0);
tMemBucketDestroy(pBucket);
pBucket = createUnsignedDataBucket(0, 10000, TSDB_DATA_TYPE_UBIGINT);
result = getPercentile(pBucket, 100);
ASSERT_DOUBLE_EQ(result, 10000.0);
result = getPercentile(pBucket, 0);
ASSERT_DOUBLE_EQ(result, 0.0);
result = getPercentile(pBucket, 50);
ASSERT_DOUBLE_EQ(result, 5000);
result = getPercentile(pBucket, 75);
ASSERT_DOUBLE_EQ(result, 7500);
tMemBucketDestroy(pBucket);
}
} // namespace
TEST(testCase, percentileTest) {
// qsortTest();
intDataTest();
bigintDataTest();
doubleDataTest();
unsignedDataTest();
largeDataTest();
}

2
src/rpc/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(inc)

2
src/rpc/src/rpcTcp.c
View File

@ -241,7 +241,7 @@ static void *taosAcceptTcpConnection(void *arg) {
}
taosKeepTcpAlive(connFd);
struct timeval to={1, 0};
struct timeval to={5, 0};
int32_t ret = taosSetSockOpt(connFd, SOL_SOCKET, SO_RCVTIMEO, &to, sizeof(to));
if (ret != 0) {
taosCloseSocket(connFd);

2
src/rpc/test/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/rpc/inc)

2
src/sync/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(inc)

2
src/sync/test/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
IF (TD_LINUX)

2
src/tsdb/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(inc)

2
src/util/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/rpc/inc)

55
src/util/src/tcompare.c
View File

@ -30,24 +30,32 @@ int32_t compareInt8Val(const void *pLeft, const void *pRight) {
return 0;
}
int32_t compareIntDoubleVal(const void *pLeft, const void *pRight) {
int64_t lhs = GET_INT64_VAL(pLeft);
double rhs = GET_DOUBLE_VAL(pRight);
if (fabs(lhs - rhs) < FLT_EPSILON) {
return 0;
} else {
return (lhs > rhs) ? 1 : -1;
}
int32_t compareUint32Val(const void *pLeft, const void *pRight) {
int32_t left = GET_UINT32_VAL(pLeft), right = GET_UINT32_VAL(pRight);
if (left > right) return 1;
if (left < right) return -1;
return 0;
}
int32_t compareDoubleIntVal(const void *pLeft, const void *pRight) {
double lhs = GET_DOUBLE_VAL(pLeft);
int64_t rhs = GET_INT64_VAL(pRight);
if (fabs(lhs - rhs) < FLT_EPSILON) {
return 0;
} else {
return (lhs > rhs) ? 1 : -1;
}
int32_t compareUint64Val(const void *pLeft, const void *pRight) {
int64_t left = GET_UINT64_VAL(pLeft), right = GET_UINT64_VAL(pRight);
if (left > right) return 1;
if (left < right) return -1;
return 0;
}
int32_t compareUint16Val(const void *pLeft, const void *pRight) {
int16_t left = GET_UINT16_VAL(pLeft), right = GET_UINT16_VAL(pRight);
if (left > right) return 1;
if (left < right) return -1;
return 0;
}
int32_t compareUint8Val(const void* pLeft, const void* pRight) {
uint8_t left = GET_UINT8_VAL(pLeft), right = GET_UINT8_VAL(pRight);
if (left > right) return 1;
if (left < right) return -1;
return 0;
}
int32_t compareFloatVal(const void *pLeft, const void *pRight) {
@ -369,15 +377,24 @@ __compar_fn_t getKeyComparFunc(int32_t keyType) {
case TSDB_DATA_TYPE_DOUBLE:
comparFn = compareDoubleVal;
break;
case TSDB_DATA_TYPE_UTINYINT:
comparFn = compareUint8Val;
break;
case TSDB_DATA_TYPE_USMALLINT:
comparFn = compareUint16Val;
break;
case TSDB_DATA_TYPE_UINT:
comparFn = compareUint32Val;
break;
case TSDB_DATA_TYPE_UBIGINT:
comparFn = compareUint64Val;
break;
case TSDB_DATA_TYPE_BINARY:
comparFn = compareLenPrefixedStr;
break;
case TSDB_DATA_TYPE_NCHAR:
comparFn = compareLenPrefixedWStr;
break;
default:
comparFn = compareInt32Val;
break;

13
src/util/src/tsocket.c
View File

@ -283,6 +283,7 @@ SOCKET taosOpenTcpClientSocket(uint32_t destIp, uint16_t destPort, uint32_t clie
SOCKET sockFd = 0;
int32_t ret;
struct sockaddr_in serverAddr, clientAddr;
int32_t bufSize = 1024 * 1024;
sockFd = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
@ -300,6 +301,18 @@ SOCKET taosOpenTcpClientSocket(uint32_t destIp, uint16_t destPort, uint32_t clie
return -1;
}
if (taosSetSockOpt(sockFd, SOL_SOCKET, SO_SNDBUF, (void *)&bufSize, sizeof(bufSize)) != 0) {
uError("failed to set the send buffer size for TCP socket\n");
taosCloseSocket(sockFd);
return -1;
}
if (taosSetSockOpt(sockFd, SOL_SOCKET, SO_RCVBUF, (void *)&bufSize, sizeof(bufSize)) != 0) {
uError("failed to set the receive buffer size for TCP socket\n");
taosCloseSocket(sockFd);
return -1;
}
if (clientIp != 0) {
memset((char *)&clientAddr, 0, sizeof(clientAddr));
clientAddr.sin_family = AF_INET;

2
src/util/tests/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
FIND_PATH(HEADER_GTEST_INCLUDE_DIR gtest.h /usr/include/gtest /usr/local/include/gtest)

2
src/vnode/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/query/inc)

2
src/vnode/inc/vnodeInt.h
View File

@ -67,7 +67,7 @@ typedef struct {
void * qMgmt;
char * rootDir;
tsem_t sem;
char db[TSDB_ACCT_LEN + TSDB_DB_NAME_LEN];
char db[TSDB_ACCT_ID_LEN + TSDB_DB_NAME_LEN];
pthread_mutex_t statusMutex;
} SVnodeObj;

2
src/vnode/src/vnodeWrite.c
View File

@ -233,7 +233,7 @@ static SVWriteMsg *vnodeBuildVWriteMsg(SVnodeObj *pVnode, SWalHead *pHead, int32
pWrite->rpcMsg = *pRpcMsg;
}
memcpy(pWrite->pHead, pHead, sizeof(SWalHead) + pHead->len);
memcpy(&pWrite->pHead, pHead, sizeof(SWalHead) + pHead->len);
pWrite->pVnode = pVnode;
pWrite->qtype = qtype;

2
src/wal/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(inc)

2
src/wal/test/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
IF (TD_LINUX)

2
tests/CMakeLists.txt
View File

@ -3,7 +3,7 @@
# generate release version:
# mkdir release; cd release; cmake -DCMAKE_BUILD_TYPE=Release ..
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
SET(CMAKE_C_STANDARD 11)

2
tests/comparisonTest/tdengine/CMakeLists.txt
View File

@ -1,4 +1,4 @@
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
CMAKE_MINIMUM_REQUIRED(VERSION 3.5)
PROJECT(TDengine)
IF (TD_LINUX)

49
tests/examples/C#/taosdemo/README.md
View File

@ -10,30 +10,27 @@ run C# version taosdemo
===
Usage: mono taosdemo.exe [OPTION...]
--help Show usage.
--help Show usage.
-h host, The host to connect to TDengine. Default is localhost.
-p port, The TCP/IP port number to use for the connection. Default is 0.
-u user, The user name to use when connecting to the server. Default is 'root'.
-P password, The password to use when connecting to the server. Default is 'taosdata'.
-d database, Destination database. Default is 'test'.
-a replica, Set the replica parameters of the database, Default 1, min: 1, max: 5.
-m table_prefix, Table prefix name. Default is 't'.
-s sql file, The select sql file.
-M stable, Use super table.
-o outputfile, Direct output to the named file. Default is './output.txt'.
-q query_mode, Query mode--0: SYNC, 1: ASYNC. Default is SYNC.
-b type_of_cols, data_type of columns: 'INT', 'TINYINT', 'SMALLINT', 'BIGINT', 'FLOAT', 'DOUBLE', 'BINARY'. Default is 'INT'.
-w length_of_binary, The length of data_type 'BINARY'. Only applicable when type of cols is 'BINARY'. Default is 8
-l num_of_cols_per_record, The number of columns per record. Default is 3.
-T num_of_threads, The number of threads. Default is 10.
-r num_of_records_per_req, The number of records per request. Default is 1000.
-t num_of_tables, The number of tables. Default is 1.
-n num_of_records_per_table, The number of records per table. Default is 1.
-c config_directory, Configuration directory. Default is '/etc/taos/'.
-x flag, Insert only flag.
-O order, Insert mode--0: In order, 1: Out of order. Default is in order.
-R rate, Out of order data's rate--if order=1 Default 10, min: 0, max: 50.
-D Delete data methods 0: don't delete, 1: delete by table, 2: delete by stable, 3: delete by database.
-v Print verbose output
-y Skip read key for continous test, default is not skip
-h <hostname> host, The host to connect to TDengine. Default is localhost.
-p <port> port, The TCP/IP port number to use for the connection. Default is 0.
-u <username> user, The user name to use when connecting to the server. Default is 'root'.
-P <password> password, The password to use when connecting to the server. Default is 'taosdata'.
-d <dbname> database, Destination database. Default is 'test'.
-a <replications> replica, Set the replica parameters of the database, Default 1, min: 1, max: 5.
-m <table prefix> table_prefix, Table prefix name. Default is 't'.
-M stable, Use super table.
-s <stable prefix> stable_prefix, STable prefix name. Default is 'st'
-Q <DEFAULT | command> query, Execute query command. set 'DEFAULT' means select * from each table
-T <number> num_of_threads, The number of threads. Default is 10.
-r <number> num_of_records_per_req, The number of records per request. Default is 1000.
-t <number> num_of_tables, The number of tables. Default is 1.
-n <number> num_of_records_per_table, The number of records per table. Default is 1.
-c <path> config_directory, Configuration directory. Default is '/etc/taos/'.
-x flag, Insert only flag.
-O order, Insert mode--0: In order, 1: Out of order. Default is in order.
-R <number> rate, Out of order data's rate--if order=1 Default 10, min: 0, max: 50.
-D <number> Delete data methods 0: don't delete, 1: delete by table, 2: delete by stable, 3: delete by database.
-v Print verbose output
-g Print debug output
-y Skip read key for continous test, default is not skip

367
tests/examples/C#/taosdemo/taosdemo.cs
View File

@ -34,11 +34,12 @@ namespace TDengineDriver
//sql parameters
private string dbName = "db";
private string stableName = "st";
private string stablePrefix = "st";
private string tablePrefix = "t";
private bool isInsertOnly = false;
private int queryMode = 1;
private string query = "NONE";
private short queryMode = 1;
private long recordsPerTable = 1;
private int recordsPerRequest = 1;
@ -52,12 +53,19 @@ namespace TDengineDriver
private bool useStable = false;
private short methodOfDelete = 0;
private long numOfThreads = 1;
private long rateOfOutorder = 0;
private short rateOfOutorder = 10;
private bool order = true;
private bool skipReadKey = false;
private bool verbose = false;
private bool debug = false;
static void HelpPrint(string arg, string desc)
{
string indent = " ";
Console.WriteLine("{0}{1}", indent, arg.PadRight(25)+desc);
}
static void PrintHelp(String[] argv)
{
for (int i = 0; i < argv.Length; ++i)
@ -66,59 +74,38 @@ namespace TDengineDriver
{
Console.WriteLine("Usage: mono taosdemo.exe [OPTION...]");
Console.WriteLine("");
string indent = " ";
Console.WriteLine("{0}{1}", indent, "--help Show usage.");
HelpPrint("--help", "Show usage.");
Console.WriteLine("");
Console.Write("{0}{1}", indent, "-h");
Console.Write("{0}{1}{2}\n", indent, indent, "host, The host to connect to TDengine. Default is localhost.");
Console.Write("{0}{1}", indent, "-p");
Console.Write("{0}{1}{2}\n", indent, indent, "port, The TCP/IP port number to use for the connection. Default is 0.");
Console.Write("{0}{1}", indent, "-u");
Console.Write("{0}{1}{2}\n", indent, indent, "user, The user name to use when connecting to the server. Default is 'root'.");
Console.Write("{0}{1}", indent, "-P");
Console.Write("{0}{1}{2}\n", indent, indent, "password, The password to use when connecting to the server. Default is 'taosdata'.");
Console.Write("{0}{1}", indent, "-d");
Console.Write("{0}{1}{2}\n", indent, indent, "database, Destination database. Default is 'test'.");
Console.Write("{0}{1}", indent, "-a");
Console.Write("{0}{1}{2}\n", indent, indent, "replica, Set the replica parameters of the database, Default 1, min: 1, max: 5.");
Console.Write("{0}{1}", indent, "-m");
Console.Write("{0}{1}{2}\n", indent, indent, "table_prefix, Table prefix name. Default is 't'.");
Console.Write("{0}{1}", indent, "-s");
Console.Write("{0}{1}{2}\n", indent, indent, "sql file, The select sql file.");
Console.Write("{0}{1}", indent, "-M");
Console.Write("{0}{1}{2}\n", indent, indent, "stable, Use super table.");
Console.Write("{0}{1}", indent, "-o");
Console.Write("{0}{1}{2}\n", indent, indent, "outputfile, Direct output to the named file. Default is './output.txt'.");
Console.Write("{0}{1}", indent, "-q");
Console.Write("{0}{1}{2}\n", indent, indent, "query_mode, Query mode--0: SYNC, 1: ASYNC. Default is SYNC.");
Console.Write("{0}{1}", indent, "-b");
Console.Write("{0}{1}{2}\n", indent, indent, "type_of_cols, data_type of columns: 'INT', 'TINYINT', 'SMALLINT', 'BIGINT', 'FLOAT', 'DOUBLE', 'BINARY'. Default is 'INT'.");
Console.Write("{0}{1}", indent, "-w");
Console.Write("{0}{1}{2}\n", indent, indent, "length_of_binary, The length of data_type 'BINARY'. Only applicable when type of cols is 'BINARY'. Default is 8");
Console.Write("{0}{1}", indent, "-l");
Console.Write("{0}{1}{2}\n", indent, indent, "num_of_cols_per_record, The number of columns per record. Default is 3.");
Console.Write("{0}{1}", indent, "-T");
Console.Write("{0}{1}{2}\n", indent, indent, "num_of_threads, The number of threads. Default is 10.");
Console.Write("{0}{1}", indent, "-r");
Console.Write("{0}{1}{2}\n", indent, indent, "num_of_records_per_req, The number of records per request. Default is 1000.");
Console.Write("{0}{1}", indent, "-t");
Console.Write("{0}{1}{2}\n", indent, indent, "num_of_tables, The number of tables. Default is 1.");
Console.Write("{0}{1}", indent, "-n");
Console.Write("{0}{1}{2}\n", indent, indent, "num_of_records_per_table, The number of records per table. Default is 1.");
Console.Write("{0}{1}", indent, "-c");
Console.Write("{0}{1}{2}\n", indent, indent, "config_directory, Configuration directory. Default is '/etc/taos/'.");
Console.Write("{0}{1}", indent, "-x");
Console.Write("{0}{1}{2}\n", indent, indent, "flag, Insert only flag.");
Console.Write("{0}{1}", indent, "-O");
Console.Write("{0}{1}{2}\n", indent, indent, "order, Insert mode--0: In order, 1: Out of order. Default is in order.");
Console.Write("{0}{1}", indent, "-R");
Console.Write("{0}{1}{2}\n", indent, indent, "rate, Out of order data's rate--if order=1 Default 10, min: 0, max: 50.");
Console.Write("{0}{1}", indent, "-D");
Console.Write("{0}{1}{2}\n", indent, indent, "Delete data methods 0: don't delete, 1: delete by table, 2: delete by stable, 3: delete by database.");
Console.Write("{0}{1}", indent, "-v");
Console.Write("{0}{1}{2}\n", indent, indent, "Print verbose output");
Console.Write("{0}{1}", indent, "-y");
Console.Write("{0}{1}{2}\n", indent, indent, "Skip read key for continous test, default is not skip");
HelpPrint("-h <hostname>", "host, The host to connect to TDengine. Default is localhost.");
HelpPrint("-p <port>", "port, The TCP/IP port number to use for the connection. Default is 0.");
HelpPrint("-u <username>", "user, The user name to use when connecting to the server. Default is 'root'.");
HelpPrint("-P <password>", "password, The password to use when connecting to the server. Default is 'taosdata'.");
HelpPrint("-d <dbname>", "database, Destination database. Default is 'test'.");
HelpPrint("-a <replications>", "replica, Set the replica parameters of the database, Default 1, min: 1, max: 5.");
HelpPrint("-m <table prefix>", "table_prefix, Table prefix name. Default is 't'.");
HelpPrint("-M", "stable, Use super table.");
HelpPrint("-s <stable prefix>", "stable_prefix, STable prefix name. Default is 'st'");
HelpPrint("-Q <DEFAULT | command>", "query, Execute query command. set 'DEFAULT' means select * from each table");
/* NOT SUPPORT SO FAR
HelpPrint("-o", "outputfile, Direct output to the named file. Default is './output.txt'.");
HelpPrint("-q", "query_mode, Query mode--0: SYNC, 1: ASYNC. Default is SYNC.");
HelpPrint("-b", "type_of_cols, data_type of columns: 'INT', 'TINYINT', 'SMALLINT', 'BIGINT', 'FLOAT', 'DOUBLE', 'BINARY'. Default is 'INT'.");
HelpPrint("-w", "length_of_binary, The length of data_type 'BINARY'. Only applicable when type of cols is 'BINARY'. Default is 8");
HelpPrint("-l", "num_of_cols_per_record, The number of columns per record. Default is 3.");
*/
HelpPrint("-T <number>", "num_of_threads, The number of threads. Default is 10.");
HelpPrint("-r <number>", "num_of_records_per_req, The number of records per request. Default is 1000.");
HelpPrint("-t <number>", "num_of_tables, The number of tables. Default is 1.");
HelpPrint("-n <number>", "num_of_records_per_table, The number of records per table. Default is 1.");
HelpPrint("-c <path>", "config_directory, Configuration directory. Default is '/etc/taos/'.");
HelpPrint("-x", "flag, Insert only flag.");
HelpPrint("-O", "order, Insert mode--0: In order, 1: Out of order. Default is in order.");
HelpPrint("-R <number>", "rate, Out of order data's rate--if order=1 Default 10, min: 0, max: 50.");
HelpPrint("-D <number>", "Delete data methods 0: don't delete, 1: delete by table, 2: delete by stable, 3: delete by database.");
HelpPrint("-v", "Print verbose output");
HelpPrint("-g", "Print debug output");
HelpPrint("-y", "Skip read key for continous test, default is not skip");
System.Environment.Exit(0);
}
@ -132,49 +119,56 @@ namespace TDengineDriver
user = this.GetArgumentAsString(argv, "-u", "root");
password = this.GetArgumentAsString(argv, "-P", "taosdata");
dbName = this.GetArgumentAsString(argv, "-d", "db");
stableName = this.GetArgumentAsString(argv, "-s", "st");
stablePrefix = this.GetArgumentAsString(argv, "-s", "st");
tablePrefix = this.GetArgumentAsString(argv, "-m", "t");
isInsertOnly = this.GetArgumentAsFlag(argv, "-x");
queryMode = (int)this.GetArgumentAsLong(argv, "-q", 0, 1, 0);
isInsertOnly = this.GetArgumentAsFlag(argv, "-x", true);
query = this.GetArgumentAsString(argv, "-Q", "NONE");
queryMode = (short)this.GetArgumentAsLong(argv, "-q", 0, 1, 0);
numOfTables = this.GetArgumentAsLong(argv, "-t", 1, 1000000000, 1);
batchRows = this.GetArgumentAsLong(argv, "-r", 1, 10000, 1000);
recordsPerTable = this.GetArgumentAsLong(argv, "-n", 1, 100000000000, 1);
recordsPerRequest = (int)this.GetArgumentAsLong(argv, "-r", 1, 10000, 1);
colsPerRecord = (int)this.GetArgumentAsLong(argv, "-l", 1, 1024, 3);
configDir = this.GetArgumentAsString(argv, "-c", "C:/TDengine/cfg");
useStable = this.GetArgumentAsFlag(argv, "-M");
useStable = this.GetArgumentAsFlag(argv, "-M", true);
replica = (short)this.GetArgumentAsLong(argv, "-a", 1, 5, 1);
methodOfDelete = (short)this.GetArgumentAsLong(argv, "-D", 0, 3, 0);
numOfThreads = (short)this.GetArgumentAsLong(argv, "-T", 1, 10000, 1);
order = this.GetArgumentAsFlag(argv, "-O");
rateOfOutorder = this.GetArgumentAsLong(argv, "-R", 0, 100, 0);
order = this.GetArgumentAsFlag(argv, "-O", false);
rateOfOutorder = (short)this.GetArgumentAsLong(argv, "-R", 0, 50, 10);
skipReadKey = this.GetArgumentAsFlag(argv, "-y");
verbose = this.GetArgumentAsFlag(argv, "-v");
skipReadKey = this.GetArgumentAsFlag(argv, "-y", true);
verbose = this.GetArgumentAsFlag(argv, "-v", true);
debug = this.GetArgumentAsFlag(argv, "-g", true);
Console.Write("###################################################################\n");
Console.Write("# Server IP: {0}\n", host);
Console.Write("# User: {0}\n", user);
Console.Write("# Password: {0}\n", password);
Console.Write("# Number of Columns per record: {0}\n", colsPerRecord);
Console.Write("# Number of Threads: {0}\n", numOfThreads);
Console.Write("# Number of Tables: {0}\n", numOfTables);
Console.Write("# Number of Data per Table: {0}\n", recordsPerTable);
Console.Write("# Records/Request: {0}\n", recordsPerRequest);
Console.Write("# Database name: {0}\n", dbName);
Console.Write("# Replica: {0}\n", replica);
Console.Write("# Use STable: {0}\n", useStable);
Console.Write("# Table prefix: {0}\n", tablePrefix);
Console.Write("# Data order: {0}\n", order);
Console.Write("# Data out of order rate: {0}\n", rateOfOutorder);
Console.Write("# Delete method: {0}\n", methodOfDelete);
Console.Write("# Query Mode: {0}\n", queryMode);
Console.Write("# Insert Only: {0}\n", isInsertOnly);
Console.Write("# Verbose output {0}\n", verbose);
Console.Write("# Test time: {0}\n", DateTime.Now.ToString("h:mm:ss tt"));
VerbosePrint ("###################################################################\n");
VerbosePrintFormat ("# Server IP: {0}\n", host);
VerbosePrintFormat ("# User: {0}\n", user);
VerbosePrintFormat ("# Password: {0}\n", password);
VerbosePrintFormat ("# Number of Columns per record: {0}\n", colsPerRecord);
VerbosePrintFormat ("# Number of Threads: {0}\n", numOfThreads);
VerbosePrintFormat ("# Number of Tables: {0}\n", numOfTables);
VerbosePrintFormat ("# Number of records per Table: {0}\n", recordsPerTable);
VerbosePrintFormat ("# Records/Request: {0}\n", recordsPerRequest);
VerbosePrintFormat ("# Database name: {0}\n", dbName);
VerbosePrintFormat ("# Replica: {0}\n", replica);
VerbosePrintFormat ("# Use STable: {0}\n", useStable);
VerbosePrintFormat ("# Table prefix: {0}\n", tablePrefix);
if (useStable == true)
{
VerbosePrintFormat("# STable prefix: {0}\n", stablePrefix);
}
VerbosePrintFormat ("# Data order: {0}\n", order);
VerbosePrintFormat ("# Data out of order rate: {0}\n", rateOfOutorder);
VerbosePrintFormat ("# Delete method: {0}\n", methodOfDelete);
VerbosePrintFormat ("# Query command: {0}\n", query);
VerbosePrintFormat ("# Query Mode: {0}\n", queryMode);
VerbosePrintFormat ("# Insert Only: {0}\n", isInsertOnly);
VerbosePrintFormat ("# Verbose output {0}\n", verbose);
VerbosePrintFormat ("# Test time: {0}\n", DateTime.Now.ToString("h:mm:ss tt"));
Console.Write("###################################################################\n");
VerbosePrint ("###################################################################\n");
if (skipReadKey == false)
{
@ -183,17 +177,17 @@ namespace TDengineDriver
}
}
public bool GetArgumentAsFlag(String[] argv, String argName)
public bool GetArgumentAsFlag(String[] argv, String argName, bool defaultValue)
{
int argc = argv.Length;
for (int i = 0; i < argc; ++i)
{
if (argName == argv[i])
{
return true;
return defaultValue;
}
}
return false;
return !defaultValue;
}
public long GetArgumentAsLong(String[] argv, String argName, int minVal, long maxVal, int defaultValue)
@ -210,15 +204,15 @@ namespace TDengineDriver
String tmp = argv[i + 1];
if (tmp[0] == '-')
{
Console.WriteLine("option {0:G} requires an argument", tmp);
ExitProgram();
Console.WriteLine("option {0:G} requires an argument", argName);
ExitProgram(1);
}
long tmpVal = Convert.ToInt64(tmp);
if (tmpVal < minVal || tmpVal > maxVal)
{
Console.WriteLine("option {0:G} should in range [{1:G}, {2:G}]", argName, minVal, maxVal);
ExitProgram();
Console.WriteLine("option {0:G} value should in range [{1:G}, {2:G}]", argName, minVal, maxVal);
ExitProgram(1);
}
return tmpVal;
@ -242,8 +236,8 @@ namespace TDengineDriver
String tmp = argv[i + 1];
if (tmp[0] == '-')
{
Console.WriteLine("option {0:G} requires an argument", tmp);
ExitProgram();
Console.WriteLine("option {0:G} requires an argument", argName);
ExitProgram(1);
}
return tmp;
}
@ -252,13 +246,18 @@ namespace TDengineDriver
return defaultValue;
}
static void ExitProgram()
static void CleanAndExitProgram(int ret)
{
TDengine.Cleanup();
System.Environment.Exit(0);
System.Environment.Exit(ret);
}
private void DebugPrintFormat(string format, params object[] parameters)
static void ExitProgram(int ret)
{
System.Environment.Exit(ret);
}
private void VerbosePrintFormat(string format, params object[] parameters)
{
if (verbose == true)
{
@ -266,7 +265,7 @@ namespace TDengineDriver
}
}
private void DebugPrint(string str)
private void VerbosePrint(string str)
{
if (verbose == true)
{
@ -274,28 +273,44 @@ namespace TDengineDriver
}
}
private void DebugPrintFormat(string format, params object[] parameters)
{
if (debug == true)
{
Console.Write(format, parameters);
}
}
private void DebugPrint(string str)
{
if (debug == true)
{
Console.Write(str);
}
}
public void InitTDengine()
{
TDengine.Options((int)TDengineInitOption.TDDB_OPTION_CONFIGDIR, this.configDir);
TDengine.Options((int)TDengineInitOption.TDDB_OPTION_SHELL_ACTIVITY_TIMER, "60");
TDengine.Init();
DebugPrint("TDengine Initialization finished\n");
VerbosePrint("TDengine Initialization finished\n");
}
public void ConnectTDengine()
{
string db = "";
DebugPrintFormat("host:{0} user:{1}, pass:{2}; db:{3}, port:{4}\n",
VerbosePrintFormat("host:{0} user:{1}, pass:{2}; db:{3}, port:{4}\n",
this.host, this.user, this.password, db, this.port);
this.conn = TDengine.Connect(this.host, this.user, this.password, db, this.port);
if (this.conn == IntPtr.Zero)
{
Console.WriteLine("Connect to TDengine failed");
ExitProgram();
CleanAndExitProgram(1);
}
else
{
DebugPrint("Connect to TDengine success\n");
VerbosePrint("Connect to TDengine success\n");
}
}
@ -323,12 +338,13 @@ namespace TDengineDriver
CreateTableThread createTableThread = new CreateTableThread();
createTableThread.id = i;
createTableThread.verbose = verbose;
createTableThread.debug = debug;
createTableThread.dbName = this.dbName;
createTableThread.tablePrefix = this.tablePrefix;
createTableThread.useStable = useStable;
if (useStable)
{
createTableThread.stableName = stableName;
createTableThread.stablePrefix = stablePrefix;
}
createTableThread.conn = conn;
@ -356,12 +372,12 @@ namespace TDengineDriver
IntPtr res = TDengine.Query(this.conn, sql.ToString());
if (res != IntPtr.Zero)
{
DebugPrint(sql.ToString() + " success\n");
VerbosePrint(sql.ToString() + " success\n");
}
else
{
Console.WriteLine(sql.ToString() + " failure, reason: " + TDengine.Error(res));
ExitProgram();
CleanAndExitProgram(1);
}
}
@ -373,12 +389,12 @@ namespace TDengineDriver
IntPtr res = TDengine.Query(this.conn, sql.ToString());
if (res != IntPtr.Zero)
{
DebugPrint(sql.ToString() + " success\n");
VerbosePrint(sql.ToString() + " success\n");
}
else
{
Console.WriteLine(sql.ToString() + " failure, reason: " + TDengine.Error(res));
ExitProgram();
CleanAndExitProgram(1);
}
TDengine.FreeResult(res);
}
@ -389,17 +405,17 @@ namespace TDengineDriver
sql.Clear();
sql.Append("CREATE TABLE IF NOT EXISTS ").
Append(this.dbName).Append(".").Append(this.stableName).
Append(this.dbName).Append(".").Append(this.stablePrefix).
Append("(ts timestamp, v1 bool, v2 tinyint, v3 smallint, v4 int, v5 bigint, v6 float, v7 double, v8 binary(10), v9 nchar(10)) tags(t1 int)");
IntPtr res = TDengine.Query(this.conn, sql.ToString());
if (res != IntPtr.Zero)
{
DebugPrint(sql.ToString() + " success\n");
VerbosePrint(sql.ToString() + " success\n");
}
else
{
Console.WriteLine(sql.ToString() + " failure, reason: " + TDengine.Error(res));
ExitProgram();
CleanAndExitProgram(1);
}
TDengine.FreeResult(res);
}
@ -431,11 +447,14 @@ namespace TDengineDriver
insertThread.batchRows = batchRows;
insertThread.numOfTables = numOfTables;
insertThread.verbose = verbose;
insertThread.debug = debug;
insertThread.dbName = this.dbName;
insertThread.tablePrefix = this.tablePrefix;
insertThread.order = this.order;
insertThread.rateOfOutorder = this.rateOfOutorder;
if (useStable)
{
// insertThread.stableName = stableName;
insertThread.stablePrefix = stablePrefix;
}
insertThread.conn = conn;
@ -458,33 +477,43 @@ namespace TDengineDriver
public void ExecuteQuery()
{
// System.DateTime start = new System.DateTime();
long queryRows = 0;
for (int i = 0; i < 1/*this.numOfTables*/; ++i)
for (int i = 0; i < this.numOfTables; ++i)
{
String sql = "select * from " + this.dbName + "." + tablePrefix + i;
// Console.WriteLine(sql);
string sql;
if (query == "DEFAULT")
{
sql = "select * from " + this.dbName + "." + tablePrefix + i;
}
else
{
sql = query;
}
DebugPrintFormat("query: {0}, sql:{1}\n", query, sql);
IntPtr res = TDengine.Query(conn, sql);
DebugPrintFormat("res: {0}\n", res);
if (res == IntPtr.Zero)
{
Console.WriteLine(sql + " failure, reason: " + TDengine.Error(res));
ExitProgram();
CleanAndExitProgram(1);
}
int fieldCount = TDengine.FieldCount(res);
// Console.WriteLine("field count: " + fieldCount);
DebugPrint("field count: " + fieldCount + "\n");
List<TDengineMeta> metas = TDengine.FetchFields(res);
for (int j = 0; j < metas.Count; j++)
{
TDengineMeta meta = (TDengineMeta)metas[j];
// Console.WriteLine("index:" + j + ", type:" + meta.type + ", typename:" + meta.TypeName() + ", name:" + meta.name + ", size:" + meta.size);
DebugPrint("index:" + j + ", type:" + meta.type + ", typename:" + meta.TypeName() + ", name:" + meta.name + ", size:" + meta.size + "\n");
}
IntPtr rowdata;
StringBuilder builder = new StringBuilder();
while ((rowdata = TDengine.FetchRows(res)) != IntPtr.Zero)
{
queryRows++;
@ -548,10 +577,7 @@ namespace TDengineDriver
}
builder.Append("---");
if (queryRows <= 10)
{
Console.WriteLine(builder.ToString());
}
VerbosePrint(builder.ToString() + "\n");
builder.Clear();
}
@ -563,13 +589,6 @@ namespace TDengineDriver
TDengine.FreeResult(res);
}
/*
System.DateTime end = new System.DateTime();
TimeSpan ts = end - start;
Console.Write("Total {0:G} rows inserted, {1:G} rows query, time spend {2:G} seconds.\n"
, this.rowsInserted, queryRows, ts.TotalSeconds);
*/
}
public void CloseConnection()
@ -610,13 +629,24 @@ namespace TDengineDriver
watch.Stop();
double elapsedMs = watch.Elapsed.TotalMilliseconds;
Console.WriteLine("Spent {0} seconds to insert {1} records with {2} record(s) per request: {3} records/second",
Console.WriteLine("C# taosdemo: Spent {0} seconds to insert {1} records with {2} record(s) per request: {3} records/second",
elapsedMs / 1000,
tester.recordsPerTable * tester.numOfTables,
tester.batchRows,
(tester.recordsPerTable * tester.numOfTables * 1000) / elapsedMs);
tester.ExecuteQuery();
tester.DebugPrintFormat("query command:{0}\n", tester.query);
if (tester.query != "NONE")
{
watch = Stopwatch.StartNew();
tester.ExecuteQuery();
watch.Stop();
elapsedMs = watch.Elapsed.TotalMilliseconds;
Console.WriteLine("C# taosdemo: Spent {0} seconds to query {1} records.\n",
elapsedMs/1000,
tester.recordsPerTable * tester.numOfTables
);
}
tester.CloseConnection();
Console.WriteLine("End.");
@ -630,13 +660,16 @@ namespace TDengineDriver
public string dbName { set; get; }
public IntPtr conn { set; get; }
public string tablePrefix { set; get; }
// public string stableName { set; get; }
public string stablePrefix { set; get; }
public long recordsPerTable { set; get; }
public long batchRows { set; get; }
public long numOfTables { set; get; }
public bool verbose { set; get; }
public bool debug { set; get; }
public bool order { set; get; }
public short rateOfOutorder { set; get; }
private void DebugPrintFormat(string format, params object[] parameters)
private void VerbosePrintFormat(string format, params object[] parameters)
{
if (verbose == true)
{
@ -644,7 +677,7 @@ namespace TDengineDriver
}
}
private void DebugPrint(string str)
private void VerbosePrint(string str)
{
if (verbose == true)
{
@ -652,9 +685,25 @@ namespace TDengineDriver
}
}
private void DebugPrintFormat(string format, params object[] parameters)
{
if (debug == true)
{
Console.Write(format, parameters);
}
}
private void DebugPrint(string str)
{
if (debug == true)
{
Console.Write(str);
}
}
public void ThreadMain()
{
DebugPrintFormat("InsertDataThread {0} from {1} to {2}\n", id, start, end);
VerbosePrintFormat("InsertDataThread {0} from {1} to {2}\n", id, start, end);
StringBuilder sql = new StringBuilder();
DateTime now = DateTime.Now;
@ -663,12 +712,12 @@ namespace TDengineDriver
int s = now.Second;
long baseTimestamp = 1609430400000; // 2021/01/01 0:0:0
DebugPrintFormat("beginTime is {0} + {1}h:{2}m:{3}s\n", baseTimestamp, h, m, s);
VerbosePrintFormat("beginTime is {0} + {1}h:{2}m:{3}s\n", baseTimestamp, h, m, s);
long beginTimestamp = baseTimestamp + ((h*60 + m) * 60 + s) * 1000;
Random random = new Random();
long rowsInserted = 0;
// System.DateTime startTime = new System.DateTime();
long i = 0;
while (i < recordsPerTable)
{
@ -686,8 +735,25 @@ namespace TDengineDriver
}
for (int batch = 0; batch < batchRows; ++batch)
{
long writeTimeStamp = beginTimestamp + i + batch;
int rnd = 100;
if (this.order == false)
{
rnd = random.Next(1, 100);
if (rnd <= this.rateOfOutorder)
{
writeTimeStamp = writeTimeStamp + rnd * 10000;
DebugPrint("### ");
}
DebugPrintFormat("order:{0} rnd:{1} timestamp:{2}\n", this.order, rnd, writeTimeStamp);
}
else
{
DebugPrintFormat("order:{0} timestamp:{1}\n", this.order, writeTimeStamp);
}
sql.Append("(")
.Append(beginTimestamp + i + batch)
.Append(writeTimeStamp)
.Append(", 1, 2, 3,")
.Append(i + batch)
.Append(", 5, 6, 7, 'abc', 'def')");
@ -696,7 +762,7 @@ namespace TDengineDriver
IntPtr res = TDengine.Query(this.conn, sql.ToString());
if (res == IntPtr.Zero)
{
DebugPrint(sql.ToString() + " failure, reason: " + TDengine.Error(res) + "\n");
VerbosePrint(sql.ToString() + " failure, reason: " + TDengine.Error(res) + "\n");
}
inserted += this.batchRows;
@ -723,11 +789,12 @@ namespace TDengineDriver
public string dbName { set; get; }
public IntPtr conn { set; get; }
public string tablePrefix { set; get; }
public string stableName { set; get; }
public string stablePrefix { set; get; }
public bool verbose { set; get; }
public bool debug { set; get; }
public bool useStable { set; get; }
private void DebugPrintFormat(string format, params object[] parameters)
private void VerbosePrintFormat(string format, params object[] parameters)
{
if (verbose == true)
{
@ -735,7 +802,7 @@ namespace TDengineDriver
}
}
private void DebugPrint(string str)
private void VerbosePrint(string str)
{
if (verbose == true)
{
@ -743,9 +810,17 @@ namespace TDengineDriver
}
}
private void DebugPrintFormat(string format, params object[] parameters)
{
if (debug == true)
{
Console.Write(format, parameters);
}
}
public void ThreadMain()
{
DebugPrintFormat("CreateTable {0} from {1} to {2}\n", id, start, end);
VerbosePrintFormat("CreateTable {0} from {1} to {2}\n", id, start, end);
StringBuilder sql = new StringBuilder();
@ -756,7 +831,7 @@ namespace TDengineDriver
Append(this.dbName).Append(".").Append(this.tablePrefix).Append(tableId);
if (useStable == true)
{
sql = sql.Append(" USING ").Append(this.dbName).Append(".").Append(this.stableName).
sql = sql.Append(" USING ").Append(this.dbName).Append(".").Append(this.stablePrefix).
Append(" TAGS(").Append(tableId).Append(")");
}
else
@ -766,12 +841,12 @@ namespace TDengineDriver
IntPtr res = TDengine.Query(this.conn, sql.ToString());
if (res != IntPtr.Zero)
{
DebugPrint(sql.ToString() + " success\n");
VerbosePrint(sql.ToString() + " success\n");
}
else
{
DebugPrint(sql.ToString() + " failure, reason: " + TDengine.Error(res) + "\n");
ExitProgram();
VerbosePrint(sql.ToString() + " failure, reason: " + TDengine.Error(res) + "\n");
CleanAndExitProgram(1);
}
TDengine.FreeResult(res);
}

2
tests/pytest/alter/alter_table.py
View File

@ -131,7 +131,7 @@ class TDTestCase:
tdSql.execute("alter table t0 set tag t1=2.1")
tdSql.query("show tables")
tdSql.checkRows(1)
tdSql.checkRows(2)
def stop(self):
tdSql.close()

1
tests/pytest/fulltest.sh
View File

@ -190,6 +190,7 @@ python3 ./test.py -f stream/table_n.py
#alter table
python3 ./test.py -f alter/alter_table_crash.py
python3 ./test.py -f alter/alter_table.py
# client
python3 ./test.py -f client/client.py

51
tests/pytest/multilevel/basic.py Normal file
View File

@ -0,0 +1,51 @@
###################################################################
# 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 sys
import taos
from util.log import *
from util.cases import *
from util.sql import *
from util.dnodes import *
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
def run(self):
cfg={
'/mnt/data1' : 'dataDir',
'/mnt/data2 0 0' : 'dataDir'
}
tdSql.createDir('/mnt/data1')
tdSql.createDir('/mnt/data2')
tdLog.info("================= step1")
tdDnodes.stop(1)
tdDnodes.deploy(1,cfg)
tdDnodes.startWithoutSleep(1)
tdLog.info("================= step2")
tdSql.haveFile('/mnt/data1',1)
tdSql.haveFile('/mnt/data2',1)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

50
tests/pytest/multilevel/dirNotExist.py Normal file
View File

@ -0,0 +1,50 @@
###################################################################
# 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 sys
import taos
from util.log import *
from util.cases import *
from util.sql import *
from util.dnodes import *
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
def run(self):
cfg={
'/mnt/data1 0 0' : 'dataDir',
'/mnt/data2 0 0' : 'dataDir'
}
tdSql.createDir('/mnt/data1')
os.system('rm -rf /mnt/data2')
tdLog.info("================= step1")
tdDnodes.stop(1)
tdDnodes.deploy(1,cfg)
tdDnodes.startWithoutSleep(1)
tdLog.info("================= step2")
tdSql.taosdStatus(0)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

51
tests/pytest/multilevel/dirPermissionDenied.py Normal file
View File

@ -0,0 +1,51 @@
###################################################################
# 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 sys
import taos
from util.log import *
from util.cases import *
from util.sql import *
from util.dnodes import *
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
def run(self):
cfg={
'/mnt/data1 0 0' : 'dataDir',
'/mnt/data2 0 0' : 'dataDir'
}
tdSql.createDir('/mnt/data1')
tdSql.createDir('/mnt/data2')
os.system('chmod 111 /mnt/data2')
tdLog.info("================= step1")
tdDnodes.stop(1)
tdDnodes.deploy(1,cfg)
tdDnodes.startWithoutSleep(1)
tdLog.info("================= step2")
tdSql.taosdStatus(0)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

71
tests/pytest/multilevel/fileDistributionSameLevel.py Normal file
View File

@ -0,0 +1,71 @@
###################################################################
# 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 sys
import taos
from util.log import *
from util.cases import *
from util.sql import *
from util.dnodes import *
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
def run(self):
self.ntables = 10
self.rowsPerTable = 10
self.ts = 1520000010000
tdDnodes.stop(1)
# Test1 1 dataDir
cfg={
'10' : 'maxVgroupsPerDb',
'/mnt/data00 0 1' : 'dataDir',
'/mnt/data01 0 0' : 'dataDir',
'/mnt/data02 0 0' : 'dataDir',
'/mnt/data03 1 0' : 'dataDir',
'/mnt/data04 1 0' : 'dataDir'
}
tdSql.createDir('/mnt/data00')
tdSql.createDir('/mnt/data01')
tdSql.createDir('/mnt/data02')
tdSql.createDir('/mnt/data03')
tdSql.createDir('/mnt/data04')
tdDnodes.deploy(1,cfg)
tdDnodes.startWithoutSleep(1)
tdSql.execute("create database test days 1")
tdSql.execute("use test")
tdSql.execute("create table tb(ts timestamp, c int)")
for i in range(self.rowsPerTable):
tdSql.execute("insert into tb values(%d, 1)" % (self.ts + i * 86400000))
tdDnodes.stop(1)
tdDnodes.start(1)
tdSql.query("select * from test.tb")
tdSql.checkRows(10)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

59
tests/pytest/multilevel/illegelLevelMountPoint.py Normal file
View File

@ -0,0 +1,59 @@
###################################################################
# 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 sys
import taos
from util.log import *
from util.cases import *
from util.sql import *
from util.dnodes import *
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
def run(self):
self.ntables = 10
self.rowsPerTable = 10
self.startTime = 1520000010000
tdDnodes.stop(1)
# Test1 1 dataDir
cfg={
'/mnt/data1 3 0' : 'dataDir'
}
tdSql.createDir('/mnt/data1')
tdDnodes.deploy(1,cfg)
tdDnodes.startWithoutSleep(1)
tdDnodes.stop(1)
# Test1 1 dataDir
cfg={
'/mnt/data1 -1 0' : 'dataDir'
}
tdSql.createDir('/mnt/data1')
tdDnodes.deploy(1,cfg)
tdDnodes.startWithoutSleep(1)
tdSql.taosdStatus(0)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

64
tests/pytest/multilevel/middleLevelMissing.py Normal file
View File

@ -0,0 +1,64 @@
###################################################################
# 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 sys
import taos
from util.log import *
from util.cases import *
from util.sql import *
from util.dnodes import *
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
def run(self):
self.ntables = 10
self.rowsPerTable = 10
self.startTime = 1520000010000
tdDnodes.stop(1)
# Test1 1 dataDir
cfg={
'/mnt/data1 1 0' : 'dataDir'
}
tdSql.createDir('/mnt/data1')
tdDnodes.deploy(1,cfg)
tdDnodes.startWithoutSleep(1)
tdSql.taosdStatus(0)
tdDnodes.stop(1)
# Test2 2 dataDir
cfg={
'/mnt/data1 0 1' : 'dataDir',
'/mnt/data1 2 0' : 'dataDir'
}
tdSql.createDir('/mnt/data1')
tdDnodes.deploy(1,cfg)
tdDnodes.startWithoutSleep(1)
tdSql.taosdStatus(0)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

Some files were not shown because too many files have changed in this diff Show More