提交 96553e78 编写于 作者: S Shengliang Guan

Merge remote-tracking branch 'origin/develop' into feature/ad

......@@ -4,7 +4,7 @@ PROJECT(TDengine)
IF (DEFINED VERNUMBER)
SET(TD_VER_NUMBER ${VERNUMBER})
ELSE ()
SET(TD_VER_NUMBER "2.0.3.0")
SET(TD_VER_NUMBER "2.0.4.0")
ENDIF ()
IF (DEFINED VERCOMPATIBLE)
......
......@@ -99,6 +99,7 @@ typedef struct STableMeta {
uint8_t tableType;
int16_t sversion;
int16_t tversion;
char sTableId[TSDB_TABLE_FNAME_LEN];
SCMVgroupInfo vgroupInfo;
SCMCorVgroupInfo corVgroupInfo;
STableId id;
......
此差异已折叠。
......@@ -368,7 +368,40 @@ int32_t tscToSQLCmd(SSqlObj* pSql, struct SSqlInfo* pInfo) {
return tscGetTableMeta(pSql, pTableMetaInfo);
}
case TSDB_SQL_SHOW_CREATE_TABLE: {
SStrToken* pToken = &pInfo->pDCLInfo->a[0];
const char* msg1 = "invalid table name";
const char* msg2 = "table name is too long";
if (tscValidateName(pToken) != TSDB_CODE_SUCCESS) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg1);
}
if (!tscValidateTableNameLength(pToken->n)) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg2);
}
if (tscSetTableFullName(pTableMetaInfo, pToken, pSql) != TSDB_CODE_SUCCESS) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg2);
}
return tscGetTableMeta(pSql, pTableMetaInfo);
}
case TSDB_SQL_SHOW_CREATE_DATABASE: {
const char* msg1 = "invalid database name";
const char* msg2 = "table name is too long";
SStrToken* pToken = &pInfo->pDCLInfo->a[0];
if (tscValidateName(pToken) != TSDB_CODE_SUCCESS) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg1);
}
if (pToken->n > TSDB_DB_NAME_LEN) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg1);
}
if (tscSetTableFullName(pTableMetaInfo, pToken, pSql) != TSDB_CODE_SUCCESS) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg2);
}
return TSDB_CODE_SUCCESS;
}
case TSDB_SQL_CFG_DNODE: {
const char* msg2 = "invalid configure options or values, such as resetlog / debugFlag 135 / balance 'vnode:2-dnode:2' / monitor 1 ";
const char* msg3 = "invalid dnode ep";
......@@ -5722,7 +5755,7 @@ int32_t tscCheckCreateDbParams(SSqlCmd* pCmd, SCMCreateDbMsg* pCreate) {
char msg[512] = {0};
if (pCreate->walLevel != -1 && (pCreate->walLevel < TSDB_MIN_WAL_LEVEL || pCreate->walLevel > TSDB_MAX_WAL_LEVEL)) {
snprintf(msg, tListLen(msg), "invalid db option walLevel: %d, only 0-2 allowed", pCreate->walLevel);
snprintf(msg, tListLen(msg), "invalid db option walLevel: %d, only 1-2 allowed", pCreate->walLevel);
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg);
}
......
......@@ -170,6 +170,7 @@ STableMeta* tscCreateTableMetaFromMsg(STableMetaMsg* pTableMetaMsg, size_t* size
pTableMeta->sversion = pTableMetaMsg->sversion;
pTableMeta->tversion = pTableMetaMsg->tversion;
tstrncpy(pTableMeta->sTableId, pTableMetaMsg->sTableId, TSDB_TABLE_FNAME_LEN);
memcpy(pTableMeta->schema, pTableMetaMsg->schema, schemaSize);
......
......@@ -2093,6 +2093,9 @@ int tscProcessAlterDbMsgRsp(SSqlObj *pSql) {
UNUSED(pSql);
return 0;
}
int tscProcessShowCreateRsp(SSqlObj *pSql) {
return tscLocalResultCommonBuilder(pSql, 1);
}
int tscProcessQueryRsp(SSqlObj *pSql) {
SSqlRes *pRes = &pSql->res;
......@@ -2379,6 +2382,10 @@ void tscInitMsgsFp() {
tscProcessMsgRsp[TSDB_SQL_ALTER_TABLE] = tscProcessAlterTableMsgRsp;
tscProcessMsgRsp[TSDB_SQL_ALTER_DB] = tscProcessAlterDbMsgRsp;
tscProcessMsgRsp[TSDB_SQL_SHOW_CREATE_TABLE] = tscProcessShowCreateRsp;
tscProcessMsgRsp[TSDB_SQL_SHOW_CREATE_DATABASE] = tscProcessShowCreateRsp;
tscKeepConn[TSDB_SQL_SHOW] = 1;
tscKeepConn[TSDB_SQL_RETRIEVE] = 1;
tscKeepConn[TSDB_SQL_SELECT] = 1;
......
......@@ -264,12 +264,13 @@ void taos_close(TAOS *taos) {
}
SSqlObj* pHb = pObj->pHb;
if (pHb != NULL) {
if (pHb != NULL && atomic_val_compare_exchange_ptr(&pObj->pHb, pHb, 0) == pHb) {
if (pHb->pRpcCtx != NULL) { // wait for rsp from dnode
rpcCancelRequest(pHb->pRpcCtx);
pHb->pRpcCtx = NULL;
}
pObj->pHb = NULL;
tscDebug("%p, HB is freed", pHb);
taos_free_result(pHb);
}
......@@ -476,6 +477,8 @@ TAOS_ROW taos_fetch_row(TAOS_RES *res) {
pCmd->command == TSDB_SQL_TABLE_JOIN_RETRIEVE ||
pCmd->command == TSDB_SQL_FETCH ||
pCmd->command == TSDB_SQL_SHOW ||
pCmd->command == TSDB_SQL_SHOW_CREATE_TABLE ||
pCmd->command == TSDB_SQL_SHOW_CREATE_DATABASE ||
pCmd->command == TSDB_SQL_SELECT ||
pCmd->command == TSDB_SQL_DESCRIBE_TABLE ||
pCmd->command == TSDB_SQL_SERV_STATUS ||
......
......@@ -136,7 +136,6 @@ static void tscProcessStreamTimer(void *handle, void *tmrId) {
etime = pStream->stime + (etime - pStream->stime) / pStream->interval.interval * pStream->interval.interval;
} else {
etime = taosTimeTruncate(etime, &pStream->interval, pStream->precision);
//etime = taosGetIntervalStartTimestamp(etime, pStream->interval.sliding, pStream->interval.sliding, pStream->interval.slidingUnit, pStream->precision);
}
pQueryInfo->window.ekey = etime;
if (pQueryInfo->window.skey >= pQueryInfo->window.ekey) {
......@@ -454,17 +453,11 @@ static int64_t tscGetStreamStartTimestamp(SSqlObj *pSql, SSqlStream *pStream, in
}
} else { // timewindow based aggregation stream
if (stime == 0) { // no data in meter till now
stime = pQueryInfo->window.skey;
if (stime == INT64_MIN) {
stime = (int64_t)taosGetTimestamp(pStream->precision);
stime = taosTimeTruncate(stime, &pStream->interval, pStream->precision);
stime = taosTimeTruncate(stime - 1, &pStream->interval, pStream->precision);
//stime = taosGetIntervalStartTimestamp(stime, pStream->interval.interval, pStream->interval.interval, pStream->interval.intervalUnit, pStream->precision);
//stime = taosGetIntervalStartTimestamp(stime - 1, pStream->interval.interval, pStream->interval.interval, pStream->interval.intervalUnit, pStream->precision);
tscWarn("%p stream:%p, last timestamp:0, reset to:%" PRId64, pSql, pStream, stime);
if (pQueryInfo->window.skey != INT64_MIN) {
stime = pQueryInfo->window.skey;
}
stime = taosTimeTruncate(stime, &pStream->interval, pStream->precision);
} else {
//int64_t newStime = taosGetIntervalStartTimestamp(stime, pStream->interval.interval, pStream->interval.interval, pStream->interval.intervalUnit, pStream->precision);
int64_t newStime = taosTimeTruncate(stime, &pStream->interval, pStream->precision);
if (newStime != stime) {
tscWarn("%p stream:%p, last timestamp:%" PRId64 ", reset to:%" PRId64, pSql, pStream, stime, newStime);
......@@ -477,8 +470,10 @@ static int64_t tscGetStreamStartTimestamp(SSqlObj *pSql, SSqlStream *pStream, in
}
static int64_t tscGetLaunchTimestamp(const SSqlStream *pStream) {
int64_t timer = pStream->stime - taosGetTimestamp(pStream->precision);
if (timer < 0) timer = 0;
int64_t timer = 0, now = taosGetTimestamp(pStream->precision);
if (pStream->stime > now) {
timer = pStream->stime - now;
}
int64_t startDelay =
(pStream->precision == TSDB_TIME_PRECISION_MICRO) ? tsStreamCompStartDelay * 1000L : tsStreamCompStartDelay;
......@@ -572,6 +567,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
pStream->pSql = pSql;
pSql->pStream = pStream;
pSql->param = pStream;
pSql->maxRetry = TSDB_MAX_REPLICA;
pSql->sqlstr = calloc(1, strlen(sqlstr) + 1);
if (pSql->sqlstr == NULL) {
......@@ -579,6 +575,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
tscFreeSqlObj(pSql);
return NULL;
}
strtolower(pSql->sqlstr, sqlstr);
tscDebugL("%p SQL: %s", pSql, pSql->sqlstr);
......@@ -619,10 +616,9 @@ void taos_close_stream(TAOS_STREAM *handle) {
tscDebug("%p stream:%p is closed", pSql, pStream);
// notify CQ to release the pStream object
pStream->fp(pStream->param, NULL, NULL);
taos_free_result(pSql);
pStream->pSql = NULL;
taos_free_result(pSql);
taosTFree(pStream);
}
}
......@@ -23,6 +23,7 @@
#include "tscSubquery.h"
#include "tschemautil.h"
#include "tsclient.h"
#include "tscSubquery.h"
typedef struct SInsertSupporter {
SSubqueryState* pState;
......@@ -278,7 +279,7 @@ static int32_t tscLaunchRealSubqueries(SSqlObj* pSql) {
tscDebug("%p subIndex: %d, no need to launch query, ignore it", pSql, i);
tscDestroyJoinSupporter(pSupporter);
tscFreeSqlObj(pPrevSub);
taos_free_result(pPrevSub);
pSql->pSubs[i] = NULL;
continue;
......@@ -1383,7 +1384,7 @@ static void doCleanupSubqueries(SSqlObj *pSql, int32_t numOfSubs, SSubqueryState
taosTFree(pSupport->localBuffer);
taosTFree(pSupport);
tscFreeSqlObj(pSub);
taos_free_result(pSub);
}
free(pState);
......
......@@ -1743,8 +1743,6 @@ SSqlObj* createSimpleSubObj(SSqlObj* pSql, void (*fp)(), void* param, int32_t cm
}
pNew->pTscObj = pSql->pTscObj;
T_REF_INC(pNew->pTscObj);
pNew->signature = pNew;
SSqlCmd* pCmd = &pNew->cmd;
......@@ -1777,7 +1775,6 @@ SSqlObj* createSimpleSubObj(SSqlObj* pSql, void (*fp)(), void* param, int32_t cm
tscAddTableMetaInfo(pQueryInfo, pMasterTableMetaInfo->name, NULL, NULL, NULL);
T_REF_INC(pNew->pTscObj);
uint64_t p = (uint64_t) pNew;
pNew->self = taosCachePut(tscObjCache, &p, sizeof(uint64_t), &pNew, sizeof(uint64_t), 2 * 600 * 1000);
return pNew;
......
......@@ -78,6 +78,9 @@ enum {
TSDB_DEFINE_SQL_TYPE( TSDB_SQL_RETRIEVE_LOCALMERGE, "retrieve-localmerge" )
TSDB_DEFINE_SQL_TYPE( TSDB_SQL_TABLE_JOIN_RETRIEVE, "join-retrieve" )
TSDB_DEFINE_SQL_TYPE( TSDB_SQL_SHOW_CREATE_TABLE, "show-create-table")
TSDB_DEFINE_SQL_TYPE( TSDB_SQL_SHOW_CREATE_DATABASE, "show-create-database")
/*
* build empty result instead of accessing dnode to fetch result
* reset the client cache
......
......@@ -131,7 +131,7 @@ uint16_t tsHttpPort = 6041; // only tcp, range tcp[6041]
int32_t tsHttpCacheSessions = 1000;
int32_t tsHttpSessionExpire = 36000;
int32_t tsHttpMaxThreads = 2;
int32_t tsHttpEnableCompress = 0;
int32_t tsHttpEnableCompress = 1;
int32_t tsHttpEnableRecordSql = 0;
int32_t tsTelegrafUseFieldNum = 0;
......
Subproject commit 8c58c512b6acda8bcdfa48fdc7140227b5221766
Subproject commit 06ec30a0f1762e8169bf6b9045c82bcaa52bcdf0
......@@ -673,6 +673,7 @@ typedef struct {
typedef struct STableMetaMsg {
int32_t contLen;
char tableId[TSDB_TABLE_FNAME_LEN]; // table id
char sTableId[TSDB_TABLE_FNAME_LEN];
uint8_t numOfTags;
uint8_t precision;
uint8_t tableType;
......
......@@ -16,6 +16,7 @@
#ifndef TDENGINE_TTOKENDEF_H
#define TDENGINE_TTOKENDEF_H
#define TK_ID 1
#define TK_BOOL 2
#define TK_TINYINT 3
......@@ -75,24 +76,24 @@
#define TK_VNODES 57
#define TK_IPTOKEN 58
#define TK_DOT 59
#define TK_TABLES 60
#define TK_STABLES 61
#define TK_VGROUPS 62
#define TK_DROP 63
#define TK_TABLE 64
#define TK_DATABASE 65
#define TK_DNODE 66
#define TK_USER 67
#define TK_ACCOUNT 68
#define TK_USE 69
#define TK_DESCRIBE 70
#define TK_ALTER 71
#define TK_PASS 72
#define TK_PRIVILEGE 73
#define TK_LOCAL 74
#define TK_IF 75
#define TK_EXISTS 76
#define TK_CREATE 77
#define TK_CREATE 60
#define TK_TABLE 61
#define TK_DATABASE 62
#define TK_TABLES 63
#define TK_STABLES 64
#define TK_VGROUPS 65
#define TK_DROP 66
#define TK_DNODE 67
#define TK_USER 68
#define TK_ACCOUNT 69
#define TK_USE 70
#define TK_DESCRIBE 71
#define TK_ALTER 72
#define TK_PASS 73
#define TK_PRIVILEGE 74
#define TK_LOCAL 75
#define TK_IF 76
#define TK_EXISTS 77
#define TK_PPS 78
#define TK_TSERIES 79
#define TK_DBS 80
......@@ -222,7 +223,6 @@
#define TK_INTO 204
#define TK_VALUES 205
#define TK_SPACE 300
#define TK_COMMENT 301
#define TK_ILLEGAL 302
......
......@@ -44,6 +44,7 @@ typedef void* twalh; // WAL HANDLE
typedef int (*FWalWrite)(void *ahandle, void *pHead, int type);
twalh walOpen(const char *path, const SWalCfg *pCfg);
int walAlter(twalh pWal, const SWalCfg *pCfg);
void walClose(twalh);
int walRenew(twalh);
int walWrite(twalh, SWalHead *);
......
......@@ -910,13 +910,13 @@ static SDbCfg mnodeGetAlterDbOption(SDbObj *pDb, SCMAlterDbMsg *pAlter) {
}
if (walLevel > 0 && walLevel != pDb->cfg.walLevel) {
mError("db:%s, can't alter walLevel option", pDb->name);
terrno = TSDB_CODE_MND_INVALID_DB_OPTION;
mDebug("db:%s, walLevel:%d change to %d", pDb->name, pDb->cfg.walLevel, walLevel);
newCfg.walLevel = walLevel;
}
if (fsyncPeriod >= 0 && fsyncPeriod != pDb->cfg.fsyncPeriod) {
mError("db:%s, can't alter fsyncPeriod option", pDb->name);
terrno = TSDB_CODE_MND_INVALID_DB_OPTION;
mDebug("db:%s, fsyncPeriod:%d change to %d", pDb->name, pDb->cfg.fsyncPeriod, fsyncPeriod);
newCfg.fsyncPeriod = fsyncPeriod;
}
if (replications > 0 && replications != pDb->cfg.replications) {
......
......@@ -65,7 +65,7 @@ int32_t mnodeInitShow() {
mnodeAddReadMsgHandle(TSDB_MSG_TYPE_CM_CONNECT, mnodeProcessConnectMsg);
mnodeAddReadMsgHandle(TSDB_MSG_TYPE_CM_USE_DB, mnodeProcessUseMsg);
tsMnodeShowCache = taosCacheInit(TSDB_DATA_TYPE_BIGINT, 5, false, mnodeFreeShowObj, "show");
tsMnodeShowCache = taosCacheInit(TSDB_DATA_TYPE_BIGINT, 5, true, mnodeFreeShowObj, "show");
return 0;
}
......@@ -389,10 +389,12 @@ static bool mnodeAccquireShowObj(SShowObj *pShow) {
}
static void* mnodePutShowObj(SShowObj *pShow) {
const int32_t DEFAULT_SHOWHANDLE_LIFE_SPAN = tsShellActivityTimer * 6 * 1000;
if (tsMnodeShowCache != NULL) {
pShow->index = atomic_add_fetch_32(&tsShowObjIndex, 1);
uint64_t handleVal = (uint64_t)pShow;
SShowObj **ppShow = taosCachePut(tsMnodeShowCache, &handleVal, sizeof(int64_t), &pShow, sizeof(int64_t), 6000);
SShowObj **ppShow = taosCachePut(tsMnodeShowCache, &handleVal, sizeof(int64_t), &pShow, sizeof(int64_t), DEFAULT_SHOWHANDLE_LIFE_SPAN);
pShow->ppShow = (void**)ppShow;
mDebug("%p, show is put into cache, data:%p index:%d", pShow, ppShow, pShow->index);
return pShow;
......
......@@ -1384,6 +1384,9 @@ int32_t mnodeRetrieveShowSuperTables(SShowObj *pShow, char *data, int32_t rows,
}
pShow->numOfReads += numOfRows;
const int32_t NUM_OF_COLUMNS = 5;
mnodeVacuumResult(data, NUM_OF_COLUMNS, numOfRows, rows, pShow);
mnodeDecDbRef(pDb);
return numOfRows;
......@@ -2090,6 +2093,9 @@ static int32_t mnodeDoGetChildTableMeta(SMnodeMsg *pMsg, STableMetaMsg *pMeta) {
pMeta->precision = pDb->cfg.precision;
pMeta->tableType = pTable->info.type;
tstrncpy(pMeta->tableId, pTable->info.tableId, TSDB_TABLE_FNAME_LEN);
if (pTable->superTable) {
tstrncpy(pMeta->sTableId, pTable->superTable->info.tableId, TSDB_TABLE_FNAME_LEN);
}
if (pTable->info.type == TSDB_CHILD_TABLE) {
pMeta->sversion = htons(pTable->superTable->sversion);
......
......@@ -19,7 +19,7 @@
#include <stdint.h>
#include <stdbool.h>
#define JSON_BUFFER_SIZE 10240
#define JSON_BUFFER_SIZE 16384
struct HttpContext;
enum { JsonNumber, JsonString, JsonBoolean, JsonArray, JsonObject, JsonNull };
......
......@@ -52,12 +52,12 @@ int32_t httpWriteBufByFd(struct HttpContext* pContext, const char* buf, int32_t
}
if (len < 0) {
httpDebug("context:%p, fd:%d, socket write errno:%d, times:%d", pContext, pContext->fd, errno, countWait);
httpDebug("context:%p, fd:%d, socket write errno:%d:%s, times:%d", pContext, pContext->fd, errno, strerror(errno), countWait);
if (++countWait > HTTP_WRITE_RETRY_TIMES) break;
taosMsleep(HTTP_WRITE_WAIT_TIME_MS);
continue;
} else if (len == 0) {
httpDebug("context:%p, fd:%d, socket write errno:%d, connect already closed", pContext, pContext->fd, errno);
httpDebug("context:%p, fd:%d, socket write errno:%d:%s, connect already closed", pContext, pContext->fd, errno, strerror(errno));
break;
} else {
countWait = 0;
......@@ -131,14 +131,14 @@ int32_t httpWriteJsonBufBody(JsonBuf* buf, bool isTheLast) {
httpWriteBufNoTrace(buf->pContext, sLen, len);
remain = httpWriteBufNoTrace(buf->pContext, (const char*)compressBuf, compressBufLen);
} else {
httpTrace("context:%p, fd:%d, last:%d, compress already dumped, response:\n%s", buf->pContext,
httpDebug("context:%p, fd:%d, last:%d, compress already dumped, response:\n%s", buf->pContext,
buf->pContext->fd, isTheLast, buf->buf);
return 0; // there is no data to dump.
remain = 0; // there is no data to dump.
}
} else {
httpError("context:%p, fd:%d, failed to compress data, chunkSize:%" PRIu64 ", last:%d, error:%d, response:\n%s",
buf->pContext, buf->pContext->fd, srcLen, isTheLast, ret, buf->buf);
return 0;
remain = 0;
}
}
......
......@@ -406,15 +406,21 @@ int32_t httpGzipCompressInit(HttpContext *pContext) {
int32_t httpGzipCompress(HttpContext *pContext, char *srcData, int32_t nSrcData, char *destData, int32_t *nDestData, bool isTheLast) {
int32_t err = 0;
int32_t lastTotalLen = (int32_t) (pContext->gzipStream.total_out);
pContext->gzipStream.next_in = (Bytef *) srcData;
pContext->gzipStream.avail_in = (uLong) nSrcData;
pContext->gzipStream.next_out = (Bytef *) destData;
pContext->gzipStream.avail_out = (uLong) (*nDestData);
while (pContext->gzipStream.avail_in != 0 && pContext->gzipStream.total_out < (uLong) (*nDestData)) {
while (pContext->gzipStream.avail_in != 0) {
if (deflate(&pContext->gzipStream, Z_FULL_FLUSH) != Z_OK) {
return -1;
}
int32_t cacheLen = pContext->gzipStream.total_out - lastTotalLen;
if (cacheLen >= *nDestData) {
return -2;
}
}
if (pContext->gzipStream.avail_in != 0) {
......@@ -427,16 +433,16 @@ int32_t httpGzipCompress(HttpContext *pContext, char *srcData, int32_t nSrcData,
break;
}
if (err != Z_OK) {
return -2;
return -3;
}
}
if (deflateEnd(&pContext->gzipStream) != Z_OK) {
return -3;
return -4;
}
}
*nDestData = (int32_t) (pContext->gzipStream.total_out);
*nDestData = (int32_t) (pContext->gzipStream.total_out) - lastTotalLen;
return 0;
}
......
......@@ -80,6 +80,7 @@ cmd ::= SHOW GRANTS. { setShowOptions(pInfo, TSDB_MGMT_TABLE_GRANTS, 0, 0);
cmd ::= SHOW VNODES. { setShowOptions(pInfo, TSDB_MGMT_TABLE_VNODES, 0, 0); }
cmd ::= SHOW VNODES IPTOKEN(X). { setShowOptions(pInfo, TSDB_MGMT_TABLE_VNODES, &X, 0); }
%type dbPrefix {SStrToken}
dbPrefix(A) ::=. {A.n = 0; A.type = 0;}
dbPrefix(A) ::= ids(X) DOT. {A = X; }
......@@ -88,6 +89,15 @@ dbPrefix(A) ::= ids(X) DOT. {A = X; }
cpxName(A) ::= . {A.n = 0; }
cpxName(A) ::= DOT ids(Y). {A = Y; A.n += 1; }
cmd ::= SHOW CREATE TABLE ids(X) cpxName(Y). {
X.n += Y.n;
setDCLSQLElems(pInfo, TSDB_SQL_SHOW_CREATE_TABLE, 1, &X);
}
cmd ::= SHOW CREATE DATABASE ids(X). {
setDCLSQLElems(pInfo, TSDB_SQL_SHOW_CREATE_DATABASE, 1, &X);
}
cmd ::= SHOW dbPrefix(X) TABLES. {
setShowOptions(pInfo, TSDB_MGMT_TABLE_TABLE, &X, 0);
}
......
此差异已折叠。
......@@ -266,7 +266,6 @@ void *taosCacheAcquireByKey(SCacheObj *pCacheObj, const void *key, size_t keyLen
if (taosHashGetSize(pCacheObj->pHashTable) == 0) {
atomic_add_fetch_32(&pCacheObj->statistics.missCount, 1);
uError("cache:%s, key:%p, not in cache, retrieved failed, reason: empty sqlObj cache", pCacheObj->name, key);
return NULL;
}
......@@ -280,7 +279,7 @@ void *taosCacheAcquireByKey(SCacheObj *pCacheObj, const void *key, size_t keyLen
uDebug("cache:%s, key:%p, %p is retrieved from cache, refcnt:%d", pCacheObj->name, key, pData, T_REF_VAL_GET(ptNode));
} else {
atomic_add_fetch_32(&pCacheObj->statistics.missCount, 1);
uError("cache:%s, key:%p, not in cache, retrieved failed", pCacheObj->name, key);
uDebug("cache:%s, key:%p, not in cache, retrieved failed", pCacheObj->name, key);
}
atomic_add_fetch_32(&pCacheObj->statistics.totalAccess, 1);
......
......@@ -186,6 +186,12 @@ int32_t vnodeAlter(void *param, SMDCreateVnodeMsg *pVnodeCfg) {
return code;
}
code = walAlter(pVnode->wal, &pVnode->walCfg);
if (code != TSDB_CODE_SUCCESS) {
pVnode->status = TAOS_VN_STATUS_READY;
return code;
}
code = syncReconfig(pVnode->sync, &pVnode->syncCfg);
if (code != TSDB_CODE_SUCCESS) {
pVnode->status = TAOS_VN_STATUS_READY;
......@@ -390,6 +396,7 @@ void vnodeRelease(void *pVnodeRaw) {
if (0 == tsEnableVnodeBak) {
vInfo("vgId:%d, vnode backup not enabled", pVnode->vgId);
} else {
taosRemoveDir(newDir);
taosRename(rootDir, newDir);
}
......
......@@ -58,7 +58,8 @@ int32_t vnodeProcessRead(void *param, SReadMsg *pReadMsg) {
return TSDB_CODE_APP_NOT_READY;
// TODO: Later, let slave to support query
if (pVnode->syncCfg.replica > 1 && pVnode->role != TAOS_SYNC_ROLE_MASTER) {
// if (pVnode->syncCfg.replica > 1 && pVnode->role != TAOS_SYNC_ROLE_MASTER) {
if (pVnode->role != TAOS_SYNC_ROLE_SLAVE && pVnode->role != TAOS_SYNC_ROLE_MASTER) {
vDebug("vgId:%d, msgType:%s not processed, replica:%d role:%d", pVnode->vgId, taosMsg[msgType], pVnode->syncCfg.replica, pVnode->role);
return TSDB_CODE_APP_NOT_READY;
}
......
......@@ -69,6 +69,13 @@ static void walModuleInitFunc() {
wDebug("WAL module is initialized");
}
static inline bool walNeedFsyncTimer(SWal *pWal) {
if (pWal->fsyncPeriod > 0 && pWal->level == TAOS_WAL_FSYNC) {
return true;
}
return false;
}
void *walOpen(const char *path, const SWalCfg *pCfg) {
SWal *pWal = calloc(sizeof(SWal), 1);
if (pWal == NULL) {
......@@ -95,7 +102,7 @@ void *walOpen(const char *path, const SWalCfg *pCfg) {
tstrncpy(pWal->path, path, sizeof(pWal->path));
pthread_mutex_init(&pWal->mutex, NULL);
if (pWal->fsyncPeriod > 0 && pWal->level == TAOS_WAL_FSYNC) {
if (walNeedFsyncTimer(pWal)) {
pWal->timer = taosTmrStart(walProcessFsyncTimer, pWal->fsyncPeriod, pWal, walTmrCtrl);
if (pWal->timer == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
......@@ -127,6 +134,37 @@ void *walOpen(const char *path, const SWalCfg *pCfg) {
return pWal;
}
int walAlter(twalh wal, const SWalCfg *pCfg) {
SWal *pWal = wal;
if (pWal == NULL) {
return TSDB_CODE_WAL_APP_ERROR;
}
if (pWal->level == pCfg->walLevel && pWal->fsyncPeriod == pCfg->fsyncPeriod) {
wDebug("wal:%s, old walLevel:%d fsync:%d, new walLevel:%d fsync:%d not change", pWal->name, pWal->level,
pWal->fsyncPeriod, pCfg->walLevel, pCfg->fsyncPeriod);
return TSDB_CODE_SUCCESS;
}
wInfo("wal:%s, change old walLevel:%d fsync:%d, new walLevel:%d fsync:%d", pWal->name, pWal->level, pWal->fsyncPeriod,
pCfg->walLevel, pCfg->fsyncPeriod);
pthread_mutex_lock(&pWal->mutex);
pWal->level = pCfg->walLevel;
pWal->fsyncPeriod = pCfg->fsyncPeriod;
if (walNeedFsyncTimer(pWal)) {
wInfo("wal:%s, reset fsync timer, walLevel:%d fsyncPeriod:%d", pWal->name, pWal->level, pWal->fsyncPeriod);
taosTmrReset(walProcessFsyncTimer, pWal->fsyncPeriod, pWal, &pWal->timer,walTmrCtrl);
} else {
wInfo("wal:%s, stop fsync timer, walLevel:%d fsyncPeriod:%d", pWal->name, pWal->level, pWal->fsyncPeriod);
taosTmrStop(pWal->timer);
pWal->timer = NULL;
}
pthread_mutex_unlock(&pWal->mutex);
return TSDB_CODE_SUCCESS;
}
void walClose(void *handle) {
if (handle == NULL) return;
......@@ -484,6 +522,12 @@ static void walProcessFsyncTimer(void *param, void *tmrId) {
if (fsync(pWal->fd) < 0) {
wError("wal:%s, fsync failed(%s)", pWal->name, strerror(errno));
}
pWal->timer = taosTmrStart(walProcessFsyncTimer, pWal->fsyncPeriod, pWal, walTmrCtrl);
if (walNeedFsyncTimer(pWal)) {
pWal->timer = taosTmrStart(walProcessFsyncTimer, pWal->fsyncPeriod, pWal, walTmrCtrl);
} else {
wInfo("wal:%s, stop fsync timer for walLevel:%d fsyncPeriod:%d", pWal->name, pWal->level, pWal->fsyncPeriod);
taosTmrStop(pWal->timer);
pWal->timer = NULL;
}
}
......@@ -53,7 +53,7 @@ function buildTDengine {
function runGeneralCaseOneByOne {
while read -r line; do
if [[ $line =~ ^./test.sh* ]]; then
case=`echo $line | grep -w "general\|unique\/mnode\/mgmt33.sim\|unique\/stable\/dnode3.sim\|unique\/cluster\/balance3.sim\|unique\/arbitrator\/offline_replica2_alterTable_online.sim"|awk '{print $NF}'`
case=`echo $line | grep sim$ |awk '{print $NF}'`
if [ -n "$case" ]; then
./test.sh -f $case > /dev/null 2>&1 && \
......
......@@ -16,6 +16,7 @@ python3 ./test.py -f insert/nchar.py
python3 ./test.py -f insert/nchar-unicode.py
python3 ./test.py -f insert/multi.py
python3 ./test.py -f insert/randomNullCommit.py
python3 insert/retentionpolicy.py
python3 ./test.py -f table/column_name.py
python3 ./test.py -f table/column_num.py
......@@ -154,6 +155,7 @@ 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
python3 ./test.py -f stream/history.py
#alter table
python3 ./test.py -f alter/alter_table_crash.py
......@@ -192,3 +194,8 @@ python3 test.py -f query/queryInterval.py
# tools
python3 test.py -f tools/taosdemo.py
# subscribe
python3 test.py -f subscribe/singlemeter.py
python3 test.py -f subscribe/stability.py
python3 test.py -f subscribe/supertable.py
\ No newline at end of file
###################################################################
# 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 datetime
sys.path.insert(0, os.getcwd())
import taos
from util.log import *
from util.cases import *
from util.sql import *
from util.dnodes import *
class TDTestRetetion:
def init(self):
self.queryRows=0
tdLog.debug("start to execute %s" % __file__)
tdLog.info("prepare cluster")
tdDnodes.init("")
tdDnodes.setTestCluster(False)
tdDnodes.setValgrind(False)
tdDnodes.stopAll()
tdDnodes.deploy(1)
tdDnodes.start(1)
print(tdDnodes.getDnodesRootDir())
self.conn = taos.connect(config=tdDnodes.getSimCfgPath())
tdSql.init(self.conn.cursor())
tdSql.execute('reset query cache')
def checkRows(self, expectRows,sql):
if self.queryRows == expectRows:
tdLog.info("sql:%s, queryRows:%d == expect:%d" % (sql, self.queryRows, expectRows))
else:
caller = inspect.getframeinfo(inspect.stack()[1][0])
args = (caller.filename, caller.lineno, sql, self.queryRows, expectRows)
os.system("timedatectl set-ntp true")
tdLog.exit("%s(%d) failed: sql:%s, queryRows:%d != expect:%d" % args)
def run(self):
tdLog.info("=============== step1")
tdSql.execute('create database test keep 3 days 1;')
tdSql.execute('use test;')
tdSql.execute('create table test(ts timestamp,i int);')
cmd = 'insert into test values(now-2d,11)(now-1d,11)(now,11)(now+1d,11);'
tdLog.info(cmd)
tdSql.execute(cmd)
tdSql.query('select * from test')
tdSql.checkRows(4)
tdLog.info("=============== step2")
tdDnodes.stop(1)
os.system("timedatectl set-ntp false")
os.system("date -s $(date -d \"${DATE} 2 days\" \"+%Y%m%d\")")
tdDnodes.start(1)
cmd = 'insert into test values(now,11);'
tdLog.info(cmd)
tdSql.execute(cmd)
queryRows=tdSql.query('select * from test')
if queryRows==4:
tdSql.checkRows(4)
return 0
else:
tdSql.checkRows(5)
tdLog.info("=============== step3")
tdDnodes.stop(1)
os.system("date -s $(date -d \"${DATE} 2 days\" \"+%Y%m%d\")")
tdDnodes.start(1)
cmd = 'insert into test values(now-1d,11);'
tdLog.info(cmd)
tdSql.execute(cmd)
queryRows=tdSql.query('select * from test')
tdSql.checkRows(6)
tdLog.info("=============== step4")
tdDnodes.stop(1)
tdDnodes.start(1)
cmd = 'insert into test values(now,11);'
tdLog.info(cmd)
tdSql.execute(cmd)
tdSql.query('select * from test')
tdSql.checkRows(7)
tdLog.info("=============== step5")
tdDnodes.stop(1)
tdDnodes.start(1)
cmd='select * from test where ts > now-1d'
queryRows=tdSql.query('select * from test where ts > now-1d')
self.checkRows(1,cmd)
def stop(self):
os.system("timedatectl set-ntp true")
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
clients = TDTestRetetion()
clients.init()
clients.run()
clients.stop()
此差异已折叠。
......@@ -220,3 +220,5 @@ run general/stream/table_del.sim
run general/stream/metrics_del.sim
run general/stream/table_replica1_vnoden.sim
run general/stream/metrics_replica1_vnoden.sim
run general/db/show_create_db.sim
run general/db/show_create_table.sim
此差异已折叠。
此差异已折叠。
此差异已折叠。
......@@ -79,6 +79,7 @@ cd ../../../debug; make
./test.sh -f general/http/autocreate.sim
./test.sh -f general/http/chunked.sim
./test.sh -f general/http/gzip.sim
./test.sh -f general/http/restful.sim
./test.sh -f general/http/restful_insert.sim
./test.sh -f general/http/restful_limit.sim
......
此差异已折叠。
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册