提交 8fe45602 编写于 作者: S Shuduo Sang

merge develop branch and fix conflict in crash_gen.py.

......@@ -164,7 +164,7 @@ matrix:
script:
- cmake .. > /dev/null
- make > /dev/null
- make
- os: linux
dist: bionic
......
......@@ -18,7 +18,7 @@ import (
"sync"
"time"
_ "github.com/taosdata/TDengine/src/connector/go/taosSql"
_ "github.com/taosdata/driver-go/taosSql"
)
const (
......
......@@ -87,6 +87,16 @@ typedef struct SVgroupTableInfo {
SArray* itemList; //SArray<STableIdInfo>
} SVgroupTableInfo;
static FORCE_INLINE SQueryInfo* tscGetQueryInfoDetail(SSqlCmd* pCmd, int32_t subClauseIndex) {
assert(pCmd != NULL && subClauseIndex >= 0 && subClauseIndex < TSDB_MAX_UNION_CLAUSE);
if (pCmd->pQueryInfo == NULL || subClauseIndex >= pCmd->numOfClause) {
return NULL;
}
return pCmd->pQueryInfo[subClauseIndex];
}
int32_t tscCreateDataBlock(size_t initialSize, int32_t rowSize, int32_t startOffset, const char* name,
STableMeta* pTableMeta, STableDataBlocks** dataBlocks);
void tscDestroyDataBlock(STableDataBlocks* pDataBlock);
......
......@@ -320,6 +320,8 @@ typedef struct SSqlStream {
SSqlObj *pSql;
uint32_t streamId;
char listed;
bool isProject;
int16_t precision;
int64_t num; // number of computing count
/*
......@@ -334,7 +336,6 @@ typedef struct SSqlStream {
int64_t etime; // stream end query time, when time is larger then etime, the stream will be closed
int64_t interval;
int64_t slidingTime;
int16_t precision;
void * pTimer;
void (*fp)();
......
......@@ -477,7 +477,13 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
tscDebug("%p redo parse sql string to build submit block", pSql);
pCmd->parseFinished = false;
if ((code = tsParseSql(pSql, true)) == TSDB_CODE_SUCCESS) {
code = tsParseSql(pSql, true);
if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
return;
}
if (code == TSDB_CODE_SUCCESS) {
/*
* Discard previous built submit blocks, and then parse the sql string again and build up all submit blocks,
* and send the required submit block according to index value in supporter to server.
......
......@@ -340,13 +340,12 @@ bool stableQueryFunctChanged(int32_t funcId) {
*/
void resetResultInfo(SResultInfo *pResInfo) { pResInfo->initialized = false; }
void setResultInfoBuf(SResultInfo *pResInfo, int32_t size, bool superTable) {
void setResultInfoBuf(SResultInfo *pResInfo, int32_t size, bool superTable, char* buf) {
assert(pResInfo->interResultBuf == NULL);
pResInfo->bufLen = size;
pResInfo->superTableQ = superTable;
pResInfo->interResultBuf = calloc(1, (size_t)size);
pResInfo->interResultBuf = buf;
}
// set the query flag to denote that query is completed
......
......@@ -406,7 +406,7 @@ int tscProcessLocalCmd(SSqlObj *pSql) {
pSql->res.qhandle = 0x1;
pSql->res.numOfRows = 0;
} else if (pCmd->command == TSDB_SQL_RESET_CACHE) {
taosCacheEmpty(tscCacheHandle,false);
taosCacheEmpty(tscCacheHandle);
} else if (pCmd->command == TSDB_SQL_SERV_VERSION) {
tscProcessServerVer(pSql);
} else if (pCmd->command == TSDB_SQL_CLI_VERSION) {
......
......@@ -1310,6 +1310,11 @@ int tsParseSql(SSqlObj *pSql, bool initial) {
tscDebug("%p resume to parse sql: %s", pSql, pCmd->curSql);
}
ret = tscAllocPayload(&pSql->cmd, TSDB_DEFAULT_PAYLOAD_SIZE);
if (TSDB_CODE_SUCCESS != ret) {
return ret;
}
if (tscIsInsertData(pSql->sqlstr)) {
/*
* Set the fp before parse the sql string, in case of getTableMeta failed, in which
......@@ -1326,11 +1331,6 @@ int tsParseSql(SSqlObj *pSql, bool initial) {
ret = tsParseInsertSql(pSql);
} else {
ret = tscAllocPayload(&pSql->cmd, TSDB_DEFAULT_PAYLOAD_SIZE);
if (TSDB_CODE_SUCCESS != ret) {
return ret;
}
SSqlInfo SQLInfo = qSQLParse(pSql->sqlstr);
ret = tscToSQLCmd(pSql, &SQLInfo);
SQLInfoDestroy(&SQLInfo);
......
......@@ -90,6 +90,7 @@ static int32_t validateSqlFunctionInStreamSql(SSqlCmd* pCmd, SQueryInfo* pQueryI
static int32_t buildArithmeticExprString(tSQLExpr* pExpr, char** exprString);
static int32_t validateFunctionsInIntervalOrGroupbyQuery(SSqlCmd* pCmd, SQueryInfo* pQueryInfo);
static int32_t validateArithmeticSQLExpr(SSqlCmd* pCmd, tSQLExpr* pExpr, SQueryInfo* pQueryInfo, SColumnList* pList, int32_t* type);
static int32_t validateEp(char* ep);
static int32_t validateDNodeConfig(tDCLSQL* pOptions);
static int32_t validateLocalConfig(tDCLSQL* pOptions);
static int32_t validateColumnName(char* name);
......@@ -359,6 +360,7 @@ int32_t tscToSQLCmd(SSqlObj* pSql, struct SSqlInfo* pInfo) {
case TSDB_SQL_CFG_DNODE: {
const char* msg2 = "invalid configure options or values";
const char* msg3 = "invalid dnode ep";
/* validate the ip address */
tDCLSQL* pDCL = pInfo->pDCLInfo;
......@@ -375,6 +377,10 @@ int32_t tscToSQLCmd(SSqlObj* pSql, struct SSqlInfo* pInfo) {
strncpy(pCfg->ep, pDCL->a[0].z, pDCL->a[0].n);
if (validateEp(pCfg->ep) != TSDB_CODE_SUCCESS) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg3);
}
strncpy(pCfg->config, pDCL->a[1].z, pDCL->a[1].n);
if (pDCL->nTokens == 3) {
......@@ -654,11 +660,14 @@ int32_t parseIntervalClause(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SQuerySQL* pQ
int32_t parseSlidingClause(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SQuerySQL* pQuerySql) {
const char* msg0 = "sliding value too small";
const char* msg1 = "sliding value no larger than the interval value";
const char* msg2 = "sliding value can not less than 1% of interval value";
const static int32_t INTERVAL_SLIDING_FACTOR = 100;
STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
SSQLToken* pSliding = &pQuerySql->sliding;
STableComInfo tinfo = tscGetTableInfo(pTableMetaInfo->pTableMeta);
SSQLToken* pSliding = &pQuerySql->sliding;
if (pSliding->n != 0) {
getTimestampInUsFromStr(pSliding->z, pSliding->n, &pQueryInfo->slidingTime);
if (tinfo.precision == TSDB_TIME_PRECISION_MILLI) {
......@@ -676,6 +685,10 @@ int32_t parseSlidingClause(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SQuerySQL* pQu
pQueryInfo->slidingTime = pQueryInfo->intervalTime;
}
if ((pQueryInfo->intervalTime != 0) && (pQueryInfo->intervalTime/pQueryInfo->slidingTime > INTERVAL_SLIDING_FACTOR)) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg2);
}
return TSDB_CODE_SUCCESS;
}
......@@ -4629,6 +4642,24 @@ typedef struct SDNodeDynConfOption {
int32_t len; // name string length
} SDNodeDynConfOption;
int32_t validateEp(char* ep) {
char buf[TSDB_EP_LEN + 1] = {0};
tstrncpy(buf, ep, TSDB_EP_LEN);
char *pos = strchr(buf, ':');
if (NULL == pos) {
return TSDB_CODE_TSC_INVALID_SQL;
}
uint16_t port = atoi(pos+1);
if (0 == port) {
return TSDB_CODE_TSC_INVALID_SQL;
}
return TSDB_CODE_SUCCESS;
}
int32_t validateDNodeConfig(tDCLSQL* pOptions) {
if (pOptions->nTokens < 2 || pOptions->nTokens > 3) {
return TSDB_CODE_TSC_INVALID_SQL;
......@@ -6096,16 +6127,12 @@ int32_t exprTreeFromSqlExpr(SSqlCmd* pCmd, tExprNode **pExpr, const tSQLExpr* pS
}
}
// NOTE: binary|nchar data allows the >|< type filter
if ((*pExpr)->_node.optr != TSDB_RELATION_EQUAL && (*pExpr)->_node.optr != TSDB_RELATION_NOT_EQUAL) {
if (pRight->nodeType == TSQL_NODE_VALUE) {
if (pRight->pVal->nType == TSDB_DATA_TYPE_BOOL) {
return TSDB_CODE_TSC_INVALID_SQL;
}
if ((pRight->pVal->nType == TSDB_DATA_TYPE_BINARY || pRight->pVal->nType == TSDB_DATA_TYPE_NCHAR)
&& (*pExpr)->_node.optr != TSDB_RELATION_LIKE) {
return TSDB_CODE_TSC_INVALID_SQL;
}
}
}
}
......
......@@ -1956,7 +1956,7 @@ int tscProcessUseDbRsp(SSqlObj *pSql) {
}
int tscProcessDropDbRsp(SSqlObj *UNUSED_PARAM(pSql)) {
taosCacheEmpty(tscCacheHandle, false);
taosCacheEmpty(tscCacheHandle);
return 0;
}
......@@ -2002,7 +2002,7 @@ int tscProcessAlterTableMsgRsp(SSqlObj *pSql) {
if (isSuperTable) { // if it is a super table, reset whole query cache
tscDebug("%p reset query cache since table:%s is stable", pSql, pTableMetaInfo->name);
taosCacheEmpty(tscCacheHandle, false);
taosCacheEmpty(tscCacheHandle);
}
}
......
......@@ -71,6 +71,7 @@ static void tscProcessStreamLaunchQuery(SSchedMsg *pMsg) {
pSql->fp = tscProcessStreamQueryCallback;
pSql->param = pStream;
pSql->res.completed = false;
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
......@@ -86,7 +87,7 @@ static void tscProcessStreamLaunchQuery(SSchedMsg *pMsg) {
// failed to get meter/metric meta, retry in 10sec.
if (code != TSDB_CODE_SUCCESS) {
int64_t retryDelayTime = tscGetRetryDelayTime(pStream->slidingTime, pStream->precision);
tscError("%p stream:%p,get metermeta failed, retry in %" PRId64 "ms", pStream->pSql, pStream, retryDelayTime);
tscDebug("%p stream:%p,get metermeta failed, retry in %" PRId64 "ms", pStream->pSql, pStream, retryDelayTime);
tscSetRetryTimer(pStream, pSql, retryDelayTime);
} else {
......@@ -108,7 +109,7 @@ static void tscProcessStreamTimer(void *handle, void *tmrId) {
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
tscDebug("%p add into timer", pSql);
if (isProjectStream(pQueryInfo)) {
if (pStream->isProject) {
/*
* pQueryInfo->window.ekey, which is the start time, does not change in case of
* repeat first execution, once the first execution failed.
......@@ -121,7 +122,19 @@ static void tscProcessStreamTimer(void *handle, void *tmrId) {
}
} else {
pQueryInfo->window.skey = pStream->stime - pStream->interval;
pQueryInfo->window.ekey = pStream->stime - 1;
int64_t etime = taosGetTimestamp(pStream->precision);
// delay to wait all data in last time window
if (pStream->precision == TSDB_TIME_PRECISION_MICRO) {
etime -= tsMaxStreamComputDelay * 1000l;
} else {
etime -= tsMaxStreamComputDelay;
}
if (etime > pStream->etime) {
etime = pStream->etime;
} else {
etime = pStream->stime + (etime - pStream->stime) / pStream->interval * pStream->interval;
}
pQueryInfo->window.ekey = etime;
}
// launch stream computing in a new thread
......@@ -137,7 +150,7 @@ static void tscProcessStreamQueryCallback(void *param, TAOS_RES *tres, int numOf
SSqlStream *pStream = (SSqlStream *)param;
if (tres == NULL || numOfRows < 0) {
int64_t retryDelay = tscGetRetryDelayTime(pStream->slidingTime, pStream->precision);
tscError("%p stream:%p, query data failed, code:%d, retry in %" PRId64 "ms", pStream->pSql, pStream, numOfRows,
tscError("%p stream:%p, query data failed, code:0x%08x, retry in %" PRId64 "ms", pStream->pSql, pStream, numOfRows,
retryDelay);
STableMetaInfo* pTableMetaInfo = tscGetTableMetaInfoFromCmd(&pStream->pSql->cmd, 0, 0);
......@@ -151,17 +164,45 @@ static void tscProcessStreamQueryCallback(void *param, TAOS_RES *tres, int numOf
taos_fetch_rows_a(tres, tscProcessStreamRetrieveResult, param);
}
static void tscSetTimestampForRes(SSqlStream *pStream, SSqlObj *pSql) {
SSqlRes *pRes = &pSql->res;
// no need to be called as this is alreay done in the query
static void tscStreamFillTimeGap(SSqlStream* pStream, TSKEY ts) {
#if 0
SSqlObj * pSql = pStream->pSql;
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
if (pQueryInfo->fillType != TSDB_FILL_SET_VALUE && pQueryInfo->fillType != TSDB_FILL_NULL) {
return;
}
int64_t timestamp = *(int64_t *)pRes->data;
int64_t actualTimestamp = pStream->stime - pStream->interval;
SSqlRes *pRes = &pSql->res;
/* failed to retrieve any result in this retrieve */
pSql->res.numOfRows = 1;
void *row[TSDB_MAX_COLUMNS] = {0};
char tmpRes[TSDB_MAX_BYTES_PER_ROW] = {0};
void *oldPtr = pSql->res.data;
pSql->res.data = tmpRes;
int32_t rowNum = 0;
while (pStream->stime + pStream->slidingTime < ts) {
pStream->stime += pStream->slidingTime;
*(TSKEY*)row[0] = pStream->stime;
for (int32_t i = 1; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
int16_t offset = tscFieldInfoGetOffset(pQueryInfo, i);
TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
assignVal(pSql->res.data + offset, (char *)(&pQueryInfo->fillVal[i]), pField->bytes, pField->type);
row[i] = pSql->res.data + offset;
}
(*pStream->fp)(pStream->param, pSql, row);
++rowNum;
}
if (timestamp != actualTimestamp) {
// reset the timestamp of each agg point by using start time of each interval
*((int64_t *)pRes->data) = actualTimestamp;
tscWarn("%p stream:%p, timestamp of points is:%" PRId64 ", reset to %" PRId64, pSql, pStream, timestamp, actualTimestamp);
if (rowNum > 0) {
tscDebug("%p stream:%p %d rows padded", pSql, pStream, rowNum);
}
pRes->numOfRows = 0;
pRes->data = oldPtr;
#endif
}
static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOfRows) {
......@@ -170,7 +211,7 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
if (pSql == NULL || numOfRows < 0) {
int64_t retryDelayTime = tscGetRetryDelayTime(pStream->slidingTime, pStream->precision);
tscError("%p stream:%p, retrieve data failed, code:%d, retry in %" PRId64 "ms", pSql, pStream, numOfRows, retryDelayTime);
tscError("%p stream:%p, retrieve data failed, code:0x%08x, retry in %" PRId64 "ms", pSql, pStream, numOfRows, retryDelayTime);
tscSetRetryTimer(pStream, pStream->pSql, retryDelayTime);
return;
......@@ -180,16 +221,11 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
if (numOfRows > 0) { // when reaching here the first execution of stream computing is successful.
pStream->numOfRes += numOfRows;
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
for(int32_t i = 0; i < numOfRows; ++i) {
TAOS_ROW row = taos_fetch_row(res);
tscDebug("%p stream:%p fetch result", pSql, pStream);
if (isProjectStream(pQueryInfo)) {
pStream->stime = *(TSKEY *)row[0];
} else {
tscSetTimestampForRes(pStream, pSql);
}
tscStreamFillTimeGap(pStream, *(TSKEY*)row[0]);
pStream->stime = *(TSKEY *)row[0];
// user callback function
(*pStream->fp)(pStream->param, res, row);
......@@ -199,55 +235,18 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
taos_fetch_rows_a(res, tscProcessStreamRetrieveResult, pStream);
} else { // numOfRows == 0, all data has been retrieved
pStream->useconds += pSql->res.useconds;
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
if (pStream->numOfRes == 0) {
if (pQueryInfo->fillType == TSDB_FILL_SET_VALUE || pQueryInfo->fillType == TSDB_FILL_NULL) {
SSqlRes *pRes = &pSql->res;
/* failed to retrieve any result in this retrieve */
pSql->res.numOfRows = 1;
void *row[TSDB_MAX_COLUMNS] = {0};
char tmpRes[TSDB_MAX_BYTES_PER_ROW] = {0};
void *oldPtr = pSql->res.data;
pSql->res.data = tmpRes;
for (int32_t i = 1; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
int16_t offset = tscFieldInfoGetOffset(pQueryInfo, i);
TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
assignVal(pSql->res.data + offset, (char *)(&pQueryInfo->fillVal[i]), pField->bytes, pField->type);
row[i] = pSql->res.data + offset;
}
tscSetTimestampForRes(pStream, pSql);
row[0] = pRes->data;
// char result[512] = {0};
// taos_print_row(result, row, pQueryInfo->fieldsInfo.pFields, pQueryInfo->fieldsInfo.numOfOutput);
// tscInfo("%p stream:%p query result: %s", pSql, pStream, result);
tscDebug("%p stream:%p fetch result", pSql, pStream);
// user callback function
(*pStream->fp)(pStream->param, res, row);
pRes->numOfRows = 0;
pRes->data = oldPtr;
} else if (isProjectStream(pQueryInfo)) {
if (pStream->isProject) {
/* no resuls in the query range, retry */
// todo set retry dynamic time
int32_t retry = tsProjectExecInterval;
tscError("%p stream:%p, retrieve no data, code:%d, retry in %" PRId32 "ms", pSql, pStream, numOfRows, retry);
tscError("%p stream:%p, retrieve no data, code:0x%08x, retry in %" PRId32 "ms", pSql, pStream, numOfRows, retry);
tscSetRetryTimer(pStream, pStream->pSql, retry);
return;
}
} else {
if (isProjectStream(pQueryInfo)) {
pStream->stime += 1;
}
} else if (pStream->isProject) {
pStream->stime += 1;
}
tscDebug("%p stream:%p, query on:%s, fetch result completed, fetched rows:%" PRId64, pSql, pStream, pTableMetaInfo->name,
......@@ -262,10 +261,9 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
}
static void tscSetRetryTimer(SSqlStream *pStream, SSqlObj *pSql, int64_t timer) {
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
int64_t delay = getDelayValueAfterTimewindowClosed(pStream, timer);
if (isProjectStream(pQueryInfo)) {
if (pStream->isProject) {
int64_t now = taosGetTimestamp(pStream->precision);
int64_t etime = now > pStream->etime ? pStream->etime : now;
......@@ -323,8 +321,7 @@ static int64_t getLaunchTimeDelay(const SSqlStream* pStream) {
static void tscSetNextLaunchTimer(SSqlStream *pStream, SSqlObj *pSql) {
int64_t timer = 0;
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
if (isProjectStream(pQueryInfo)) {
if (pStream->isProject) {
/*
* for project query, no mater fetch data successfully or not, next launch will issue
* more than the sliding time window
......@@ -342,7 +339,6 @@ static void tscSetNextLaunchTimer(SSqlStream *pStream, SSqlObj *pSql) {
return;
}
} else {
pStream->stime += pStream->slidingTime;
if ((pStream->stime - pStream->interval) >= pStream->etime) {
tscDebug("%p stream:%p, stime:%" PRId64 " is larger than end time: %" PRId64 ", stop the stream", pStream->pSql, pStream,
pStream->stime, pStream->etime);
......@@ -409,14 +405,16 @@ static void tscSetSlidingWindowInfo(SSqlObj *pSql, SSqlStream *pStream) {
pStream->slidingTime = pQueryInfo->slidingTime;
pQueryInfo->intervalTime = 0; // clear the interval value to avoid the force time window split by query processor
pQueryInfo->slidingTime = 0;
if (pStream->isProject) {
pQueryInfo->intervalTime = 0; // clear the interval value to avoid the force time window split by query processor
pQueryInfo->slidingTime = 0;
}
}
static int64_t tscGetStreamStartTimestamp(SSqlObj *pSql, SSqlStream *pStream, int64_t stime) {
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
if (isProjectStream(pQueryInfo)) {
if (pStream->isProject) {
// no data in table, flush all data till now to destination meter, 10sec delay
pStream->interval = tsProjectExecInterval;
pStream->slidingTime = tsProjectExecInterval;
......@@ -489,7 +487,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
SSqlStream *pStream = (SSqlStream *)calloc(1, sizeof(SSqlStream));
if (pStream == NULL) {
tscError("%p open stream failed, sql:%s, reason:%s, code:%d", pSql, sqlstr, pCmd->payload, pRes->code);
tscError("%p open stream failed, sql:%s, reason:%s, code:0x%08x", pSql, sqlstr, pCmd->payload, pRes->code);
tscFreeSqlObj(pSql);
return NULL;
}
......@@ -514,7 +512,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
if (pRes->code != TSDB_CODE_SUCCESS) {
setErrorInfo(pSql, pRes->code, pCmd->payload);
tscError("%p open stream failed, sql:%s, reason:%s, code:%d", pSql, sqlstr, pCmd->payload, pRes->code);
tscError("%p open stream failed, sql:%s, reason:%s, code:0x%08x", pSql, sqlstr, pCmd->payload, pRes->code);
tscFreeSqlObj(pSql);
return NULL;
}
......@@ -523,6 +521,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
STableComInfo tinfo = tscGetTableInfo(pTableMetaInfo->pTableMeta);
pStream->isProject = isProjectStream(pQueryInfo);
pStream->fp = fp;
pStream->callback = callback;
pStream->param = param;
......@@ -565,6 +564,8 @@ void taos_close_stream(TAOS_STREAM *handle) {
taosTmrStopA(&(pStream->pTimer));
tscDebug("%p stream:%p is closed", pSql, pStream);
// notify CQ to release the pStream object
pStream->fp(pStream->param, NULL, NULL);
tscFreeSqlObj(pSql);
pStream->pSql = NULL;
......
......@@ -1464,16 +1464,6 @@ STableMetaInfo* tscGetMetaInfo(SQueryInfo* pQueryInfo, int32_t tableIndex) {
return pQueryInfo->pTableMetaInfo[tableIndex];
}
SQueryInfo* tscGetQueryInfoDetail(SSqlCmd* pCmd, int32_t subClauseIndex) {
assert(pCmd != NULL && subClauseIndex >= 0 && subClauseIndex < TSDB_MAX_UNION_CLAUSE);
if (pCmd->pQueryInfo == NULL || subClauseIndex >= pCmd->numOfClause) {
return NULL;
}
return pCmd->pQueryInfo[subClauseIndex];
}
int32_t tscGetQueryInfoDetailSafely(SSqlCmd* pCmd, int32_t subClauseIndex, SQueryInfo** pQueryInfo) {
int32_t ret = TSDB_CODE_SUCCESS;
......@@ -2097,7 +2087,7 @@ void tscTryQueryNextClause(SSqlObj* pSql, void (*queryFp)()) {
}
void tscGetResultColumnChr(SSqlRes* pRes, SFieldInfo* pFieldInfo, int32_t columnIndex) {
SFieldSupInfo* pInfo = taosArrayGet(pFieldInfo->pSupportInfo, columnIndex);//tscFieldInfoGetSupp(pFieldInfo, columnIndex);
SFieldSupInfo* pInfo = taosArrayGet(pFieldInfo->pSupportInfo, columnIndex);
assert(pInfo->pSqlExpr != NULL);
int32_t type = pInfo->pSqlExpr->resType;
......@@ -2112,7 +2102,7 @@ void tscGetResultColumnChr(SSqlRes* pRes, SFieldInfo* pFieldInfo, int32_t column
if (isNull(pData, type)) {
pRes->tsrow[columnIndex] = NULL;
} else {
pRes->tsrow[columnIndex] = pData + VARSTR_HEADER_SIZE;
pRes->tsrow[columnIndex] = ((tstr*)pData)->data;
}
if (realLen < pInfo->pSqlExpr->resBytes - VARSTR_HEADER_SIZE) { // todo refactor
......
......@@ -129,7 +129,7 @@ int32_t tsMnodeEqualVnodeNum = 4;
int32_t tsEnableHttpModule = 1;
int32_t tsRestRowLimit = 10240;
uint16_t tsHttpPort = 6020; // only tcp, range tcp[6020]
int32_t tsHttpCacheSessions = 100;
int32_t tsHttpCacheSessions = 1000;
int32_t tsHttpSessionExpire = 36000;
int32_t tsHttpMaxThreads = 2;
int32_t tsHttpEnableCompress = 0;
......
......@@ -81,7 +81,7 @@ int64_t taosGetIntervalStartTimestamp(int64_t startTime, int64_t slidingTime, in
return startTime;
}
int64_t start = ((startTime - slidingTime) / slidingTime + 1) * slidingTime;
int64_t start = ((startTime - intervalTime) / slidingTime + 1) * slidingTime;
if (!(timeUnit == 'a' || timeUnit == 'm' || timeUnit == 's' || timeUnit == 'h')) {
/*
* here we revised the start time of day according to the local time zone,
......
#!/bin/bash
ulimit -c unlimited
function buildTDengine {
cd /root/TDengine
git remote update
REMOTE_COMMIT=`git rev-parse --short remotes/origin/develop`
LOCAL_COMMIT=`git rev-parse --short @`
echo " LOCAL: $LOCAL_COMMIT"
echo "REMOTE: $REMOTE_COMMIT"
if [ "$LOCAL_COMMIT" == "$REMOTE_COMMIT" ]; then
echo "repo up-to-date"
else
echo "repo need to pull"
git pull
LOCAL_COMMIT=`git rev-parse --short @`
cd /root/TDengine/debug
rm -rf /root/TDengine/debug/*
cmake ..
make > /dev/null
make install
fi
}
function restartTaosd {
systemctl stop taosd
pkill -KILL -x taosd
sleep 10
logDir=`grep 'logDir' /etc/taos/taos.cfg|awk 'END{print $2}'`
dataDir=`grep 'dataDir' /etc/taos/taos.cfg|awk '{print $2}'`
rm -rf $logDir/*
rm -rf $dataDir/*
taosd 2>&1 > /dev/null &
sleep 10
}
buildTDengine
restartTaosd
package com.taosdata.jdbc.utils;
import java.io.File;
import java.util.*;
import java.util.concurrent.TimeUnit;
public class TDNode {
private int index;
private int running;
private int deployed;
private boolean testCluster;
private String path;
private String cfgDir;
private String dataDir;
private String logDir;
private String cfgPath;
public TDNode(int index) {
this.index = index;
running = 0;
deployed = 0;
testCluster = false;
}
public void setPath(String path) {
this.path = path;
}
public void setTestCluster(boolean testCluster) {
this.testCluster = testCluster;
}
public void searchTaosd(File dir, ArrayList<String> taosdPath) {
File[] fileList = dir.listFiles();
if(fileList == null || fileList.length == 0) {
return;
}
for(File file : fileList) {
if(file.isFile()) {
if(file.getName().equals("taosd")) {
taosdPath.add(file.getAbsolutePath());
}
} else {
searchTaosd(file, taosdPath);
}
}
}
public void start() {
String selfPath = System.getProperty("user.dir");
String binPath = "";
String projDir = selfPath + "/../../../../";
try {
ArrayList<String> taosdPath = new ArrayList<>();
File dir = new File(projDir);
String realProjDir = dir.getCanonicalPath();
dir = new File(realProjDir);
System.out.println("project Dir: " + projDir);
searchTaosd(dir, taosdPath);
if(taosdPath.size() == 0) {
System.out.println("The project path doens't exist");
return;
} else {
for(String p : taosdPath) {
if(!p.contains("packing")) {
binPath = p;
}
}
}
} catch (Exception e) {
e.printStackTrace();
}
if(binPath.equals("")) {
System.out.println("taosd not found");
return;
} else {
System.out.println("taosd found in " + binPath);
}
if(this.deployed == 0) {
System.out.println("dnode" + index + "is not deployed");
return;
}
String cmd = "nohup " + binPath + " -c " + cfgDir + " > /dev/null 2>&1 & ";
System.out.println("start taosd cmd: " + cmd);
try{
Runtime.getRuntime().exec(cmd);
TimeUnit.SECONDS.sleep(5);
} catch (Exception e) {
e.printStackTrace();
}
this.running = 1;
}
public void stop() {
String toBeKilled = "taosd";
if (this.running != 0) {
String psCmd = "ps -ef|grep -w %s| grep -v grep | awk '{print " + toBeKilled + "}'";
try {
Process ps = Runtime.getRuntime().exec(psCmd);
ps.waitFor();
long pid = ps.pid();
String killCmd = "kill -9 " + pid;
Runtime.getRuntime().exec(killCmd).waitFor();
for(int port = 6030; port < 6041; port ++) {
String fuserCmd = "fuser -k -n tcp " + port;
Runtime.getRuntime().exec(fuserCmd).waitFor();
}
} catch (Exception e) {
e.printStackTrace();
}
this.running = 0;
System.out.println("dnode:" + this.index + "is stopped by kill -9");
}
}
public void startIP() {
try{
String cmd = "sudo ifconfig lo:" + index + "192.168.0." + index + " up";
Runtime.getRuntime().exec(cmd).waitFor();
} catch (Exception e) {
e.printStackTrace();
}
}
public void stopIP() {
try{
String cmd = "sudo ifconfig lo:" + index + "192.168.0." + index + " down";
Runtime.getRuntime().exec(cmd).waitFor();
} catch (Exception e) {
e.printStackTrace();
}
}
public void setCfgConfig(String option, String value) {
try{
String cmd = "echo " + option + " " + value + " >> " + this.cfgPath;
String[] cmdLine = {"sh", "-c", cmd};
Process ps = Runtime.getRuntime().exec(cmdLine);
ps.waitFor();
} catch (Exception e) {
e.printStackTrace();
}
}
public String getDnodeRootDir() {
String dnodeRootDir = this.path + "/sim/psim/dnode" + this.index;
return dnodeRootDir;
}
public String getDnodesRootDir() {
String dnodesRootDir = this.path + "/sim/psim" + this.index;
return dnodesRootDir;
}
public void deploy() {
this.logDir = this.path + "/sim/dnode" + this.index + "/log";
this.dataDir = this.path + "/sim/dnode" + this.index + "/data";
this.cfgDir = this.path + "/sim/dnode" + this.index + "/cfg";
this.cfgPath = this.path + "/sim/dnode" + this.index + "/cfg/taos.cfg";
try {
String cmd = "rm -rf " + this.logDir;
Runtime.getRuntime().exec(cmd).waitFor();
cmd = "rm -rf " + this.cfgDir;
Runtime.getRuntime().exec(cmd).waitFor();
cmd = "rm -rf " + this.dataDir;
Runtime.getRuntime().exec(cmd).waitFor();
cmd = "mkdir -p " + this.logDir;
Runtime.getRuntime().exec(cmd).waitFor();
cmd = "mkdir -p " + this.cfgDir;
Runtime.getRuntime().exec(cmd).waitFor();
cmd = "mkdir -p " + this.dataDir;
Runtime.getRuntime().exec(cmd).waitFor();
cmd = "touch " + this.cfgPath;
Runtime.getRuntime().exec(cmd).waitFor();
} catch (Exception e) {
e.printStackTrace();
}
if(this.testCluster) {
startIP();
setCfgConfig("masterIp", "192.168.0.1");
setCfgConfig("secondIp", "192.168.0.2");
setCfgConfig("publicIp", "192.168.0." + this.index);
setCfgConfig("internalIp", "192.168.0." + this.index);
setCfgConfig("privateIp", "192.168.0." + this.index);
}
setCfgConfig("dataDir", this.dataDir);
setCfgConfig("logDir", this.logDir);
setCfgConfig("numOfLogLines", "1000000/00");
setCfgConfig("mnodeEqualVnodeNum", "0");
setCfgConfig("walLevel", "1");
setCfgConfig("statusInterval", "1");
setCfgConfig("numOfTotalVnodes", "64");
setCfgConfig("numOfMnodes", "3");
setCfgConfig("numOfThreadsPerCore", "2.0");
setCfgConfig("monitor", "0");
setCfgConfig("maxVnodeConnections", "30000");
setCfgConfig("maxMgmtConnections", "30000");
setCfgConfig("maxMeterConnections", "30000");
setCfgConfig("maxShellConns", "30000");
setCfgConfig("locale", "en_US.UTF-8");
setCfgConfig("charset", "UTF-8");
setCfgConfig("asyncLog", "0");
setCfgConfig("anyIp", "0");
setCfgConfig("dDebugFlag", "135");
setCfgConfig("mDebugFlag", "135");
setCfgConfig("sdbDebugFlag", "135");
setCfgConfig("rpcDebugFlag", "135");
setCfgConfig("tmrDebugFlag", "131");
setCfgConfig("cDebugFlag", "135");
setCfgConfig("httpDebugFlag", "135");
setCfgConfig("monitorDebugFlag", "135");
setCfgConfig("udebugFlag", "135");
setCfgConfig("jnidebugFlag", "135");
setCfgConfig("qdebugFlag", "135");
this.deployed = 1;
}
}
\ No newline at end of file
package com.taosdata.jdbc.utils;
import java.io.File;
import java.util.*;
public class TDNodes {
private ArrayList<TDNode> tdNodes;
private boolean testCluster;
public TDNodes () {
tdNodes = new ArrayList<>();
for(int i = 1; i < 11; i ++) {
tdNodes.add(new TDNode(i));
}
}
public void setPath(String path) {
try {
String psCmd = "ps -ef|grep -w taosd| grep -v grep | awk '{print $2}'" ;
Process ps = Runtime.getRuntime().exec(psCmd);
ps.waitFor();
String killCmd = "kill -9 " + ps.pid();
Runtime.getRuntime().exec(killCmd).waitFor();
String binPath = System.getProperty("user.dir");
binPath += "/../../../debug";
System.out.println("binPath: " + binPath);
File file = new File(path);
binPath = file.getCanonicalPath();
System.out.println("binPath real path: " + binPath);
if(path.isEmpty()){
file = new File(path + "/../../");
path = file.getCanonicalPath();
}
for(int i = 0; i < tdNodes.size(); i++) {
tdNodes.get(i).setPath(path);
}
} catch (Exception e) {
e.printStackTrace();
}
}
public void setTestCluster(boolean testCluster) {
this.testCluster = testCluster;
}
public void check(int index) {
if(index < 1 || index > 10) {
System.out.println("index: " + index + " should on a scale of [1, 10]");
return;
}
}
public void deploy(int index) {
try {
File file = new File(System.getProperty("user.dir") + "/../../../");
String projectRealPath = file.getCanonicalPath();
check(index);
tdNodes.get(index - 1).setTestCluster(this.testCluster);
tdNodes.get(index - 1).setPath(projectRealPath);
tdNodes.get(index - 1).deploy();
} catch (Exception e) {
e.printStackTrace();
System.out.println("deploy Test Exception");
}
}
public void cfg(int index, String option, String value) {
check(index);
tdNodes.get(index - 1).setCfgConfig(option, value);
}
public void start(int index) {
check(index);
tdNodes.get(index - 1).start();
}
public void stop(int index) {
check(index);
tdNodes.get(index - 1).stop();
}
public void startIP(int index) {
check(index);
tdNodes.get(index - 1).startIP();
}
public void stopIP(int index) {
check(index);
tdNodes.get(index - 1).stopIP();
}
}
\ No newline at end of file
package com.taosdata.jdbc;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.File;
import com.taosdata.jdbc.utils.TDNodes;
import org.junit.AfterClass;
import org.junit.BeforeClass;
public class BaseTest {
private static boolean testCluster = false;
private static String deployPath = System.getProperty("user.dir");
private static TDNodes tdNodes = new TDNodes();
@BeforeClass
public static void setupEnv() {
try{
String path = System.getProperty("user.dir");
String bashPath = path + "/buildTDengine.sh";
File file = new File(deployPath + "/../../../");
String rootPath = file.getCanonicalPath();
tdNodes.setPath(rootPath);
tdNodes.setTestCluster(testCluster);
Process ps = Runtime.getRuntime().exec(bashPath);
ps.waitFor();
tdNodes.deploy(1);
tdNodes.start(1);
BufferedReader br = new BufferedReader(new InputStreamReader(ps.getInputStream()));
while(br.readLine() != null) {
System.out.println(br.readLine());
}
} catch (Exception e) {
e.printStackTrace();
System.out.println("Base Test Exception");
}
}
@AfterClass
public static void cleanUpEnv() {
tdNodes.stop(1);
}
}
\ No newline at end of file
......@@ -109,6 +109,8 @@ void cqClose(void *handle) {
while (pObj) {
SCqObj *pTemp = pObj;
pObj = pObj->next;
tdFreeSchema(pTemp->pSchema);
tfree(pTemp->sqlStr);
free(pTemp);
}
......@@ -242,6 +244,10 @@ static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) {
static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
SCqObj *pObj = (SCqObj *)param;
if (tres == NULL && row == NULL) {
pObj->pStream = NULL;
return;
}
SCqContext *pContext = pObj->pContext;
STSchema *pSchema = pObj->pSchema;
if (pObj->pStream == NULL) return;
......@@ -263,8 +269,14 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
void* val = row[i];
if (val == NULL) {
val = getNullValue(c->type);
} else if (IS_VAR_DATA_TYPE(c->type)) {
} else if (c->type == TSDB_DATA_TYPE_BINARY) {
val = ((char*)val) - sizeof(VarDataLenT);
} else if (c->type == TSDB_DATA_TYPE_NCHAR) {
char buf[TSDB_MAX_NCHAR_LEN];
size_t len = taos_fetch_lengths(tres)[i];
taosMbsToUcs4(val, len, buf, sizeof(buf), &len);
memcpy(val + sizeof(VarDataLenT), buf, len);
varDataLen(val) = len;
}
tdAppendColVal(trow, val, c->type, c->bytes, c->offset);
}
......
......@@ -402,6 +402,7 @@ static int32_t dnodeProcessCreateVnodeMsg(SRpcMsg *rpcMsg) {
void *pVnode = vnodeAcquireVnode(pCreate->cfg.vgId);
if (pVnode != NULL) {
dDebug("vgId:%d, already exist, processed as alter msg", pCreate->cfg.vgId);
int32_t code = vnodeAlter(pVnode, pCreate);
vnodeRelease(pVnode);
return code;
......
......@@ -87,8 +87,8 @@ int32_t qKillQuery(qinfo_t qinfo);
void* qOpenQueryMgmt(int32_t vgId);
void qSetQueryMgmtClosed(void* pExecutor);
void qCleanupQueryMgmt(void* pExecutor);
void** qRegisterQInfo(void* pMgmt, void* qInfo);
void** qAcquireQInfo(void* pMgmt, void** key);
void** qRegisterQInfo(void* pMgmt, uint64_t qInfo);
void** qAcquireQInfo(void* pMgmt, uint64_t key);
void** qReleaseQInfo(void* pMgmt, void* pQInfo, bool needFree);
#ifdef __cplusplus
......
......@@ -129,9 +129,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_CLUSTER_CFG_INCONSISTENT, 0, 0x0335, "mnode clus
TAOS_DEFINE_ERROR(TSDB_CODE_MND_ACCT_ALREADY_EXIST, 0, 0x0340, "mnode accounts already exist")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_ACCT, 0, 0x0341, "mnode invalid account")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_ACCT_PARA, 0, 0x0342, "mnode invalid account parameter")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_ACCT_OPTION, 0, 0x0343, "mnode invalid acct option")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_ACCTS, 0, 0x0344, "mnode too many accounts")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_ACCT_OPTION, 0, 0x0342, "mnode invalid acct option")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_USER_ALREADY_EXIST, 0, 0x0350, "mnode user already exist")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_USER, 0, 0x0351, "mnode invalid user")
......@@ -145,7 +143,6 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_TABLE_ID, 0, 0x0361, "mnode inva
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_TABLE_NAME, 0, 0x0362, "mnode invalid table name")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_TABLE_TYPE, 0, 0x0363, "mnode invalid table type")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_TAGS, 0, 0x0364, "mnode too many tags")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_TABLES, 0, 0x0365, "mnode too many tables")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_TIMESERIES, 0, 0x0366, "mnode not enough time series")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_NOT_SUPER_TABLE, 0, 0x0367, "mnode no super table") // operation only available for super table
TAOS_DEFINE_ERROR(TSDB_CODE_MND_COL_NAME_TOO_LONG, 0, 0x0368, "mnode column name too long")
......@@ -161,13 +158,13 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_DB_OPTION, 0, 0x0382, "mnode inva
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_DB, 0, 0x0383, "mnode invalid database")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_MONITOR_DB_FORBIDDEN, 0, 0x0384, "mnode monitor db forbidden")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_DATABASES, 0, 0x0385, "mnode too many databases")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_DB_IN_DROPPING, 0, 0x0386, "mnode db in dropping")
// dnode
TAOS_DEFINE_ERROR(TSDB_CODE_DND_MSG_NOT_PROCESSED, 0, 0x0400, "dnode message not processed")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_OUT_OF_MEMORY, 0, 0x0401, "dnode out of memory")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_NO_WRITE_ACCESS, 0, 0x0402, "dnode no disk write access")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_INVALID_MSG_LEN, 0, 0x0403, "dnode invalid message length")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_INVALID_FILE_FORMAT, 0, 0x0404, "dnode invalid file format")
// vnode
TAOS_DEFINE_ERROR(TSDB_CODE_VND_ACTION_IN_PROGRESS, 0, 0x0500, "vnode action in progress")
......
......@@ -473,7 +473,7 @@ typedef struct {
typedef struct {
int32_t code;
uint64_t qhandle;
uint64_t qhandle; // query handle
} SQueryTableRsp;
typedef struct {
......@@ -486,7 +486,7 @@ typedef struct SRetrieveTableRsp {
int32_t numOfRows;
int8_t completed; // all results are returned to client
int16_t precision;
int64_t offset; // updated offset value for multi-vnode projection query
int64_t offset; // updated offset value for multi-vnode projection query
int64_t useconds;
char data[];
} SRetrieveTableRsp;
......
......@@ -852,7 +852,6 @@ void multiThreadCreateTable(char* cols, bool use_metric, int threads, int ntable
for (int i = 0; i < threads; i++) {
info *t_info = infos + i;
taos_close(t_info->taos);
sem_destroy(&(t_info->mutex_sem));
sem_destroy(&(t_info->lock_sem));
}
......
......@@ -40,7 +40,7 @@ struct arguments {
static error_t parse_opt(int key, char *arg, struct argp_state *state) {
struct arguments *arguments = state->input;
switch (key) {
case 'w':
case 'r':
arguments->dataDir = arg;
break;
case 'd':
......
......@@ -44,6 +44,7 @@ void mnodeDecMnodeRef(struct SMnodeObj *pMnode);
char * mnodeGetMnodeRoleStr();
void mnodeGetMnodeIpSetForPeer(SRpcIpSet *ipSet);
void mnodeGetMnodeIpSetForShell(SRpcIpSet *ipSet);
char* mnodeGetMnodeMasterEp();
void mnodeGetMnodeInfos(void *mnodes);
void mnodeUpdateMnodeIpSet();
......
......@@ -128,6 +128,7 @@ int32_t mnodeInitAccts() {
void mnodeCleanupAccts() {
acctCleanUp();
sdbCloseTable(tsAcctSdb);
tsAcctSdb = NULL;
}
void *mnodeGetAcct(char *name) {
......
......@@ -459,6 +459,7 @@ void mnodeMoveVgroupToHead(SVgObj *pVgroup) {
void mnodeCleanupDbs() {
sdbCloseTable(tsDbSdb);
tsDbSdb = NULL;
}
static int32_t mnodeGetDbMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
......@@ -964,6 +965,11 @@ static int32_t mnodeProcessAlterDbMsg(SMnodeMsg *pMsg) {
mError("db:%s, failed to alter, invalid db", pAlter->db);
return TSDB_CODE_MND_INVALID_DB;
}
if (pMsg->pDb->status != TSDB_DB_STATUS_READY) {
mError("db:%s, status:%d, in dropping", pAlter->db, pMsg->pDb->status);
return TSDB_CODE_MND_DB_IN_DROPPING;
}
return mnodeAlterDb(pMsg->pDb, pAlter, pMsg);
}
......
......@@ -176,6 +176,7 @@ int32_t mnodeInitDnodes() {
void mnodeCleanupDnodes() {
sdbCloseTable(tsDnodeSdb);
tsDnodeSdb = NULL;
}
void *mnodeGetNextDnode(void *pIter, SDnodeObj **pDnode) {
......@@ -468,7 +469,7 @@ static int32_t mnodeDropDnodeByEp(char *ep, SMnodeMsg *pMsg) {
}
mnodeDecDnodeRef(pDnode);
if (strcmp(pDnode->dnodeEp, dnodeGetMnodeMasterEp()) == 0) {
if (strcmp(pDnode->dnodeEp, mnodeGetMnodeMasterEp()) == 0) {
mError("dnode:%d, can't drop dnode:%s which is master", pDnode->dnodeId, ep);
return TSDB_CODE_MND_NO_REMOVE_MASTER;
}
......
......@@ -121,9 +121,9 @@ void mnodeCleanupSystem() {
dnodeFreeMnodeWqueue();
dnodeFreeMnodeRqueue();
dnodeFreeMnodePqueue();
mnodeCleanupComponents(sizeof(tsMnodeComponents) / sizeof(tsMnodeComponents[0]) - 1);
mnodeCleanupTimer();
mnodeCleanupComponents(sizeof(tsMnodeComponents) / sizeof(tsMnodeComponents[0]) - 1);
mInfo("mnode is cleaned up");
}
......
......@@ -165,6 +165,7 @@ int32_t mnodeInitMnodes() {
void mnodeCleanupMnodes() {
sdbCloseTable(tsMnodeSdb);
tsMnodeSdb = NULL;
mnodeMnodeDestroyLock();
}
......@@ -267,6 +268,10 @@ void mnodeGetMnodeIpSetForShell(SRpcIpSet *ipSet) {
mnodeMnodeUnLock();
}
char* mnodeGetMnodeMasterEp() {
return tsMnodeInfos.nodeInfos[tsMnodeInfos.inUse].nodeEp;
}
void mnodeGetMnodeInfos(void *mnodeInfos) {
mnodeMnodeRdLock();
*(SDMMnodeInfos *)mnodeInfos = tsMnodeInfos;
......
......@@ -199,7 +199,7 @@ static void sdbRestoreTables() {
sdbDebug("table:%s, is restored, numOfRows:%" PRId64, pTable->tableName, pTable->numOfRows);
}
sdbInfo("sdb is restored, version:%" PRId64 " totalRows:%d numOfTables:%d", tsSdbObj.version, totalRows, numOfTables);
sdbInfo("sdb is restored, ver:%" PRId64 " totalRows:%d numOfTables:%d", tsSdbObj.version, totalRows, numOfTables);
}
void sdbUpdateMnodeRoles() {
......@@ -373,7 +373,7 @@ void sdbCleanUp() {
tsSdbObj.status = SDB_STATUS_CLOSING;
sdbCleanupWriteWorker();
sdbDebug("sdb will be closed, version:%" PRId64, tsSdbObj.version);
sdbDebug("sdb will be closed, ver:%" PRId64, tsSdbObj.version);
if (tsSdbObj.sync) {
syncStop(tsSdbObj.sync);
......@@ -471,8 +471,8 @@ static int32_t sdbInsertHash(SSdbTable *pTable, SSdbOper *pOper) {
atomic_add_fetch_32(&pTable->autoIndex, 1);
}
sdbDebug("table:%s, insert record:%s to hash, rowSize:%d numOfRows:%" PRId64 " version:%" PRIu64, pTable->tableName,
sdbGetKeyStrFromObj(pTable, pOper->pObj), pOper->rowSize, pTable->numOfRows, sdbGetVersion());
sdbDebug("table:%s, insert record:%s to hash, rowSize:%d numOfRows:%" PRId64 ", msg:%p", pTable->tableName,
sdbGetKeyStrFromObj(pTable, pOper->pObj), pOper->rowSize, pTable->numOfRows, pOper->pMsg);
(*pTable->insertFp)(pOper);
return TSDB_CODE_SUCCESS;
......@@ -490,8 +490,8 @@ static int32_t sdbDeleteHash(SSdbTable *pTable, SSdbOper *pOper) {
taosHashRemove(pTable->iHandle, key, keySize);
atomic_sub_fetch_32(&pTable->numOfRows, 1);
sdbDebug("table:%s, delete record:%s from hash, numOfRows:%" PRId64 " version:%" PRIu64, pTable->tableName,
sdbGetKeyStrFromObj(pTable, pOper->pObj), pTable->numOfRows, sdbGetVersion());
sdbDebug("table:%s, delete record:%s from hash, numOfRows:%" PRId64 ", msg:%p", pTable->tableName,
sdbGetKeyStrFromObj(pTable, pOper->pObj), pTable->numOfRows, pOper->pMsg);
int8_t *updateEnd = pOper->pObj + pTable->refCountPos - 1;
*updateEnd = 1;
......@@ -501,8 +501,8 @@ static int32_t sdbDeleteHash(SSdbTable *pTable, SSdbOper *pOper) {
}
static int32_t sdbUpdateHash(SSdbTable *pTable, SSdbOper *pOper) {
sdbDebug("table:%s, update record:%s in hash, numOfRows:%" PRId64 " version:%" PRIu64, pTable->tableName,
sdbGetKeyStrFromObj(pTable, pOper->pObj), pTable->numOfRows, sdbGetVersion());
sdbDebug("table:%s, update record:%s in hash, numOfRows:%" PRId64 ", msg:%p", pTable->tableName,
sdbGetKeyStrFromObj(pTable, pOper->pObj), pTable->numOfRows, pOper->pMsg);
(*pTable->updateFp)(pOper);
return TSDB_CODE_SUCCESS;
......@@ -527,12 +527,12 @@ static int sdbWrite(void *param, void *data, int type) {
// for data from WAL or forward, version may be smaller
if (pHead->version <= tsSdbObj.version) {
pthread_mutex_unlock(&tsSdbObj.mutex);
sdbDebug("table:%s, failed to restore %s record:%s from source(%d), version:%" PRId64 " too large, sdb version:%" PRId64,
sdbDebug("table:%s, failed to restore %s record:%s from source(%d), ver:%" PRId64 " too large, sdb ver:%" PRId64,
pTable->tableName, sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), type, pHead->version, tsSdbObj.version);
return TSDB_CODE_SUCCESS;
} else if (pHead->version != tsSdbObj.version + 1) {
pthread_mutex_unlock(&tsSdbObj.mutex);
sdbError("table:%s, failed to restore %s record:%s from source(%d), version:%" PRId64 " too large, sdb version:%" PRId64,
sdbError("table:%s, failed to restore %s record:%s from source(%d), ver:%" PRId64 " too large, sdb ver:%" PRId64,
pTable->tableName, sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), type, pHead->version, tsSdbObj.version);
return TSDB_CODE_MND_APP_ERROR;
} else {
......@@ -556,19 +556,19 @@ static int sdbWrite(void *param, void *data, int type) {
if (syncCode <= 0) pOper->processedCount = 1;
if (syncCode < 0) {
sdbError("table:%s, failed to forward request, result:%s action:%s record:%s version:%" PRId64, pTable->tableName,
tstrerror(syncCode), sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), pHead->version);
sdbError("table:%s, failed to forward request, result:%s action:%s record:%s ver:%" PRId64 ", msg:%p", pTable->tableName,
tstrerror(syncCode), sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), pHead->version, pOper->pMsg);
} else if (syncCode > 0) {
sdbDebug("table:%s, forward request is sent, action:%s record:%s version:%" PRId64, pTable->tableName,
sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), pHead->version);
sdbDebug("table:%s, forward request is sent, action:%s record:%s ver:%" PRId64 ", msg:%p", pTable->tableName,
sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), pHead->version, pOper->pMsg);
} else {
sdbTrace("table:%s, no need to send fwd request, action:%s record:%s version:%" PRId64, pTable->tableName,
sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), pHead->version);
sdbTrace("table:%s, no need to send fwd request, action:%s record:%s ver:%" PRId64 ", msg:%p", pTable->tableName,
sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), pHead->version, pOper->pMsg);
}
return syncCode;
}
sdbDebug("table:%s, record from wal/fwd is disposed, action:%s record:%s version:%" PRId64, pTable->tableName,
sdbDebug("table:%s, record from wal/fwd is disposed, action:%s record:%s ver:%" PRId64, pTable->tableName,
sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), pHead->version);
// even it is WAL/FWD, it shall be called to update version in sync
......@@ -957,7 +957,7 @@ static void *sdbWorkerFp(void *param) {
pOper->processedCount = 1;
pHead = (void *)pOper + sizeof(SSdbOper) + SDB_SYNC_HACK;
if (pOper->pMsg != NULL) {
sdbDebug("app:%p:%p, table:%s record:%p:%s version:%" PRIu64 ", will be processed in sdb queue",
sdbDebug("app:%p:%p, table:%s record:%p:%s ver:%" PRIu64 ", will be processed in sdb queue",
pOper->pMsg->rpcMsg.ahandle, pOper->pMsg, ((SSdbTable *)pOper->table)->tableName, pOper->pObj,
sdbGetKeyStr(pOper->table, pHead->cont), pHead->version);
}
......@@ -967,7 +967,11 @@ static void *sdbWorkerFp(void *param) {
}
int32_t code = sdbWrite(pOper, pHead, type);
if (pOper && code <= 0) pOper->retCode = code;
if (code > 0) code = 0;
if (pOper)
pOper->retCode = code;
else
pHead->len = code; // hackway
}
walFsync(tsSdbObj.wal);
......@@ -982,7 +986,8 @@ static void *sdbWorkerFp(void *param) {
sdbDecRef(pOper->table, pOper->pObj);
sdbConfirmForward(NULL, pOper, pOper->retCode);
} else if (type == TAOS_QTYPE_FWD) {
syncConfirmForward(tsSdbObj.sync, pHead->version, TSDB_CODE_SUCCESS);
pHead = (SWalHead *)item;
syncConfirmForward(tsSdbObj.sync, pHead->version, pHead->len);
taosFreeQitem(item);
} else {
taosFreeQitem(item);
......
......@@ -307,6 +307,12 @@ static int32_t mnodeProcessConnectMsg(SMnodeMsg *pMsg) {
code = TSDB_CODE_MND_INVALID_DB;
goto connect_over;
}
if (pDb->status != TSDB_DB_STATUS_READY) {
mError("db:%s, status:%d, in dropping", pDb->name, pDb->status);
code = TSDB_CODE_MND_DB_IN_DROPPING;
goto connect_over;
}
mnodeDecDbRef(pDb);
}
......@@ -350,7 +356,12 @@ static int32_t mnodeProcessUseMsg(SMnodeMsg *pMsg) {
int32_t code = TSDB_CODE_SUCCESS;
if (pMsg->pDb == NULL) pMsg->pDb = mnodeGetDb(pUseDbMsg->db);
if (pMsg->pDb == NULL) {
code = TSDB_CODE_MND_INVALID_DB;
return TSDB_CODE_MND_INVALID_DB;
}
if (pMsg->pDb->status != TSDB_DB_STATUS_READY) {
mError("db:%s, status:%d, in dropping", pMsg->pDb->name, pMsg->pDb->status);
return TSDB_CODE_MND_DB_IN_DROPPING;
}
return code;
......@@ -403,4 +414,4 @@ void mnodeVacuumResult(char *data, int32_t numOfCols, int32_t rows, int32_t capa
memmove(data + pShow->offset[i] * rows, data + pShow->offset[i] * capacity, pShow->bytes[i] * rows);
}
}
}
\ No newline at end of file
}
......@@ -116,6 +116,11 @@ static int32_t mnodeChildTableActionInsert(SSdbOper *pOper) {
mError("ctable:%s, vgId:%d not in db:%s", pTable->info.tableId, pVgroup->vgId, pVgroup->dbName);
return TSDB_CODE_MND_INVALID_DB;
}
if (pDb->status != TSDB_DB_STATUS_READY) {
mError("db:%s, status:%d, in dropping", pDb->name, pDb->status);
return TSDB_CODE_MND_DB_IN_DROPPING;
}
mnodeDecDbRef(pDb);
SAcctObj *pAcct = mnodeGetAcct(pDb->acct);
......@@ -284,8 +289,8 @@ static int32_t mnodeChildTableActionRestored() {
if (pTable == NULL) break;
SDbObj *pDb = mnodeGetDbByTableId(pTable->info.tableId);
if (pDb == NULL) {
mError("ctable:%s, failed to get db, discard it", pTable->info.tableId);
if (pDb == NULL || pDb->status != TSDB_DB_STATUS_READY) {
mError("ctable:%s, failed to get db or db in dropping, discard it", pTable->info.tableId);
SSdbOper desc = {.type = SDB_OPER_LOCAL, .pObj = pTable, .table = tsChildTableSdb};
sdbDeleteRow(&desc);
mnodeDecTableRef(pTable);
......@@ -376,6 +381,7 @@ static int32_t mnodeInitChildTables() {
static void mnodeCleanupChildTables() {
sdbCloseTable(tsChildTableSdb);
tsChildTableSdb = NULL;
}
static void mnodeAddTableIntoStable(SSuperTableObj *pStable, SChildTableObj *pCtable) {
......@@ -422,7 +428,7 @@ static int32_t mnodeSuperTableActionDestroy(SSdbOper *pOper) {
static int32_t mnodeSuperTableActionInsert(SSdbOper *pOper) {
SSuperTableObj *pStable = pOper->pObj;
SDbObj *pDb = mnodeGetDbByTableId(pStable->info.tableId);
if (pDb != NULL) {
if (pDb != NULL && pDb->status == TSDB_DB_STATUS_READY) {
mnodeAddSuperTableIntoDb(pDb);
}
mnodeDecDbRef(pDb);
......@@ -554,6 +560,7 @@ static int32_t mnodeInitSuperTables() {
static void mnodeCleanupSuperTables() {
sdbCloseTable(tsSuperTableSdb);
tsSuperTableSdb = NULL;
}
int32_t mnodeInitTables() {
......@@ -683,10 +690,15 @@ static int32_t mnodeProcessCreateTableMsg(SMnodeMsg *pMsg) {
SCMCreateTableMsg *pCreate = pMsg->rpcMsg.pCont;
if (pMsg->pDb == NULL) pMsg->pDb = mnodeGetDb(pCreate->db);
if (pMsg->pDb == NULL || pMsg->pDb->status != TSDB_DB_STATUS_READY) {
if (pMsg->pDb == NULL) {
mError("app:%p:%p, table:%s, failed to create, db not selected", pMsg->rpcMsg.ahandle, pMsg, pCreate->tableId);
return TSDB_CODE_MND_DB_NOT_SELECTED;
}
if (pMsg->pDb->status != TSDB_DB_STATUS_READY) {
mError("db:%s, status:%d, in dropping", pMsg->pDb->name, pMsg->pDb->status);
return TSDB_CODE_MND_DB_IN_DROPPING;
}
if (pMsg->pTable == NULL) pMsg->pTable = mnodeGetTable(pCreate->tableId);
if (pMsg->pTable != NULL && pMsg->retry == 0) {
......@@ -717,10 +729,15 @@ static int32_t mnodeProcessCreateTableMsg(SMnodeMsg *pMsg) {
static int32_t mnodeProcessDropTableMsg(SMnodeMsg *pMsg) {
SCMDropTableMsg *pDrop = pMsg->rpcMsg.pCont;
if (pMsg->pDb == NULL) pMsg->pDb = mnodeGetDbByTableId(pDrop->tableId);
if (pMsg->pDb == NULL || pMsg->pDb->status != TSDB_DB_STATUS_READY) {
mError("app:%p:%p, table:%s, failed to drop table, db not selected", pMsg->rpcMsg.ahandle, pMsg, pDrop->tableId);
if (pMsg->pDb == NULL) {
mError("app:%p:%p, table:%s, failed to drop table, db not selected or db in dropping", pMsg->rpcMsg.ahandle, pMsg, pDrop->tableId);
return TSDB_CODE_MND_DB_NOT_SELECTED;
}
if (pMsg->pDb->status != TSDB_DB_STATUS_READY) {
mError("db:%s, status:%d, in dropping", pMsg->pDb->name, pMsg->pDb->status);
return TSDB_CODE_MND_DB_IN_DROPPING;
}
if (mnodeCheckIsMonitorDB(pMsg->pDb->name, tsMonitorDbName)) {
mError("app:%p:%p, table:%s, failed to drop table, in monitor database", pMsg->rpcMsg.ahandle, pMsg,
......@@ -755,11 +772,16 @@ static int32_t mnodeProcessTableMetaMsg(SMnodeMsg *pMsg) {
pInfo->tableId, pMsg->rpcMsg.handle, pInfo->createFlag);
if (pMsg->pDb == NULL) pMsg->pDb = mnodeGetDbByTableId(pInfo->tableId);
if (pMsg->pDb == NULL || pMsg->pDb->status != TSDB_DB_STATUS_READY) {
if (pMsg->pDb == NULL) {
mError("app:%p:%p, table:%s, failed to get table meta, db not selected", pMsg->rpcMsg.ahandle, pMsg,
pInfo->tableId);
return TSDB_CODE_MND_DB_NOT_SELECTED;
}
if (pMsg->pDb->status != TSDB_DB_STATUS_READY) {
mError("db:%s, status:%d, in dropping", pMsg->pDb->name, pMsg->pDb->status);
return TSDB_CODE_MND_DB_IN_DROPPING;
}
if (pMsg->pTable == NULL) pMsg->pTable = mnodeGetTable(pInfo->tableId);
if (pMsg->pTable == NULL) {
......@@ -1207,6 +1229,11 @@ static int32_t mnodeDropSuperTableColumn(SMnodeMsg *pMsg, char *colName) {
static int32_t mnodeGetShowSuperTableMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
SDbObj *pDb = mnodeGetDb(pShow->db);
if (pDb == NULL) return TSDB_CODE_MND_DB_NOT_SELECTED;
if (pDb->status != TSDB_DB_STATUS_READY) {
mError("db:%s, status:%d, in dropping", pDb->name, pDb->status);
return TSDB_CODE_MND_DB_IN_DROPPING;
}
int32_t cols = 0;
SSchema *pSchema = pMeta->schema;
......@@ -1266,6 +1293,11 @@ int32_t mnodeRetrieveShowSuperTables(SShowObj *pShow, char *data, int32_t rows,
SDbObj *pDb = mnodeGetDb(pShow->db);
if (pDb == NULL) return 0;
if (pDb->status != TSDB_DB_STATUS_READY) {
mError("db:%s, status:%d, in dropping", pDb->name, pDb->status);
return 0;
}
tstrncpy(prefix, pDb->name, 64);
strcat(prefix, TS_PATH_DELIMITER);
......@@ -2297,7 +2329,7 @@ static int32_t mnodeProcessMultiTableMetaMsg(SMnodeMsg *pMsg) {
if (pTable == NULL) continue;
if (pMsg->pDb == NULL) pMsg->pDb = mnodeGetDbByTableId(tableId);
if (pMsg->pDb == NULL) {
if (pMsg->pDb == NULL || pMsg->pDb->status != TSDB_DB_STATUS_READY) {
mnodeDecTableRef(pTable);
continue;
}
......@@ -2335,6 +2367,11 @@ static int32_t mnodeProcessMultiTableMetaMsg(SMnodeMsg *pMsg) {
static int32_t mnodeGetShowTableMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
SDbObj *pDb = mnodeGetDb(pShow->db);
if (pDb == NULL) return TSDB_CODE_MND_DB_NOT_SELECTED;
if (pDb->status != TSDB_DB_STATUS_READY) {
mError("db:%s, status:%d, in dropping", pDb->name, pDb->status);
return TSDB_CODE_MND_DB_IN_DROPPING;
}
int32_t cols = 0;
SSchema *pSchema = pMeta->schema;
......@@ -2383,6 +2420,11 @@ static int32_t mnodeGetShowTableMeta(STableMetaMsg *pMeta, SShowObj *pShow, void
static int32_t mnodeRetrieveShowTables(SShowObj *pShow, char *data, int32_t rows, void *pConn) {
SDbObj *pDb = mnodeGetDb(pShow->db);
if (pDb == NULL) return 0;
if (pDb->status != TSDB_DB_STATUS_READY) {
mError("db:%s, status:%d, in dropping", pDb->name, pDb->status);
return 0;
}
int32_t numOfRows = 0;
SChildTableObj *pTable = NULL;
......@@ -2474,10 +2516,15 @@ static int32_t mnodeProcessAlterTableMsg(SMnodeMsg *pMsg) {
pAlter->tableId, pMsg->rpcMsg.handle);
if (pMsg->pDb == NULL) pMsg->pDb = mnodeGetDbByTableId(pAlter->tableId);
if (pMsg->pDb == NULL || pMsg->pDb->status != TSDB_DB_STATUS_READY) {
if (pMsg->pDb == NULL) {
mError("app:%p:%p, table:%s, failed to alter table, db not selected", pMsg->rpcMsg.ahandle, pMsg, pAlter->tableId);
return TSDB_CODE_MND_DB_NOT_SELECTED;
}
if (pMsg->pDb->status != TSDB_DB_STATUS_READY) {
mError("db:%s, status:%d, in dropping", pMsg->pDb->name, pMsg->pDb->status);
return TSDB_CODE_MND_DB_IN_DROPPING;
}
if (mnodeCheckIsMonitorDB(pMsg->pDb->name, tsMonitorDbName)) {
mError("app:%p:%p, table:%s, failed to alter table, its log db", pMsg->rpcMsg.ahandle, pMsg, pAlter->tableId);
......@@ -2537,6 +2584,11 @@ static int32_t mnodeProcessAlterTableMsg(SMnodeMsg *pMsg) {
static int32_t mnodeGetStreamTableMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
SDbObj *pDb = mnodeGetDb(pShow->db);
if (pDb == NULL) return TSDB_CODE_MND_DB_NOT_SELECTED;
if (pDb->status != TSDB_DB_STATUS_READY) {
mError("db:%s, status:%d, in dropping", pDb->name, pDb->status);
return TSDB_CODE_MND_DB_IN_DROPPING;
}
int32_t cols = 0;
SSchema *pSchema = pMeta->schema;
......@@ -2584,7 +2636,11 @@ static int32_t mnodeGetStreamTableMeta(STableMetaMsg *pMeta, SShowObj *pShow, vo
static int32_t mnodeRetrieveStreamTables(SShowObj *pShow, char *data, int32_t rows, void *pConn) {
SDbObj *pDb = mnodeGetDb(pShow->db);
if (pDb == NULL) return 0;
if (pDb->status != TSDB_DB_STATUS_READY) {
mError("db:%s, status:%d, in dropping", pDb->name, pDb->status);
return 0;
}
int32_t numOfRows = 0;
SChildTableObj *pTable = NULL;
......
......@@ -154,6 +154,7 @@ int32_t mnodeInitUsers() {
void mnodeCleanupUsers() {
sdbCloseTable(tsUserSdb);
tsUserSdb = NULL;
}
SUserObj *mnodeGetUser(char *name) {
......
......@@ -75,6 +75,11 @@ static int32_t mnodeVgroupActionInsert(SSdbOper *pOper) {
if (pDb == NULL) {
return TSDB_CODE_MND_INVALID_DB;
}
if (pDb->status != TSDB_DB_STATUS_READY) {
mError("db:%s, status:%d, in dropping", pDb->name, pDb->status);
return TSDB_CODE_MND_DB_IN_DROPPING;
}
pVgroup->pDb = pDb;
pVgroup->prev = NULL;
......@@ -367,7 +372,6 @@ static int32_t mnodeCreateVgroupCb(SMnodeMsg *pMsg, int32_t code) {
pVgroup->vnodeGid[i].dnodeId);
}
mnodeIncVgroupRef(pVgroup);
pMsg->expected = pVgroup->numOfVnodes;
mnodeSendCreateVgroupMsg(pVgroup, pMsg);
......@@ -388,6 +392,9 @@ int32_t mnodeCreateVgroup(SMnodeMsg *pMsg, SDbObj *pDb) {
return TSDB_CODE_MND_NO_ENOUGH_DNODES;
}
pMsg->pVgroup = pVgroup;
mnodeIncVgroupRef(pVgroup);
SSdbOper oper = {
.type = SDB_OPER_GLOBAL,
.table = tsVgroupSdb,
......@@ -397,8 +404,6 @@ int32_t mnodeCreateVgroup(SMnodeMsg *pMsg, SDbObj *pDb) {
.cb = mnodeCreateVgroupCb
};
pMsg->pVgroup = pVgroup;
int32_t code = sdbInsertRow(&oper);
if (code != TSDB_CODE_SUCCESS) {
pMsg->pVgroup = NULL;
......@@ -427,6 +432,7 @@ void mnodeDropVgroup(SVgObj *pVgroup, void *ahandle) {
void mnodeCleanupVgroups() {
sdbCloseTable(tsVgroupSdb);
tsVgroupSdb = NULL;
}
int32_t mnodeGetVgroupMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
......@@ -434,6 +440,11 @@ int32_t mnodeGetVgroupMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
if (pDb == NULL) {
return TSDB_CODE_MND_DB_NOT_SELECTED;
}
if (pDb->status != TSDB_DB_STATUS_READY) {
mError("db:%s, status:%d, in dropping", pDb->name, pDb->status);
return TSDB_CODE_MND_DB_IN_DROPPING;
}
int32_t cols = 0;
SSchema *pSchema = pMeta->schema;
......@@ -522,6 +533,11 @@ int32_t mnodeRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, void *pC
SDbObj *pDb = mnodeGetDb(pShow->db);
if (pDb == NULL) return 0;
if (pDb->status != TSDB_DB_STATUS_READY) {
mError("db:%s, status:%d, in dropping", pDb->name, pDb->status);
return 0;
}
pVgroup = pDb->pHead;
while (pVgroup != NULL) {
......@@ -696,9 +712,9 @@ static void mnodeProcessCreateVnodeRsp(SRpcMsg *rpcMsg) {
if (rpcMsg->ahandle == NULL) return;
SMnodeMsg *mnodeMsg = rpcMsg->ahandle;
mnodeMsg->received++;
atomic_add_fetch_8(&mnodeMsg->received, 1);
if (rpcMsg->code == TSDB_CODE_SUCCESS) {
mnodeMsg->successed++;
atomic_add_fetch_8(&mnodeMsg->successed, 1);
} else {
mnodeMsg->code = rpcMsg->code;
}
......@@ -798,19 +814,20 @@ static int32_t mnodeProcessVnodeCfgMsg(SMnodeMsg *pMsg) {
mDebug("dnode:%s, vgId:%d, invalid dnode", taosIpStr(pCfg->dnodeId), pCfg->vgId);
return TSDB_CODE_MND_VGROUP_NOT_EXIST;
}
mnodeDecDnodeRef(pDnode);
SVgObj *pVgroup = mnodeGetVgroup(pCfg->vgId);
if (pVgroup == NULL) {
mDebug("dnode:%s, vgId:%d, no vgroup info", taosIpStr(pCfg->dnodeId), pCfg->vgId);
mnodeDecDnodeRef(pDnode);
return TSDB_CODE_MND_VGROUP_NOT_EXIST;
}
mnodeDecVgroupRef(pVgroup);
mDebug("vgId:%d, send create vnode msg to dnode %s for vnode cfg msg", pVgroup->vgId, pDnode->dnodeEp);
SRpcIpSet ipSet = mnodeGetIpSetFromIp(pDnode->dnodeEp);
mnodeSendCreateVnodeMsg(pVgroup, &ipSet, NULL);
mnodeDecDnodeRef(pDnode);
mnodeDecVgroupRef(pVgroup);
return TSDB_CODE_SUCCESS;
}
......
......@@ -206,7 +206,7 @@ typedef struct HttpThread {
pthread_mutex_t threadMutex;
bool stop;
int pollFd;
int numOfFds;
int numOfContexts;
int threadId;
char label[HTTP_LABEL_SIZE];
bool (*processData)(HttpContext *pContext);
......
......@@ -44,7 +44,7 @@ static void httpDestroyContext(void *data) {
HttpThread *pThread = pContext->pThread;
httpRemoveContextFromEpoll(pContext);
httpReleaseSession(pContext);
atomic_sub_fetch_32(&pThread->numOfFds, 1);
atomic_sub_fetch_32(&pThread->numOfContexts, 1);
pContext->pThread = 0;
pContext->state = HTTP_CONTEXT_STATE_CLOSED;
......@@ -171,38 +171,39 @@ bool httpInitContext(HttpContext *pContext) {
void httpCloseContextByApp(HttpContext *pContext) {
pContext->parsed = false;
bool keepAlive = true;
if (pContext->httpVersion == HTTP_VERSION_10 && pContext->httpKeepAlive != HTTP_KEEPALIVE_ENABLE) {
keepAlive = false;
} else if (pContext->httpVersion != HTTP_VERSION_10 && pContext->httpKeepAlive == HTTP_KEEPALIVE_DISABLE) {
keepAlive = false;
} else {}
} else {
}
if (keepAlive) {
if (httpAlterContextState(pContext, HTTP_CONTEXT_STATE_HANDLING, HTTP_CONTEXT_STATE_READY)) {
httpDebug("context:%p, fd:%d, ip:%s, last state:handling, keepAlive:true, reuse connect",
pContext, pContext->fd, pContext->ipstr);
httpDebug("context:%p, fd:%d, ip:%s, last state:handling, keepAlive:true, reuse context", pContext, pContext->fd,
pContext->ipstr);
} else if (httpAlterContextState(pContext, HTTP_CONTEXT_STATE_DROPPING, HTTP_CONTEXT_STATE_CLOSED)) {
httpRemoveContextFromEpoll(pContext);
httpDebug("context:%p, fd:%d, ip:%s, last state:dropping, keepAlive:true, close connect",
pContext, pContext->fd, pContext->ipstr);
httpDebug("context:%p, fd:%d, ip:%s, last state:dropping, keepAlive:true, close connect", pContext, pContext->fd,
pContext->ipstr);
} else if (httpAlterContextState(pContext, HTTP_CONTEXT_STATE_READY, HTTP_CONTEXT_STATE_READY)) {
httpDebug("context:%p, fd:%d, ip:%s, last state:ready, keepAlive:true, reuse connect",
pContext, pContext->fd, pContext->ipstr);
httpDebug("context:%p, fd:%d, ip:%s, last state:ready, keepAlive:true, reuse context", pContext, pContext->fd,
pContext->ipstr);
} else if (httpAlterContextState(pContext, HTTP_CONTEXT_STATE_CLOSED, HTTP_CONTEXT_STATE_CLOSED)) {
httpRemoveContextFromEpoll(pContext);
httpDebug("context:%p, fd:%d, ip:%s, last state:ready, keepAlive:true, close connect",
pContext, pContext->fd, pContext->ipstr);
httpDebug("context:%p, fd:%d, ip:%s, last state:ready, keepAlive:true, close connect", pContext, pContext->fd,
pContext->ipstr);
} else {
httpRemoveContextFromEpoll(pContext);
httpError("context:%p, fd:%d, ip:%s, last state:%s:%d, keepAlive:true, close connect",
pContext, pContext->fd, pContext->ipstr, httpContextStateStr(pContext->state), pContext->state);
httpError("context:%p, fd:%d, ip:%s, last state:%s:%d, keepAlive:true, close connect", pContext, pContext->fd,
pContext->ipstr, httpContextStateStr(pContext->state), pContext->state);
}
} else {
httpRemoveContextFromEpoll(pContext);
httpDebug("context:%p, fd:%d, ip:%s, last state:%s:%d, keepAlive:false, close connect",
pContext, pContext->fd, pContext->ipstr, httpContextStateStr(pContext->state), pContext->state);
httpDebug("context:%p, fd:%d, ip:%s, last state:%s:%d, keepAlive:false, close context", pContext, pContext->fd,
pContext->ipstr, httpContextStateStr(pContext->state), pContext->state);
}
httpReleaseContext(pContext);
......@@ -214,7 +215,7 @@ void httpCloseContextByServer(HttpContext *pContext) {
} else if (httpAlterContextState(pContext, HTTP_CONTEXT_STATE_DROPPING, HTTP_CONTEXT_STATE_DROPPING)) {
httpDebug("context:%p, fd:%d, ip:%s, epoll already finished, wait app finished", pContext, pContext->fd, pContext->ipstr);
} else if (httpAlterContextState(pContext, HTTP_CONTEXT_STATE_READY, HTTP_CONTEXT_STATE_CLOSED)) {
httpDebug("context:%p, fd:%d, ip:%s, epoll finished, close context", pContext, pContext->fd, pContext->ipstr);
httpDebug("context:%p, fd:%d, ip:%s, epoll finished, close connect", pContext, pContext->fd, pContext->ipstr);
} else if (httpAlterContextState(pContext, HTTP_CONTEXT_STATE_CLOSED, HTTP_CONTEXT_STATE_CLOSED)) {
httpDebug("context:%p, fd:%d, ip:%s, epoll finished, will be closed soon", pContext, pContext->fd, pContext->ipstr);
} else {
......
......@@ -313,9 +313,9 @@ bool httpParseRequest(HttpContext* pContext) {
return true;
}
httpTraceL("context:%p, fd:%d, ip:%s, thread:%s, numOfFds:%d, read size:%d, raw data:\n%s", pContext, pContext->fd,
pContext->ipstr, pContext->pThread->label, pContext->pThread->numOfFds, pContext->parser.bufsize,
pContext->parser.buffer);
httpTraceL("context:%p, fd:%d, ip:%s, thread:%s, numOfContexts:%d, read size:%d, raw data:\n%s", pContext,
pContext->fd, pContext->ipstr, pContext->pThread->label, pContext->pThread->numOfContexts,
pContext->parser.bufsize, pContext->parser.buffer);
if (!httpGetHttpMethod(pContext)) {
return false;
......
......@@ -293,7 +293,7 @@ static void *httpAcceptHttpConnection(void *arg) {
totalFds = 1;
for (int i = 0; i < pServer->numOfThreads; ++i) {
totalFds += pServer->pThreads[i].numOfFds;
totalFds += pServer->pThreads[i].numOfContexts;
}
if (totalFds > tsHttpCacheSessions * 100) {
......@@ -332,9 +332,9 @@ static void *httpAcceptHttpConnection(void *arg) {
}
// notify the data process, add into the FdObj list
atomic_add_fetch_32(&pThread->numOfFds, 1);
httpDebug("context:%p, fd:%d, ip:%s, thread:%s numOfFds:%d totalFds:%d, accept a new connection", pContext, connFd,
pContext->ipstr, pThread->label, pThread->numOfFds, totalFds);
atomic_add_fetch_32(&pThread->numOfContexts, 1);
httpDebug("context:%p, fd:%d, ip:%s, thread:%s numOfContexts:%d totalFds:%d, accept a new connection", pContext,
connFd, pContext->ipstr, pThread->label, pThread->numOfContexts, totalFds);
// pick up next thread for next connection
threadId++;
......
......@@ -172,6 +172,7 @@ typedef struct SQueryRuntimeEnv {
bool topBotQuery; // false
bool groupbyNormalCol; // denote if this is a groupby normal column query
bool hasTagResults; // if there are tag values in final result or not
int32_t interBufSize; // intermediate buffer sizse
int32_t prevGroupId; // previous executed group id
SDiskbasedResultBuf* pResultBuf; // query result buffer based on blocked-wised disk file
} SQueryRuntimeEnv;
......
......@@ -15,6 +15,8 @@
#ifndef TDENGINE_QUERYUTIL_H
#define TDENGINE_QUERYUTIL_H
int32_t getOutputInterResultBufSize(SQuery* pQuery);
void clearTimeWindowResBuf(SQueryRuntimeEnv* pRuntimeEnv, SWindowResult* pOneOutputRes);
void copyTimeWindowResBuf(SQueryRuntimeEnv* pRuntimeEnv, SWindowResult* dst, const SWindowResult* src);
......@@ -35,7 +37,7 @@ SWindowResult *getWindowResult(SWindowResInfo *pWindowResInfo, int32_t slot);
#define curTimeWindow(_winres) ((_winres)->curIndex)
bool isWindowResClosed(SWindowResInfo *pWindowResInfo, int32_t slot);
void createQueryResultInfo(SQuery *pQuery, SWindowResult *pResultRow, bool isSTableQuery, SPosInfo *posInfo);
void createQueryResultInfo(SQuery *pQuery, SWindowResult *pResultRow, bool isSTableQuery, SPosInfo *posInfo, size_t interBufSize);
char *getPosInResultPage(SQueryRuntimeEnv *pRuntimeEnv, int32_t columnIndex, SWindowResult *pResult);
......
......@@ -272,7 +272,7 @@ bool top_bot_datablock_filter(SQLFunctionCtx *pCtx, int32_t functionId, char *mi
bool stableQueryFunctChanged(int32_t funcId);
void resetResultInfo(SResultInfo *pResInfo);
void setResultInfoBuf(SResultInfo *pResInfo, int32_t size, bool superTable);
void setResultInfoBuf(SResultInfo *pResInfo, int32_t size, bool superTable, char* buf);
static FORCE_INLINE void initResultInfo(SResultInfo *pResInfo) {
pResInfo->initialized = true; // the this struct has been initialized flag
......
此差异已折叠。
......@@ -17,15 +17,24 @@
#include "hash.h"
#include "taosmsg.h"
#include "qextbuffer.h"
#include "ttime.h"
#include "qfill.h"
#include "ttime.h"
#include "qExecutor.h"
#include "qUtil.h"
int32_t getOutputInterResultBufSize(SQuery* pQuery) {
int32_t size = 0;
for (int32_t i = 0; i < pQuery->numOfOutput; ++i) {
assert(pQuery->pSelectExpr[i].interBytes <= DEFAULT_INTERN_BUF_PAGE_SIZE);
size += pQuery->pSelectExpr[i].interBytes;
}
assert(size > 0);
return size;
}
int32_t initWindowResInfo(SWindowResInfo *pWindowResInfo, SQueryRuntimeEnv *pRuntimeEnv, int32_t size,
int32_t threshold, int16_t type) {
pWindowResInfo->capacity = size;
......@@ -43,7 +52,7 @@ int32_t initWindowResInfo(SWindowResInfo *pWindowResInfo, SQueryRuntimeEnv *pRun
pWindowResInfo->pResult = calloc(threshold, sizeof(SWindowResult));
for (int32_t i = 0; i < pWindowResInfo->capacity; ++i) {
SPosInfo posInfo = {-1, -1};
createQueryResultInfo(pRuntimeEnv->pQuery, &pWindowResInfo->pResult[i], pRuntimeEnv->stableQuery, &posInfo);
createQueryResultInfo(pRuntimeEnv->pQuery, &pWindowResInfo->pResult[i], pRuntimeEnv->stableQuery, &posInfo, pRuntimeEnv->interBufSize);
}
return TSDB_CODE_SUCCESS;
......@@ -54,11 +63,7 @@ void destroyTimeWindowRes(SWindowResult *pWindowRes, int32_t nOutputCols) {
return;
}
// TODO opt malloc strategy
for (int32_t i = 0; i < nOutputCols; ++i) {
free(pWindowRes->resultInfo[i].interResultBuf);
}
free(pWindowRes->resultInfo[0].interResultBuf);
free(pWindowRes->resultInfo);
}
......@@ -241,10 +246,9 @@ void clearTimeWindowResBuf(SQueryRuntimeEnv *pRuntimeEnv, SWindowResult *pWindow
}
pWindowRes->numOfRows = 0;
// pWindowRes->nAlloc = 0;
pWindowRes->pos = (SPosInfo){-1, -1};
pWindowRes->status.closed = false;
pWindowRes->window = (STimeWindow){0, 0};
pWindowRes->window = TSWINDOW_INITIALIZER;
}
/**
......@@ -254,7 +258,6 @@ void clearTimeWindowResBuf(SQueryRuntimeEnv *pRuntimeEnv, SWindowResult *pWindow
*/
void copyTimeWindowResBuf(SQueryRuntimeEnv *pRuntimeEnv, SWindowResult *dst, const SWindowResult *src) {
dst->numOfRows = src->numOfRows;
// dst->nAlloc = src->nAlloc;
dst->window = src->window;
dst->status = src->status;
......
......@@ -174,7 +174,7 @@ int32_t taosNumOfRemainRows(SFillInfo* pFillInfo) {
return 0;
}
return FILL_IS_ASC_FILL(pFillInfo) ? (pFillInfo->numOfRows - pFillInfo->rowIdx) : pFillInfo->rowIdx + 1;
return pFillInfo->numOfRows - pFillInfo->rowIdx;
}
// todo: refactor
......
......@@ -1366,6 +1366,7 @@ static int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen) {
}
int32_t compLen = LZ4_compress_default(pCont, buf, contLen, contLen + overhead);
tDebug("compress rpc msg, before:%d, after:%d, overhead:%d", contLen, compLen, overhead);
/*
* only the compressed size is less than the value of contLen - overhead, the compression is applied
......@@ -1378,7 +1379,7 @@ static int32_t rpcCompressRpcMsg(char* pCont, int32_t contLen) {
memcpy(pCont + overhead, buf, compLen);
pHead->comp = 1;
//tDebug("compress rpc msg, before:%d, after:%d", contLen, compLen);
tDebug("compress rpc msg, before:%d, after:%d", contLen, compLen);
finalLen = compLen + overhead;
} else {
finalLen = contLen;
......
......@@ -24,14 +24,13 @@ extern "C" {
#include "tref.h"
#include "hash.h"
typedef void (*__cache_freeres_fn_t)(void*);
typedef void (*__cache_free_fn_t)(void*);
typedef struct SCacheStatis {
int64_t missCount;
int64_t hitCount;
int64_t totalAccess;
int64_t refreshCount;
int32_t numOfCollision;
} SCacheStatis;
typedef struct SCacheDataNode {
......@@ -70,7 +69,7 @@ typedef struct {
// void * pTimer;
SCacheStatis statistics;
SHashObj * pHashTable;
__cache_freeres_fn_t freeFp;
__cache_free_fn_t freeFp;
uint32_t numOfElemsInTrash; // number of element in trash
uint8_t deleting; // set the deleting flag to stop refreshing ASAP.
pthread_t refreshWorker;
......@@ -91,15 +90,7 @@ typedef struct {
* @param fn free resource callback function
* @return
*/
SCacheObj *taosCacheInit(int32_t keyType, int64_t refreshTimeInSeconds, bool extendLifespan, __cache_freeres_fn_t fn, const char *cacheName);
/**
* initialize the cache object and set the free object callback function
* @param refreshTimeInSeconds
* @param freeCb
* @return
*/
SCacheObj *taosCacheInitWithCb(int32_t keyType, int64_t refreshTimeInSeconds, bool extendLifespan, __cache_freeres_fn_t fn, const char *cacheName);
SCacheObj *taosCacheInit(int32_t keyType, int64_t refreshTimeInSeconds, bool extendLifespan, __cache_free_fn_t fn, const char *cacheName);
/**
* add data into cache
......@@ -163,9 +154,8 @@ void taosCacheRelease(SCacheObj *pCacheObj, void **data, bool _remove);
/**
* move all data node into trash, clear node in trash can if it is not referenced by any clients
* @param handle
* @param _remove remove the data or not if refcount is greater than 0
*/
void taosCacheEmpty(SCacheObj *pCacheObj, bool _remove);
void taosCacheEmpty(SCacheObj *pCacheObj);
/**
* release all allocated memory and destroy the cache object.
......@@ -180,6 +170,14 @@ void taosCacheEmpty(SCacheObj *pCacheObj, bool _remove);
*/
void taosCacheCleanup(SCacheObj *pCacheObj);
/**
*
* @param pCacheObj
* @param fp
* @return
*/
void taosCacheRefresh(SCacheObj *pCacheObj, __cache_free_fn_t fp);
#ifdef __cplusplus
}
#endif
......
......@@ -51,6 +51,7 @@ typedef struct SSkipListNode {
#define SL_GET_NODE_KEY(s, n) ((s)->keyFn(SL_GET_NODE_DATA(n)))
#define SL_GET_SL_MIN_KEY(s) (SL_GET_NODE_KEY((s), SL_GET_FORWARD_POINTER((s)->pHead, 0)))
#define SL_GET_SL_MAX_KEY(s) (SL_GET_NODE_KEY((s), SL_GET_BACKWARD_POINTER((s)->pTail, 0)))
#define SL_GET_NODE_LEVEL(n) *(uint8_t *)((n))
......@@ -119,7 +120,6 @@ typedef struct SSkipList {
pthread_rwlock_t *lock;
SSkipListNode * pHead; // point to the first element
SSkipListNode * pTail; // point to the last element
void * lastKey; // last key in the skiplist
#if SKIP_LIST_RECORD_PERFORMANCE
tSkipListState state; // skiplist state
#endif
......
......@@ -223,9 +223,9 @@ static void doCleanupDataCache(SCacheObj *pCacheObj);
* refresh cache to remove data in both hash list and trash, if any nodes' refcount == 0, every pCacheObj->refreshTime
* @param handle Cache object handle
*/
static void* taosCacheRefresh(void *handle);
static void* taosCacheTimedRefresh(void *pCacheObj);
SCacheObj *taosCacheInit(int32_t keyType, int64_t refreshTimeInSeconds, bool extendLifespan, __cache_freeres_fn_t fn, const char* cacheName) {
SCacheObj *taosCacheInit(int32_t keyType, int64_t refreshTimeInSeconds, bool extendLifespan, __cache_free_fn_t fn, const char* cacheName) {
if (refreshTimeInSeconds <= 0) {
return NULL;
}
......@@ -261,7 +261,7 @@ SCacheObj *taosCacheInit(int32_t keyType, int64_t refreshTimeInSeconds, bool ext
pthread_attr_init(&thattr);
pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE);
pthread_create(&pCacheObj->refreshWorker, &thattr, taosCacheRefresh, pCacheObj);
pthread_create(&pCacheObj->refreshWorker, &thattr, taosCacheTimedRefresh, pCacheObj);
pthread_attr_destroy(&thattr);
return pCacheObj;
......@@ -450,7 +450,7 @@ void taosCacheRelease(SCacheObj *pCacheObj, void **data, bool _remove) {
}
}
void taosCacheEmpty(SCacheObj *pCacheObj, bool _remove) {
void taosCacheEmpty(SCacheObj *pCacheObj) {
SHashMutableIterator *pIter = taosHashCreateIter(pCacheObj->pHashTable);
__cache_wr_lock(pCacheObj);
......@@ -459,8 +459,8 @@ void taosCacheEmpty(SCacheObj *pCacheObj, bool _remove) {
break;
}
SCacheDataNode *pNode = *(SCacheDataNode **)taosHashIterGet(pIter);
if (T_REF_VAL_GET(pNode) == 0 || _remove) {
SCacheDataNode *pNode = *(SCacheDataNode **) taosHashIterGet(pIter);
if (T_REF_VAL_GET(pNode) == 0) {
taosCacheReleaseNode(pCacheObj, pNode);
} else {
taosCacheMoveToTrash(pCacheObj, pNode);
......@@ -469,7 +469,7 @@ void taosCacheEmpty(SCacheObj *pCacheObj, bool _remove) {
__cache_unlock(pCacheObj);
taosHashDestroyIter(pIter);
taosTrashCanEmpty(pCacheObj, _remove);
taosTrashCanEmpty(pCacheObj, false);
}
void taosCacheCleanup(SCacheObj *pCacheObj) {
......@@ -623,8 +623,29 @@ void doCleanupDataCache(SCacheObj *pCacheObj) {
free(pCacheObj);
}
void* taosCacheRefresh(void *handle) {
SCacheObj *pCacheObj = (SCacheObj *)handle;
static void doCacheRefresh(SCacheObj* pCacheObj, int64_t time, __cache_free_fn_t fp) {
SHashMutableIterator *pIter = taosHashCreateIter(pCacheObj->pHashTable);
__cache_wr_lock(pCacheObj);
while (taosHashIterNext(pIter)) {
SCacheDataNode *pNode = *(SCacheDataNode **)taosHashIterGet(pIter);
if ((pNode->addedTime + pNode->lifespan * pNode->extendFactor) <= time && T_REF_VAL_GET(pNode) <= 0) {
taosCacheReleaseNode(pCacheObj, pNode);
continue;
}
if (fp) {
fp(pNode->data);
}
}
__cache_unlock(pCacheObj);
taosHashDestroyIter(pIter);
}
void* taosCacheTimedRefresh(void *handle) {
SCacheObj* pCacheObj = handle;
if (pCacheObj == NULL) {
uDebug("object is destroyed. no refresh retry");
return NULL;
......@@ -657,21 +678,8 @@ void* taosCacheRefresh(void *handle) {
// refresh data in hash table
if (elemInHash > 0) {
int64_t expiredTime = taosGetTimestampMs();
SHashMutableIterator *pIter = taosHashCreateIter(pCacheObj->pHashTable);
__cache_wr_lock(pCacheObj);
while (taosHashIterNext(pIter)) {
SCacheDataNode *pNode = *(SCacheDataNode **)taosHashIterGet(pIter);
if ((pNode->addedTime + pNode->lifespan * pNode->extendFactor) <= expiredTime && T_REF_VAL_GET(pNode) <= 0) {
taosCacheReleaseNode(pCacheObj, pNode);
}
}
__cache_unlock(pCacheObj);
taosHashDestroyIter(pIter);
int64_t now = taosGetTimestampMs();
doCacheRefresh(pCacheObj, now, NULL);
}
taosTrashCanEmpty(pCacheObj, false);
......@@ -679,3 +687,12 @@ void* taosCacheRefresh(void *handle) {
return NULL;
}
void taosCacheRefresh(SCacheObj *pCacheObj, __cache_free_fn_t fp) {
if (pCacheObj == NULL) {
return;
}
int64_t now = taosGetTimestampMs();
doCacheRefresh(pCacheObj, now, fp);
}
......@@ -93,16 +93,18 @@ void taosCloseQueue(taos_queue param) {
void *taosAllocateQitem(int size) {
STaosQnode *pNode = (STaosQnode *)calloc(sizeof(STaosQnode) + size, 1);
if (pNode == NULL) return NULL;
uTrace("item:%p, node:%p is allocated", pNode->item, pNode);
return (void *)pNode->item;
}
void taosFreeQitem(void *param) {
if (param == NULL) return;
uTrace("item:%p is freed", param);
char *temp = (char *)param;
temp -= sizeof(STaosQnode);
uTrace("item:%p, node:%p is freed", param, temp);
free(temp);
}
......
......@@ -5,6 +5,7 @@
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
......@@ -238,7 +239,7 @@ SSkipListNode *tSkipListPut(SSkipList *pSkipList, SSkipListNode *pNode) {
// if the new key is greater than the maximum key of skip list, push back this node at the end of skip list
char *newDatakey = SL_GET_NODE_KEY(pSkipList, pNode);
if (pSkipList->size == 0 || pSkipList->comparFn(pSkipList->lastKey, newDatakey) < 0) {
if (pSkipList->size == 0 || pSkipList->comparFn(SL_GET_SL_MAX_KEY(pSkipList), newDatakey) < 0) {
return tSkipListPushBack(pSkipList, pNode);
}
......@@ -498,7 +499,7 @@ void tSkipListRemoveNode(SSkipList *pSkipList, SSkipListNode *pNode) {
if (pSkipList->lock) {
pthread_rwlock_wrlock(pSkipList->lock);
}
for (int32_t j = level - 1; j >= 0; --j) {
SSkipListNode* prev = SL_GET_BACKWARD_POINTER(pNode, j);
SSkipListNode* next = SL_GET_FORWARD_POINTER(pNode, j);
......@@ -699,8 +700,6 @@ SSkipListNode* tSkipListPushBack(SSkipList *pSkipList, SSkipListNode *pNode) {
SL_GET_BACKWARD_POINTER(pSkipList->pTail, i) = pNode;
}
pSkipList->lastKey = SL_GET_NODE_KEY(pSkipList, pNode);
atomic_add_fetch_32(&pSkipList->size, 1);
if (pSkipList->lock) {
pthread_rwlock_unlock(pSkipList->lock);
......
......@@ -383,10 +383,15 @@ int taosOpenTcpServerSocket(uint32_t ip, uint16_t port) {
return -1;
}
if (taosKeepTcpAlive(sockFd) < 0) return -1;
if (taosKeepTcpAlive(sockFd) < 0) {
uError("failed to set tcp server keep-alive option, 0x%x:%hu(%s)", ip, port, strerror(errno));
close(sockFd);
return -1;
}
if (listen(sockFd, 10) < 0) {
uError("listen tcp server socket failed, 0x%x:%hu(%s)", ip, port, strerror(errno));
close(sockFd);
return -1;
}
......
......@@ -161,11 +161,17 @@ int32_t vnodeDrop(int32_t vgId) {
int32_t vnodeAlter(void *param, SMDCreateVnodeMsg *pVnodeCfg) {
SVnodeObj *pVnode = param;
if (pVnode->status != TAOS_VN_STATUS_READY)
return TSDB_CODE_VND_INVALID_STATUS;
// vnode in non-ready state and still needs to return success instead of TSDB_CODE_VND_INVALID_STATUS
// cfgVersion can be corrected by status msg
if (pVnode->status != TAOS_VN_STATUS_READY) {
vDebug("vgId:%d, vnode is not ready, do alter operation later", pVnode->vgId);
return TSDB_CODE_SUCCESS;
}
if (pVnode->syncCfg.replica > 1 && pVnode->role == TAOS_SYNC_ROLE_UNSYNCED)
return TSDB_CODE_VND_NOT_SYNCED;
// the vnode may always fail to synchronize because of it in low cfgVersion
// so cannot use the following codes
// if (pVnode->syncCfg.replica > 1 && pVnode->role == TAOS_SYNC_ROLE_UNSYNCED)
// return TSDB_CODE_VND_NOT_SYNCED;
pVnode->status = TAOS_VN_STATUS_UPDATING;
......
......@@ -61,7 +61,7 @@ int32_t vnodeProcessRead(void *param, SReadMsg *pReadMsg) {
}
static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
void * pCont = pReadMsg->pCont;
void *pCont = pReadMsg->pCont;
int32_t contLen = pReadMsg->contLen;
SRspRet *pRet = &pReadMsg->rspRet;
......@@ -74,19 +74,14 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
killQueryMsg->free = htons(killQueryMsg->free);
killQueryMsg->qhandle = htobe64(killQueryMsg->qhandle);
void* handle = NULL;
if ((void**) killQueryMsg->qhandle != NULL) {
handle = *(void**) killQueryMsg->qhandle;
}
vWarn("QInfo:%p connection %p broken, kill query", handle, pReadMsg->rpcMsg.handle);
vWarn("QInfo:%p connection %p broken, kill query", (void*) killQueryMsg->qhandle, pReadMsg->rpcMsg.handle);
assert(pReadMsg->rpcMsg.contLen > 0 && killQueryMsg->free == 1);
void** qhandle = qAcquireQInfo(pVnode->qMgmt, (void**) killQueryMsg->qhandle);
void** qhandle = qAcquireQInfo(pVnode->qMgmt, (uint64_t) killQueryMsg->qhandle);
if (qhandle == NULL || *qhandle == NULL) {
vWarn("QInfo:%p invalid qhandle, no matched query handle, conn:%p", (void*) killQueryMsg->qhandle, pReadMsg->rpcMsg.handle);
} else {
assert(qhandle == (void**) killQueryMsg->qhandle);
assert(*qhandle == (void*) killQueryMsg->qhandle);
qReleaseQInfo(pVnode->qMgmt, (void**) &qhandle, true);
}
......@@ -110,7 +105,7 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
// current connect is broken
if (code == TSDB_CODE_SUCCESS) {
handle = qRegisterQInfo(pVnode->qMgmt, pQInfo);
handle = qRegisterQInfo(pVnode->qMgmt, (uint64_t) pQInfo);
if (handle == NULL) { // failed to register qhandle
pRsp->code = TSDB_CODE_QRY_INVALID_QHANDLE;
......@@ -118,11 +113,11 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
qKillQuery(pQInfo);
} else {
assert(*handle == pQInfo);
pRsp->qhandle = htobe64((uint64_t) (handle));
pRsp->qhandle = htobe64((uint64_t) pQInfo);
}
pQInfo = NULL;
if (handle != NULL && vnodeNotifyCurrentQhandle(pReadMsg->rpcMsg.handle, handle, pVnode->vgId) != TSDB_CODE_SUCCESS) {
if (handle != NULL && vnodeNotifyCurrentQhandle(pReadMsg->rpcMsg.handle, *handle, pVnode->vgId) != TSDB_CODE_SUCCESS) {
vError("vgId:%d, QInfo:%p, query discarded since link is broken, %p", pVnode->vgId, *handle, pReadMsg->rpcMsg.handle);
pRsp->code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
......@@ -136,18 +131,18 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
assert(pQInfo == NULL);
}
if (handle != NULL) {
dnodePutItemIntoReadQueue(pVnode, handle);
dnodePutItemIntoReadQueue(pVnode, *handle);
qReleaseQInfo(pVnode->qMgmt, (void**) &handle, false);
}
vDebug("vgId:%d, QInfo:%p, dnode query msg disposed", vgId, pQInfo);
} else {
assert(pCont != NULL);
handle = qAcquireQInfo(pVnode->qMgmt, (void**) pCont);
handle = qAcquireQInfo(pVnode->qMgmt, (uint64_t) pCont);
if (handle == NULL) {
vWarn("QInfo:%p invalid qhandle in continuing exec query, conn:%p", *(void**) pCont, pReadMsg->rpcMsg.handle);
vWarn("QInfo:%p invalid qhandle in continuing exec query, conn:%p", (void*) pCont, pReadMsg->rpcMsg.handle);
code = TSDB_CODE_QRY_INVALID_QHANDLE;
} else {
vDebug("vgId:%d, QInfo:%p, dnode query msg in progress", pVnode->vgId, *(void**) pCont);
vDebug("vgId:%d, QInfo:%p, dnode query msg in progress", pVnode->vgId, (void*) pCont);
code = TSDB_CODE_VND_ACTION_IN_PROGRESS;
qTableQuery(*handle); // do execute query
}
......@@ -169,10 +164,10 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
memset(pRet, 0, sizeof(SRspRet));
int32_t code = TSDB_CODE_SUCCESS;
void** handle = qAcquireQInfo(pVnode->qMgmt, (void**) pRetrieve->qhandle);
if (handle == NULL || handle != (void**) pRetrieve->qhandle) {
void** handle = qAcquireQInfo(pVnode->qMgmt, pRetrieve->qhandle);
if (handle == NULL || (*handle) != (void*) pRetrieve->qhandle) {
code = TSDB_CODE_QRY_INVALID_QHANDLE;
vDebug("vgId:%d, invalid qhandle in fetch result, QInfo:%p", pVnode->vgId, *(void**) pRetrieve->qhandle);
vDebug("vgId:%d, invalid qhandle in fetch result, QInfo:%p", pVnode->vgId, (void*) pRetrieve->qhandle);
pRet->rsp = (SRetrieveTableRsp *)rpcMallocCont(sizeof(SRetrieveTableRsp));
pRet->len = sizeof(SRetrieveTableRsp);
......@@ -180,8 +175,8 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
memset(pRet->rsp, 0, sizeof(SRetrieveTableRsp));
SRetrieveTableRsp* pRsp = pRet->rsp;
pRsp->numOfRows = 0;
pRsp->completed = true;
pRsp->useconds = 0;
pRsp->completed = true;
return code;
}
......@@ -211,8 +206,8 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
} else { // if failed to dump result, free qhandle immediately
if ((code = qDumpRetrieveResult(*handle, (SRetrieveTableRsp **)&pRet->rsp, &pRet->len)) == TSDB_CODE_SUCCESS) {
if (qHasMoreResultsToRetrieve(*handle)) {
dnodePutItemIntoReadQueue(pVnode, handle);
pRet->qhandle = handle;
dnodePutItemIntoReadQueue(pVnode, *handle);
pRet->qhandle = *handle;
freeHandle = false;
}
}
......
......@@ -49,19 +49,26 @@ int32_t vnodeProcessWrite(void *param1, int qtype, void *param2, void *item) {
SVnodeObj *pVnode = (SVnodeObj *)param1;
SWalHead *pHead = param2;
if (vnodeProcessWriteMsgFp[pHead->msgType] == NULL)
if (vnodeProcessWriteMsgFp[pHead->msgType] == NULL) {
vDebug("vgId:%d, msgType:%s not processed, no handle", pVnode->vgId, taosMsg[pHead->msgType]);
return TSDB_CODE_VND_MSG_NOT_PROCESSED;
}
if (!(pVnode->accessState & TSDB_VN_WRITE_ACCCESS)) {
vDebug("vgId:%d, msgType:%s not processed, no write auth", pVnode->vgId, taosMsg[pHead->msgType]);
return TSDB_CODE_VND_NO_WRITE_AUTH;
}
if (pHead->version == 0) { // from client or CQ
if (pVnode->status != TAOS_VN_STATUS_READY)
if (pVnode->status != TAOS_VN_STATUS_READY) {
vDebug("vgId:%d, msgType:%s not processed, vnode status is %d", pVnode->vgId, taosMsg[pHead->msgType], pVnode->status);
return TSDB_CODE_VND_INVALID_STATUS; // it may be in deleting or closing state
}
if (pVnode->syncCfg.replica > 1 && pVnode->role != TAOS_SYNC_ROLE_MASTER)
if (pVnode->syncCfg.replica > 1 && pVnode->role != TAOS_SYNC_ROLE_MASTER) {
vDebug("vgId:%d, msgType:%s not processed, replica:%d role:%d", pVnode->vgId, taosMsg[pHead->msgType], pVnode->syncCfg.replica, pVnode->role);
return TSDB_CODE_RPC_NOT_READY;
}
// assign version
pVnode->version++;
......
......@@ -28,6 +28,7 @@
#include "taoserror.h"
#include "twal.h"
#include "tqueue.h"
#include "tfile.h"
#define walPrefix "wal"
......@@ -180,7 +181,7 @@ int walWrite(void *handle, SWalHead *pHead) {
taosCalcChecksumAppend(0, (uint8_t *)pHead, sizeof(SWalHead));
int contLen = pHead->len + sizeof(SWalHead);
if(write(pWal->fd, pHead, contLen) != contLen) {
if(twrite(pWal->fd, pHead, contLen) != contLen) {
wError("wal:%s, failed to write(%s)", pWal->name, strerror(errno));
terrno = TAOS_SYSTEM_ERROR(errno);
} else {
......@@ -325,7 +326,7 @@ static int walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp) {
wDebug("wal:%s, start to restore", name);
while (1) {
int ret = read(fd, pHead, sizeof(SWalHead));
int ret = tread(fd, pHead, sizeof(SWalHead));
if ( ret == 0) break;
if (ret != sizeof(SWalHead)) {
......@@ -340,7 +341,7 @@ static int walRestoreWalFile(SWal *pWal, void *pVnode, FWalWrite writeFp) {
break;
}
ret = read(fd, pHead->cont, pHead->len);
ret = tread(fd, pHead->cont, pHead->len);
if ( ret != pHead->len) {
wWarn("wal:%s, failed to read body, skip, len:%d ret:%d", name, pHead->len, ret);
terrno = TAOS_SYSTEM_ERROR(errno);
......
......@@ -40,8 +40,10 @@ class TDTestCase:
ret = tdSql.query('select server_status() as result')
tdSql.checkData(0, 0, 1)
ret = tdSql.execute('alter dnode 127.0.0.1 debugFlag 135')
tdLog.info("alter dnode ret: %d" % ret)
ret = tdSql.query('show dnodes')
ret = tdSql.execute('alter dnode "%s" debugFlag 135' % tdSql.getData(0,1))
tdLog.info('alter dnode "%s" debugFlag 135 -> ret: %d' % (tdSql.getData(0, 1), ret))
def stop(self):
tdSql.close()
......
此差异已折叠。
......@@ -121,7 +121,7 @@ python3 ./test.py -f import_merge/importTORestart.py
python3 ./test.py -f import_merge/importTPORestart.py
python3 ./test.py -f import_merge/importTRestart.py
python3 ./test.py -f import_merge/importInsertThenImport.py
python3 ./test.py -f import_merge/importCSV.py
# user
python3 ./test.py -f user/user_create.py
python3 ./test.py -f user/pass_len.py
......@@ -145,8 +145,11 @@ python3 ./test.py -f query/queryJoin.py
python3 ./test.py -f query/select_last_crash.py
#stream
python3 ./test.py -f stream/metric_1.py
python3 ./test.py -f stream/new.py
python3 ./test.py -f stream/stream1.py
python3 ./test.py -f stream/stream2.py
python3 ./test.py -f stream/parser.py
#alter table
python3 ./test.py -f alter/alter_table_crash.py
......
###################################################################
# 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 os
import csv
import random
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)
self.csvfile = "/tmp/file.csv"
self.rows = 10000
self.ntables = 1
self.startTime = 1520000010000
def genRandomStr(self, maxLen):
H = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789'
salt = ''
if maxLen <= 1:
maxLen = 2
l = random.randint(1,maxLen)
for i in range(l):
salt += random.choice(H)
return salt
def createCSVFile(self):
f = open(self.csvfile,'w',encoding='utf-8')
csv_writer = csv.writer(f, quoting=csv.QUOTE_NONNUMERIC)
for i in range(self.rows):
csv_writer.writerow([self.startTime + i,
self.genRandomStr(5),
self.genRandomStr(6),
self.genRandomStr(7),
self.genRandomStr(8),
self.genRandomStr(9),
self.genRandomStr(10),
self.genRandomStr(11),
self.genRandomStr(12),
self.genRandomStr(13),
self.genRandomStr(14)])
f.close()
def destroyCSVFile(self):
os.remove(self.csvfile)
def run(self):
self.createCSVFile()
tdDnodes.stop(1)
tdDnodes.deploy(1)
tdDnodes.start(1)
tdSql.execute('reset query cache')
tdSql.execute('drop database if exists db')
tdSql.execute('create database db')
tdSql.execute('use db')
tdSql.execute('''create table tbx (ts TIMESTAMP,
collect_area NCHAR(5),
device_id BINARY(6),
imsi BINARY(7),
imei BINARY(8),
mdn BINARY(9),
net_type BINARY(10),
mno NCHAR(11),
province NCHAR(12),
city NCHAR(13),
alarm BINARY(14))''')
tdSql.execute("import into tbx file \'%s\'"%(self.csvfile))
tdSql.query('select * from tbx')
tdSql.checkRows(self.rows)
def stop(self):
self.destroyCSVFile()
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdDnodes.stop(1)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())
......@@ -121,7 +121,7 @@ python3 ./test.py -f import_merge/importTORestart.py
python3 ./test.py -f import_merge/importTPORestart.py
python3 ./test.py -f import_merge/importTRestart.py
python3 ./test.py -f import_merge/importInsertThenImport.py
python3 ./test.py -f import_merge/importCSV.py
# user
python3 ./test.py -f user/user_create.py
python3 ./test.py -f user/pass_len.py
......
###################################################################
# 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 time
import taos
from util.log import tdLog
from util.cases import tdCases
from util.sql import tdSql
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
def createFuncStream(self, expr, suffix, value):
tbname = "strm_" + suffix
tdLog.info("create stream table %s" % tbname)
tdSql.query("select %s from stb interval(1d)" % expr)
tdSql.checkData(0, 1, value)
tdSql.execute("create table %s as select %s from stb interval(1d)" % (tbname, expr))
def checkStreamData(self, suffix, value):
sql = "select * from strm_" + suffix
tdSql.waitedQuery(sql, 1, 120)
tdSql.checkData(0, 1, value)
def run(self):
tbNum = 10
rowNum = 20
tdSql.prepare()
tdLog.info("===== preparing data =====")
tdSql.execute(
"create table stb(ts timestamp, tbcol int, tbcol2 float) tags(tgcol int)")
for i in range(tbNum):
tdSql.execute("create table tb%d using stb tags(%d)" % (i, i))
for j in range(rowNum):
tdSql.execute(
"insert into tb%d values (now - %dm, %d, %d)" %
(i, 1440 - j, j, j))
time.sleep(0.1)
self.createFuncStream("count(*)", "c1", 200)
self.createFuncStream("count(tbcol)", "c2", 200)
self.createFuncStream("count(tbcol2)", "c3", 200)
self.createFuncStream("avg(tbcol)", "av", 9.5)
self.createFuncStream("sum(tbcol)", "su", 1900)
self.createFuncStream("min(tbcol)", "mi", 0)
self.createFuncStream("max(tbcol)", "ma", 19)
self.createFuncStream("first(tbcol)", "fi", 0)
self.createFuncStream("last(tbcol)", "la", 19)
#tdSql.query("select stddev(tbcol) from stb interval(1d)")
#tdSql.query("select leastsquares(tbcol, 1, 1) from stb interval(1d)")
tdSql.query("select top(tbcol, 1) from stb interval(1d)")
tdSql.query("select bottom(tbcol, 1) from stb interval(1d)")
#tdSql.query("select percentile(tbcol, 1) from stb interval(1d)")
#tdSql.query("select diff(tbcol) from stb interval(1d)")
tdSql.query("select count(tbcol) from stb where ts < now + 4m interval(1d)")
tdSql.checkData(0, 1, 200)
#tdSql.execute("create table strm_wh as select count(tbcol) from stb where ts < now + 4m interval(1d)")
self.createFuncStream("count(tbcol)", "as", 200)
tdSql.query("select count(tbcol) from stb interval(1d) group by tgcol")
tdSql.checkData(0, 1, 20)
tdSql.query("select count(tbcol) from stb where ts < now + 4m interval(1d) group by tgcol")
tdSql.checkData(0, 1, 20)
self.checkStreamData("c1", 200)
self.checkStreamData("c2", 200)
self.checkStreamData("c3", 200)
self.checkStreamData("av", 9.5)
self.checkStreamData("su", 1900)
self.checkStreamData("mi", 0)
self.checkStreamData("ma", 19)
self.checkStreamData("fi", 0)
self.checkStreamData("la", 19)
#self.checkStreamData("wh", 200)
self.checkStreamData("as", 200)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())
###################################################################
# 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 time
import taos
from util.log import tdLog
from util.cases import tdCases
from util.sql import tdSql
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
def run(self):
rowNum = 200
totalNum = 200
tdSql.prepare()
tdLog.info("=============== step1")
tdSql.execute("create table mt(ts timestamp, tbcol int, tbcol2 float) TAGS(tgcol int)")
for i in range(5):
tdSql.execute("create table tb%d using mt tags(%d)" % (i, i))
for j in range(rowNum):
tdSql.execute("insert into tb%d values(now + %ds, %d, %d)" % (i, j, j, j))
time.sleep(0.1)
tdLog.info("=============== step2")
tdSql.query("select count(*), count(tbcol), count(tbcol2) from mt interval(10s)")
tdSql.execute("create table st as select count(*), count(tbcol), count(tbcol2) from mt interval(10s)")
tdLog.info("=============== step3")
tdSql.waitedQuery("select * from st", 1, 120)
v = tdSql.getData(0, 3)
if v >= 51:
tdLog.exit("value is %d, which is larger than 51" % v)
tdLog.info("=============== step4")
for i in range(5, 10):
tdSql.execute("create table tb%d using mt tags(%d)" % (i, i))
for j in range(rowNum):
tdSql.execute("insert into tb%d values(now + %ds, %d, %d)" % (i, j, j, j))
tdLog.info("=============== step5")
tdLog.sleep(30)
tdSql.waitedQuery("select * from st order by ts desc", 1, 120)
v = tdSql.getData(0, 3)
if v <= 51:
tdLog.exit("value is %d, which is smaller than 51" % v)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())
###################################################################
# 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 time
import taos
from util.log import tdLog
from util.cases import tdCases
from util.sql import tdSql
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
'''
def bug2222(self):
tdSql.prepare()
tdSql.execute("create table superreal(ts timestamp, addr binary(5), val float) tags (deviceNo binary(20))")
tdSql.execute("create table real_001 using superreal tags('001')")
tdSql.execute("create table tj_001 as select sum(val) from real_001 interval(1m)")
t = datetime.datetime.now()
for i in range(60):
ts = t.strftime("%Y-%m-%d %H:%M")
t += datetime.timedelta(minutes=1)
sql = "insert into real_001 values('%s:0%d', '1', %d)" % (ts, 0, i)
for j in range(4):
sql += ",('%s:0%d', '%d', %d)" % (ts, j + 1, j + 1, i)
tdSql.execute(sql)
time.sleep(60 + random.random() * 60 - 30)
'''
def tbase300(self):
tdLog.debug("begin tbase300")
tdSql.prepare()
tdSql.execute("create table mt(ts timestamp, c1 int, c2 int) tags(t1 int)")
tdSql.execute("create table tb1 using mt tags(1)");
tdSql.execute("create table tb2 using mt tags(2)");
tdSql.execute("create table strm as select count(*), avg(c1), sum(c2), max(c1), min(c2),first(c1), last(c2) from mt interval(4s) sliding(2s)")
#tdSql.execute("create table strm as select count(*), avg(c1), sum(c2), max(c1), min(c2), first(c1) from mt interval(4s) sliding(2s)")
tdLog.sleep(10)
tdSql.execute("insert into tb2 values(now, 1, 1)");
tdSql.execute("insert into tb1 values(now, 1, 1)");
tdLog.sleep(4)
tdSql.query("select * from mt")
tdSql.query("select * from strm")
tdSql.execute("drop table tb1")
tdSql.waitedQuery("select * from strm", 1, 100)
if tdSql.queryRows < 1 or tdSql.queryRows > 2:
tdLog.exit("rows should be 1 or 2")
tdSql.execute("drop table tb2")
tdSql.execute("drop table mt")
tdSql.execute("drop table strm")
def tbase304(self):
tdLog.debug("begin tbase304")
# we cannot reset query cache in server side, as a workaround,
# set super table name to mt304, need to change back to mt later
tdSql.execute("create table mt304 (ts timestamp, c1 int) tags(t1 int, t2 int)")
tdSql.execute("create table tb1 using mt304 tags(1, 1)")
tdSql.execute("create table tb2 using mt304 tags(1, -1)")
time.sleep(0.1)
tdSql.execute("create table strm as select count(*), avg(c1) from mt304 where t2 >= 0 interval(4s) sliding(2s)")
tdSql.execute("insert into tb1 values (now,1)")
tdSql.execute("insert into tb2 values (now,2)")
tdSql.waitedQuery("select * from strm", 1, 100)
if tdSql.queryRows < 1 or tdSql.queryRows > 2:
tdLog.exit("rows should be 1 or 2")
tdSql.checkData(0, 1, 1)
tdSql.checkData(0, 2, 1.000000000)
tdSql.execute("alter table mt304 drop tag t2")
tdSql.execute("insert into tb2 values (now,2)")
tdSql.execute("insert into tb1 values (now,1)")
tdSql.query("select * from strm")
tdSql.execute("alter table mt304 add tag t2 int")
tdLog.sleep(1)
tdSql.query("select * from strm")
def wildcardFilterOnTags(self):
tdLog.debug("begin wildcardFilterOnTag")
tdSql.prepare()
tdSql.execute("create table stb (ts timestamp, c1 int, c2 binary(10)) tags(t1 binary(10))")
tdSql.execute("create table tb1 using stb tags('a1')")
tdSql.execute("create table tb2 using stb tags('b2')")
tdSql.execute("create table tb3 using stb tags('a3')")
tdSql.execute("create table strm as select count(*), avg(c1), first(c2) from stb where t1 like 'a%' interval(4s) sliding(2s)")
tdSql.query("describe strm")
tdSql.checkRows(4)
tdLog.sleep(1)
tdSql.execute("insert into tb1 values (now, 0, 'tb1')")
tdLog.sleep(4)
tdSql.execute("insert into tb2 values (now, 2, 'tb2')")
tdLog.sleep(4)
tdSql.execute("insert into tb3 values (now, 0, 'tb3')")
tdSql.waitedQuery("select * from strm", 4, 60)
tdSql.checkRows(4)
tdSql.checkData(0, 2, 0.000000000)
if tdSql.getData(0, 3) == 'tb2':
tdLog.exit("unexpected value of data03")
if tdSql.getData(1, 3) == 'tb2':
tdLog.exit("unexpected value of data13")
if tdSql.getData(2, 3) == 'tb2':
tdLog.exit("unexpected value of data23")
if tdSql.getData(3, 3) == 'tb2':
tdLog.exit("unexpected value of data33")
tdLog.info("add table tb4 to see if stream still works correctly")
# The vnode client needs to refresh metadata cache to allow strm calculate tb4's data.
# But the current refreshing frequency is every 10 min
# commented out the case below to save running time
tdSql.execute("create table tb4 using stb tags('a4')")
tdSql.execute("insert into tb4 values(now, 4, 'tb4')")
tdSql.waitedQuery("select * from strm order by ts desc", 6, 60)
tdSql.checkRows(6)
tdSql.checkData(0, 2, 4)
tdSql.checkData(0, 3, "tb4")
tdLog.info("change tag values to see if stream still works correctly")
tdSql.execute("alter table tb4 set tag t1='b4'")
tdLog.sleep(3)
tdSql.execute("insert into tb1 values (now, 1, 'tb1_a1')")
tdLog.sleep(4)
tdSql.execute("insert into tb4 values (now, -4, 'tb4_b4')")
tdSql.waitedQuery("select * from strm order by ts desc", 8, 100)
tdSql.checkRows(8)
tdSql.checkData(0, 2, 1)
tdSql.checkData(0, 3, "tb1_a1")
def datatypes(self):
tdLog.debug("begin data types")
tdSql.prepare()
tdSql.execute("create table stb3 (ts timestamp, c1 int, c2 bigint, c3 float, c4 double, c5 binary(15), c6 nchar(15), c7 bool) tags(t1 int, t2 binary(15))")
tdSql.execute("create table tb0 using stb3 tags(0, 'tb0')")
tdSql.execute("create table tb1 using stb3 tags(1, 'tb1')")
tdSql.execute("create table tb2 using stb3 tags(2, 'tb2')")
tdSql.execute("create table tb3 using stb3 tags(3, 'tb3')")
tdSql.execute("create table tb4 using stb3 tags(4, 'tb4')")
tdSql.execute("create table strm0 as select count(ts), count(c1), max(c2), min(c4), first(c5), last(c6) from stb3 where ts < now + 30s interval(4s) sliding(2s)")
#tdSql.execute("create table strm0 as select count(ts), count(c1), max(c2), min(c4), first(c5) from stb where ts < now + 30s interval(4s) sliding(2s)")
tdLog.sleep(1)
tdSql.execute("insert into tb0 values (now, 0, 0, 0, 0, 'binary0', '涛思0', true) tb1 values (now, 1, 1, 1, 1, 'binary1', '涛思1', false) tb2 values (now, 2, 2, 2, 2, 'binary2', '涛思2', true) tb3 values (now, 3, 3, 3, 3, 'binary3', '涛思3', false) tb4 values (now, 4, 4, 4, 4, 'binary4', '涛思4', true) ")
tdSql.waitedQuery("select * from strm0 order by ts desc", 2, 120)
tdSql.checkRows(2)
tdSql.execute("insert into tb0 values (now, 10, 10, 10, 10, 'binary0', '涛思0', true) tb1 values (now, 11, 11, 11, 11, 'binary1', '涛思1', false) tb2 values (now, 12, 12, 12, 12, 'binary2', '涛思2', true) tb3 values (now, 13, 13, 13, 13, 'binary3', '涛思3', false) tb4 values (now, 14, 14, 14, 14, 'binary4', '涛思4', true) ")
tdSql.waitedQuery("select * from strm0 order by ts desc", 4, 120)
tdSql.checkRows(4)
def run(self):
self.tbase300()
self.tbase304()
self.wildcardFilterOnTags()
self.datatypes()
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())
......@@ -55,10 +55,7 @@ class TDTestCase:
tdSql.checkRows(tbNum + 1)
tdLog.info("===== step3 =====")
tdLog.info("sleeping 120 seconds")
time.sleep(120)
tdSql.query("select * from s0")
tdSql.waitedQuery("select * from s0", 1, 120)
try:
tdSql.checkData(0, 1, rowNum)
tdSql.checkData(0, 2, rowNum)
......@@ -82,10 +79,7 @@ class TDTestCase:
tdSql.checkRows(tbNum + 1)
tdLog.info("===== step7 =====")
tdLog.info("sleeping 120 seconds")
time.sleep(120)
tdSql.query("select * from s0")
tdSql.waitedQuery("select * from s0", 1, 120)
try:
tdSql.checkData(0, 1, rowNum)
tdSql.checkData(0, 2, rowNum)
......@@ -108,10 +102,7 @@ class TDTestCase:
tdSql.checkRows(tbNum + 2)
tdLog.info("===== step9 =====")
tdLog.info("sleeping 120 seconds")
time.sleep(120)
tdSql.query("select * from s1")
tdSql.waitedQuery("select * from s1", 1, 120)
try:
tdSql.checkData(0, 1, rowNum * tbNum)
tdSql.checkData(0, 2, rowNum * tbNum)
......@@ -134,9 +125,7 @@ class TDTestCase:
tdSql.checkRows(tbNum + 2)
tdLog.info("===== step13 =====")
tdLog.info("sleeping 120 seconds")
time.sleep(120)
tdSql.query("select * from s1")
tdSql.waitedQuery("select * from s1", 1, 120)
try:
tdSql.checkData(0, 1, rowNum * tbNum)
tdSql.checkData(0, 2, rowNum * tbNum)
......
......@@ -53,8 +53,7 @@ class TDTestCase:
tdSql.checkRows(tbNum + 1)
tdLog.info("===== step3 =====")
time.sleep(120)
tdSql.query("select * from s0")
tdSql.waitedQuery("select * from s0", 1, 120)
try:
tdSql.checkData(0, 1, rowNum)
except Exception as e:
......@@ -81,8 +80,7 @@ class TDTestCase:
tdLog.info(repr(e))
tdLog.info("===== step7 =====")
time.sleep(120)
tdSql.query("select * from s0")
tdSql.waitedQuery("select * from s0", 1, 120)
try:
tdSql.checkData(0, 1, rowNum)
tdSql.checkData(0, 2, rowNum)
......@@ -107,8 +105,7 @@ class TDTestCase:
tdSql.checkRows(tbNum + 2)
tdLog.info("===== step9 =====")
time.sleep(120)
tdSql.query("select * from s1")
tdSql.waitedQuery("select * from s1", 1, 120)
try:
tdSql.checkData(0, 1, totalNum)
tdSql.checkData(0, 2, totalNum)
......@@ -137,8 +134,7 @@ class TDTestCase:
tdLog.info(repr(e))
tdLog.info("===== step13 =====")
time.sleep(120)
tdSql.query("select * from s1")
tdSql.waitedQuery("select * from s1", 1, 120)
try:
tdSql.checkData(0, 1, totalNum)
#tdSql.checkData(0, 2, None)
......
此差异已折叠。
此差异已折叠。
......@@ -22,35 +22,59 @@ class TDSimClient:
def __init__(self):
self.testCluster = False
self.cfgDict = {
"numOfLogLines": "100000000",
"numOfThreadsPerCore": "2.0",
"locale": "en_US.UTF-8",
"charset": "UTF-8",
"asyncLog": "0",
"anyIp": "0",
"sdbDebugFlag": "135",
"rpcDebugFlag": "135",
"tmrDebugFlag": "131",
"cDebugFlag": "135",
"udebugFlag": "135",
"jnidebugFlag": "135",
"qdebugFlag": "135",
}
def init(self, path):
self.__init__()
self.path = path
def getLogDir(self):
self.logDir = "%s/sim/psim/log" % (self.path)
return self.logDir
def getCfgDir(self):
self.cfgDir = "%s/sim/psim/cfg" % (self.path)
return self.cfgDir
def setTestCluster(self, value):
self.testCluster = value
def addExtraCfg(self, option, value):
self.cfgDict.update({option: value})
def cfg(self, option, value):
cmd = "echo '%s %s' >> %s" % (option, value, self.cfgPath)
if os.system(cmd) != 0:
tdLog.exit(cmd)
def deploy(self):
self.logDir = "%s/sim/psim/log" % (self.path,)
self.logDir = "%s/sim/psim/log" % (self.path)
self.cfgDir = "%s/sim/psim/cfg" % (self.path)
self.cfgPath = "%s/sim/psim/cfg/taos.cfg" % (self.path)
cmd = "rm -rf " + self.logDir
if os.system(cmd) != 0:
tdLog.exit(cmd)
cmd = "rm -rf " + self.cfgDir
cmd = "mkdir -p " + self.logDir
if os.system(cmd) != 0:
tdLog.exit(cmd)
cmd = "mkdir -p " + self.logDir
cmd = "rm -rf " + self.cfgDir
if os.system(cmd) != 0:
tdLog.exit(cmd)
......@@ -66,19 +90,10 @@ class TDSimClient:
self.cfg("masterIp", "192.168.0.1")
self.cfg("secondIp", "192.168.0.2")
self.cfg("logDir", self.logDir)
self.cfg("numOfLogLines", "100000000")
self.cfg("numOfThreadsPerCore", "2.0")
self.cfg("locale", "en_US.UTF-8")
self.cfg("charset", "UTF-8")
self.cfg("asyncLog", "0")
self.cfg("anyIp", "0")
self.cfg("sdbDebugFlag", "135")
self.cfg("rpcDebugFlag", "135")
self.cfg("tmrDebugFlag", "131")
self.cfg("cDebugFlag", "135")
self.cfg("udebugFlag", "135")
self.cfg("jnidebugFlag", "135")
self.cfg("qdebugFlag", "135")
for key, value in self.cfgDict.items():
self.cfg(key, value)
tdLog.debug("psim is deployed and configured by %s" % (self.cfgPath))
......@@ -378,6 +393,9 @@ class TDDnodes:
for i in range(len(self.dnodes)):
self.dnodes[i].init(self.path)
self.sim = TDSimClient()
self.sim.init(self.path)
def setTestCluster(self, value):
self.testCluster = value
......@@ -385,8 +403,6 @@ class TDDnodes:
self.valgrind = value
def deploy(self, index):
self.sim = TDSimClient()
self.sim.init(self.path)
self.sim.setTestCluster(self.testCluster)
if (self.simDeployed == False):
......@@ -474,5 +490,11 @@ class TDDnodes:
def getSimCfgPath(self):
return self.sim.getCfgDir()
def getSimLogPath(self):
return self.sim.getLogDir()
def addSimExtraCfg(self, option, value):
self.sim.addExtraCfg(option, value)
tdDnodes = TDDnodes()
......@@ -29,10 +29,8 @@ class TDSql:
self.cursor = cursor
if (log):
frame = inspect.stack()[1]
callerModule = inspect.getmodule(frame[0])
callerFilename = callerModule.__file__
self.cursor.log(callerFilename + ".sql")
caller = inspect.getframeinfo(inspect.stack()[1][0])
self.cursor.log(caller.filename + ".sql")
def close(self):
self.cursor.close()
......@@ -55,12 +53,8 @@ class TDSql:
except BaseException:
expectErrNotOccured = False
if expectErrNotOccured:
frame = inspect.stack()[1]
callerModule = inspect.getmodule(frame[0])
callerFilename = callerModule.__file__
tdLog.exit(
"%s failed: sql:%s, expect error not occured" %
(callerFilename, sql))
caller = inspect.getframeinfo(inspect.stack()[1][0])
tdLog.exit("%s(%d) failed: sql:%s, expect error not occured" % (caller.filename, caller.lineno, sql))
else:
self.queryRows = 0
self.queryCols = 0
......@@ -69,75 +63,70 @@ class TDSql:
def query(self, sql):
self.sql = sql
self.cursor.execute(sql)
self.queryResult = self.cursor.fetchall()
self.queryRows = len(self.queryResult)
self.queryCols = len(self.cursor.description)
# if self.queryRows == 1 and self.queryCols == 1:
# tdLog.info("sql:%s, rows:%d cols:%d data:%s" % (self.sql, self.queryRows, self.queryCols, self.queryResult[0][0]))
# else:
# tdLog.info("sql:%s, rows:%d cols:%d" % (self.sql, self.queryRows, self.queryCols))
try:
self.cursor.execute(sql)
self.queryResult = self.cursor.fetchall()
self.queryRows = len(self.queryResult)
self.queryCols = len(self.cursor.description)
except Exception as e:
caller = inspect.getframeinfo(inspect.stack()[1][0])
args = (caller.filename, caller.lineno, sql, repr(e))
tdLog.notice("%s(%d) failed: sql:%s, %s" % args)
raise Exception(repr(e))
return self.queryRows
def checkRows(self, expectRows):
if self.queryRows != expectRows:
frame = inspect.stack()[1]
callerModule = inspect.getmodule(frame[0])
callerFilename = callerModule.__file__
tdLog.exit(
"%s failed: sql:%s, queryRows:%d != expect:%d" %
(callerFilename, self.sql, self.queryRows, expectRows))
tdLog.info("sql:%s, queryRows:%d == expect:%d" %
(self.sql, self.queryRows, expectRows))
def waitedQuery(self, sql, expectRows, timeout):
tdLog.info("sql: %s, try to retrieve %d rows in %d seconds" % (sql, expectRows, timeout))
self.sql = sql
try:
for i in range(timeout):
self.cursor.execute(sql)
self.queryResult = self.cursor.fetchall()
self.queryRows = len(self.queryResult)
self.queryCols = len(self.cursor.description)
if self.queryRows >= expectRows:
return (self.queryRows, i)
time.sleep(1)
except Exception as e:
caller = inspect.getframeinfo(inspect.stack()[1][0])
args = (caller.filename, caller.lineno, sql, repr(e))
tdLog.notice("%s(%d) failed: sql:%s, %s" % args)
raise Exception(repr(e))
return (self.queryRows, timeout)
def checkDataType(self, row, col, dataType):
frame = inspect.stack()[1]
callerModule = inspect.getmodule(frame[0])
callerFilename = callerModule.__file__
def checkRows(self, expectRows):
if self.queryRows == expectRows:
tdLog.info("sql:%s, queryRows:%d == expect:%d" % (self.sql, self.queryRows, expectRows))
else:
caller = inspect.getframeinfo(inspect.stack()[1][0])
args = (caller.filename, caller.lineno, self.sql, self.queryRows, expectRows)
tdLog.exit("%s(%d) failed: sql:%s, queryRows:%d != expect:%d" % args)
def checkRowCol(self, row, col):
caller = inspect.getframeinfo(inspect.stack()[2][0])
if row < 0:
tdLog.exit(
"%s failed: sql:%s, row:%d is smaller than zero" %
(callerFilename, self.sql, row))
args = (caller.filename, caller.lineno, self.sql, row)
tdLog.exit("%s(%d) failed: sql:%s, row:%d is smaller than zero" % args)
if col < 0:
tdLog.exit(
"%s failed: sql:%s, col:%d is smaller than zero" %
(callerFilename, self.sql, col))
args = (caller.filename, caller.lineno, self.sql, row)
tdLog.exit("%s(%d) failed: sql:%s, col:%d is smaller than zero" % args)
if row > self.queryRows:
tdLog.exit(
"%s failed: sql:%s, row:%d is larger than queryRows:%d" %
(callerFilename, self.sql, row, self.queryRows))
args = (caller.filename, caller.lineno, self.sql, row, self.queryRows)
tdLog.exit("%s(%d) failed: sql:%s, row:%d is larger than queryRows:%d" % args)
if col > self.queryCols:
tdLog.exit(
"%s failed: sql:%s, col:%d is larger than queryCols:%d" %
(callerFilename, self.sql, col, self.queryCols))
args = (caller.filename, caller.lineno, self.sql, col, self.queryCols)
tdLog.exit("%s(%d) failed: sql:%s, col:%d is larger than queryCols:%d" % args)
def checkDataType(self, row, col, dataType):
self.checkRowCol(row, col)
return self.cursor.istype(col, dataType)
def checkData(self, row, col, data):
frame = inspect.stack()[1]
callerModule = inspect.getmodule(frame[0])
callerFilename = callerModule.__file__
if row < 0:
tdLog.exit(
"%s failed: sql:%s, row:%d is smaller than zero" %
(callerFilename, self.sql, row))
if col < 0:
tdLog.exit(
"%s failed: sql:%s, col:%d is smaller than zero" %
(callerFilename, self.sql, col))
if row > self.queryRows:
tdLog.exit(
"%s failed: sql:%s, row:%d is larger than queryRows:%d" %
(callerFilename, self.sql, row, self.queryRows))
if col > self.queryCols:
tdLog.exit(
"%s failed: sql:%s, col:%d is larger than queryCols:%d" %
(callerFilename, self.sql, col, self.queryCols))
self.checkRowCol(row, col)
if self.queryResult[row][col] != data:
tdLog.exit("%s failed: sql:%s row:%d col:%d data:%s != expect:%s" % (
callerFilename, self.sql, row, col, self.queryResult[row][col], data))
caller = inspect.getframeinfo(inspect.stack()[1][0])
args = (caller.filename, caller.lineno, self.sql, row, col, self.queryResult[row][col], data)
tdLog.exit("%s(%d) failed: sql:%s row:%d col:%d data:%s != expect:%s" % args)
if data is None:
tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" %
......@@ -153,26 +142,7 @@ class TDSql:
(self.sql, row, col, self.queryResult[row][col], data))
def getData(self, row, col):
frame = inspect.stack()[1]
callerModule = inspect.getmodule(frame[0])
callerFilename = callerModule.__file__
if row < 0:
tdLog.exit(
"%s failed: sql:%s, row:%d is smaller than zero" %
(callerFilename, self.sql, row))
if col < 0:
tdLog.exit(
"%s failed: sql:%s, col:%d is smaller than zero" %
(callerFilename, self.sql, col))
if row > self.queryRows:
tdLog.exit(
"%s failed: sql:%s, row:%d is larger than queryRows:%d" %
(callerFilename, self.sql, row, self.queryRows))
if col > self.queryCols:
tdLog.exit(
"%s failed: sql:%s, col:%d is larger than queryCols:%d" %
(callerFilename, self.sql, col, self.queryCols))
self.checkRowCol(row, col)
return self.queryResult[row][col]
def executeTimes(self, sql, times):
......@@ -185,20 +155,22 @@ class TDSql:
def execute(self, sql):
self.sql = sql
self.affectedRows = self.cursor.execute(sql)
try:
self.affectedRows = self.cursor.execute(sql)
except Exception as e:
caller = inspect.getframeinfo(inspect.stack()[1][0])
args = (caller.filename, caller.lineno, sql, repr(e))
tdLog.notice("%s(%d) failed: sql:%s, %s" % args)
raise Exception(repr(e))
return self.affectedRows
def checkAffectedRows(self, expectAffectedRows):
if self.affectedRows != expectAffectedRows:
frame = inspect.stack()[1]
callerModule = inspect.getmodule(frame[0])
callerFilename = callerModule.__file__
tdLog.exit(
"%s failed: sql:%s, affectedRows:%d != expect:%d" %
(callerFilename, self.sql, self.affectedRows, expectAffectedRows))
tdLog.info("sql:%s, affectedRows:%d == expect:%d" %
(self.sql, self.affectedRows, expectAffectedRows))
caller = inspect.getframeinfo(inspect.stack()[1][0])
args = (caller.filename, caller.lineno, self.sql, self.affectedRows, expectAffectedRows)
tdLog.exit("%s(%d) failed: sql:%s, affectedRows:%d != expect:%d" % args)
tdLog.info("sql:%s, affectedRows:%d == expect:%d" % (self.sql, self.affectedRows, expectAffectedRows))
tdSql = TDSql()
......@@ -68,7 +68,7 @@ endi
if $data01 != 1 then
return -1
endi
if $data02 != 1 then
if $data02 != NULL then
return -1
endi
......@@ -80,7 +80,7 @@ endi
if $data01 != 1 then
return -1
endi
if $data02 != 1 then
if $data02 != NULL then
return -1
endi
......
......@@ -843,6 +843,14 @@ if $data81 != 4 then
return -1
endi
# desc fill query
print desc fill query
sql select count(*) from m_fl_tb0 where ts>='2018-9-17 9:0:0' and ts<='2018-9-17 9:11:00' interval(1m) fill(value,10) order by ts desc;
if $rows != 12 then
return -1
endi
print =============== clear
sql drop database $db
sql show databases
......
此差异已折叠。
......@@ -422,4 +422,63 @@ if $data97 != @group_tb0@ then
return -1
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT
\ No newline at end of file
#=========================== group by multi tags ======================
sql create table st (ts timestamp, c int) tags (t1 int, t2 int, t3 int, t4 int);
sql create table t1 using st tags(1, 1, 1, 1);
sql create table t2 using st tags(1, 2, 2, 2);
sql insert into t1 values ('2020-03-27 04:11:16.000', 1)('2020-03-27 04:11:17.000', 2) ('2020-03-27 04:11:18.000', 3) ('2020-03-27 04:11:19.000', 4) ;
sql insert into t1 values ('2020-03-27 04:21:16.000', 1)('2020-03-27 04:31:17.000', 2) ('2020-03-27 04:51:18.000', 3) ('2020-03-27 05:10:19.000', 4) ;
sql insert into t2 values ('2020-03-27 04:11:16.000', 1)('2020-03-27 04:11:17.000', 2) ('2020-03-27 04:11:18.000', 3) ('2020-03-27 04:11:19.000', 4) ;
sql insert into t2 values ('2020-03-27 04:21:16.000', 1)('2020-03-27 04:31:17.000', 2) ('2020-03-27 04:51:18.000', 3) ('2020-03-27 05:10:19.000', 4) ;
sql select irate(c) from st where t1="1" and ts >= '2020-03-27 04:11:17.732' and ts < '2020-03-27 05:11:17.732' interval(1m) sliding(15s) group by tbname,t1,t2;
if $rows != 40 then
return -1
endi
if $data01 != 1.000000000 then
return -1
endi
if $data02 != t1 then
return -1
endi
if $data03 != 1 then
return -1
endi
if $data04 != 1 then
return -1
endi
if $data11 != 1.000000000 then
return -1
endi
if $data12 != t1 then
return -1
endi
if $data13 != 1 then
return -1
endi
if $data14 != 1 then
return -1
endi
sql select irate(c) from st where t1="1" and ts >= '2020-03-27 04:11:17.732' and ts < '2020-03-27 05:11:17.732' interval(1m) sliding(15s) group by tbname,t1,t2 limit 1;
if $rows != 2 then
return -1
endi
if $data11 != 1.000000000 then
return -1
endi
if $data12 != t2 then
return -1
endi
if $data13 != 1 then
return -1
endi
if $data14 != 2 then
return -1
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT
......@@ -40,7 +40,7 @@ while $x < $rowNum
endw
print ====== tables created
sleep 60000
sleep 6000
$ts = $ts0 + $delta
$ts = $ts + 1
......
......@@ -39,7 +39,7 @@ while $x < $rowNum
endw
print ====== tables created
sleep 60000
sleep 6000
$ts = $ts0 + $delta
$ts = $ts + 1
......
......@@ -39,7 +39,7 @@ while $x < $rowNum
endw
print ====== tables created
sleep 60000
sleep 6000
$ts = $ts + 1
sql insert into $tb values ( $ts , -1, -1, -1, -1, -1)
......@@ -47,7 +47,7 @@ $ts = $ts0 + $delta
$ts = $ts + 1
sql import into $tb values ( $ts , -2, -2, -2, -2, -2)
sleep 60000
sleep 6000
sql show databases
......
......@@ -327,3 +327,7 @@ endi
if $data98 != 9 then
return -1
endi
#add one more test case
sql select max(c1), last(c8) from lm2_db0.lm2_tb0 where ts >= 1537146000000 and ts <= 1543145400000 interval(5m) fill(linear) limit 10 offset 4089;"
......@@ -636,6 +636,15 @@ if $data00 != $data01 then
return -1
endi
sql select first(ts), ts from select_tags_tb1
if $row != 1 then
return -1
endi
if $data01 != @70-01-01 08:01:50.001@ then
return -1
endi
print ======= selectivity + tags + group by + tags + filter + interval ================
sql select first(c1), t2, t1, tbname from select_tags_mt0 where c1<=2 interval(1d) group by tbname;
if $row != 3 then
......
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
run general/stream/stream_1.sim
run general/stream/stream_2.sim
run general/stream/stream_3.sim
run general/stream/stream_restart.sim
run general/stream/table_1.sim
run general/stream/metrics_1.sim
run general/stream/table_n.sim
run general/stream/metrics_n.sim
run general/stream/table_del.sim
......
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册