Merge branch 'develop' into feature/2.0tsdb
This commit is contained in:
Hongze Cheng
commit
4f00e37d95
|
|
@ -47,6 +47,8 @@ Raw DataSize = numOfTables * rowSizePerTable * rowsPerTable
|
|||
|
||||
因为TDengine具有很好的水平扩展能力，根据总量，再根据单个物理机或虚拟机的资源，就可以轻松决定需要购置多少台物理机或虚拟机了。
|
||||
|
||||
具体计算公式，请参见页面：<a href='https://www.taosdata.com/config/config.html'>资源估算方法</a>
|
||||
|
||||
## 容错和灾备
|
||||
|
||||
### 容错
|
||||
|
|
|
|||
|
|
@ -162,7 +162,7 @@ Master Vnode遵循下面的写入流程：
|
|||
|
||||
<center> 图 3 TDengine Master写入流程 </center>
|
||||
1. Master vnode收到应用的数据插入请求，验证OK，进入下一步；
|
||||
2. 如果系统配置参数walLevel打开（设置为2），vnode将把该请求的原始数据包写入数据库日志文件WAL，以保证TDengine能够在断电等因素导致的服务重启时从数据库日志文件中恢复数据，避免数据的丢失；
|
||||
2. 如果系统配置参数walLevel大于0，vnode将把该请求的原始数据包写入数据库日志文件WAL。如果walLevel设置为2，而且fsync设置为0，TDengine还将WAL数据立即落盘，以保证即使宕机，也能从数据库日志文件中恢复数据，避免数据的丢失；
|
||||
3. 如果有多个副本，vnode将把数据包转发给同一虚拟节点组内slave vnodes, 该转发包带有数据的版本号(version)；
|
||||
4. 写入内存，并加记录加入到skip list；
|
||||
5. Master vnode返回确认信息给应用，表示写入成功。
|
||||
|
|
@ -174,7 +174,7 @@ Master Vnode遵循下面的写入流程：
|
|||
|
||||
<center> 图 4 TDengine Slave写入流程 </center>
|
||||
1. Slave vnode收到Master vnode转发了的数据插入请求。
|
||||
2. 如果系统配置参数walLevl设置为2，vnode将把该请求的原始数据包写入日志(WAL)；
|
||||
2. 如果系统配置参数walLevel大于0，vnode将把该请求的原始数据包写入数据库日志文件WAL。如果walLevel设置为2，而且fsync设置为0，TDengine还将WAL数据立即落盘，以保证即使宕机，也能从数据库日志文件中恢复数据，避免数据的丢失；
|
||||
3. 写入内存，更新内存中的skip list。
|
||||
|
||||
与Master vnode相比，slave vnode不存在转发环节，也不存在回复确认环节，少了两步。但写内存与WAL是完全一样的。
|
||||
|
|
|
|||
|
|
@ -221,20 +221,18 @@ typedef struct STableDataBlocks {
|
|||
SParamInfo *params;
|
||||
} STableDataBlocks;
|
||||
|
||||
//typedef struct SDataBlockList { // todo remove
|
||||
// uint32_t nSize;
|
||||
// uint32_t nAlloc;
|
||||
// STableDataBlocks **pData;
|
||||
//} SDataBlockList;
|
||||
|
||||
typedef struct SQueryInfo {
|
||||
int16_t command; // the command may be different for each subclause, so keep it seperately.
|
||||
uint32_t type; // query/insert type
|
||||
// TODO refactor
|
||||
char intervalTimeUnit;
|
||||
char slidingTimeUnit;
|
||||
uint32_t type; // query/insert type
|
||||
STimeWindow window; // query time window
|
||||
int64_t intervalTime; // aggregation time interval
|
||||
int64_t intervalTime; // aggregation time window range
|
||||
int64_t slidingTime; // sliding window in mseconds
|
||||
int64_t intervalOffset;// start offset of each time window
|
||||
int32_t tz; // query client timezone
|
||||
|
||||
SSqlGroupbyExpr groupbyExpr; // group by tags info
|
||||
SArray * colList; // SArray<SColumn*>
|
||||
SFieldInfo fieldsInfo;
|
||||
|
|
@ -401,7 +399,7 @@ int tsParseSql(SSqlObj *pSql, bool initial);
|
|||
void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet);
|
||||
int tscProcessSql(SSqlObj *pSql);
|
||||
|
||||
int tscRenewTableMeta(SSqlObj *pSql, char *tableId);
|
||||
int tscRenewTableMeta(SSqlObj *pSql, int32_t tableIndex);
|
||||
void tscQueueAsyncRes(SSqlObj *pSql);
|
||||
|
||||
void tscQueueAsyncError(void(*fp), void *param, int32_t code);
|
||||
|
|
@ -416,7 +414,7 @@ void tscRestoreSQLFuncForSTableQuery(SQueryInfo *pQueryInfo);
|
|||
int32_t tscCreateResPointerInfo(SSqlRes *pRes, SQueryInfo *pQueryInfo);
|
||||
void tscDestroyResPointerInfo(SSqlRes *pRes);
|
||||
|
||||
void tscResetSqlCmdObj(SSqlCmd *pCmd);
|
||||
void tscResetSqlCmdObj(SSqlCmd *pCmd, bool removeFromCache);
|
||||
|
||||
/**
|
||||
* free query result of the sql object
|
||||
|
|
|
|||
|
|
@ -468,7 +468,7 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
|
|||
if (pCmd->command == TSDB_SQL_INSERT || pCmd->command == TSDB_SQL_SELECT) {
|
||||
tscDebug("%p redo parse sql string and proceed", pSql);
|
||||
pCmd->parseFinished = false;
|
||||
tscResetSqlCmdObj(pCmd);
|
||||
tscResetSqlCmdObj(pCmd, false);
|
||||
|
||||
code = tsParseSql(pSql, true);
|
||||
|
||||
|
|
|
|||
|
|
@ -1327,18 +1327,40 @@ int tsParseSql(SSqlObj *pSql, bool initial) {
|
|||
pSql->fetchFp = pSql->fp;
|
||||
pSql->fp = (void(*)())tscHandleMultivnodeInsert;
|
||||
}
|
||||
|
||||
|
||||
if (initial && ((ret = tsInsertInitialCheck(pSql)) != TSDB_CODE_SUCCESS)) {
|
||||
return ret;
|
||||
}
|
||||
|
||||
|
||||
// make a backup as tsParseInsertSql may modify the string
|
||||
char* sqlstr = strdup(pSql->sqlstr);
|
||||
ret = tsParseInsertSql(pSql);
|
||||
if (sqlstr == NULL || pSql->retry >= 1 || ret != TSDB_CODE_TSC_INVALID_SQL) {
|
||||
free(sqlstr);
|
||||
} else {
|
||||
tscResetSqlCmdObj(pCmd, true);
|
||||
free(pSql->sqlstr);
|
||||
pSql->sqlstr = sqlstr;
|
||||
pSql->retry++;
|
||||
if ((ret = tsInsertInitialCheck(pSql)) == TSDB_CODE_SUCCESS) {
|
||||
ret = tsParseInsertSql(pSql);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
SSqlInfo SQLInfo = qSQLParse(pSql->sqlstr);
|
||||
ret = tscToSQLCmd(pSql, &SQLInfo);
|
||||
if (ret == TSDB_CODE_TSC_INVALID_SQL && pSql->retry == 0 && SQLInfo.type == TSDB_SQL_NULL) {
|
||||
tscResetSqlCmdObj(pCmd, true);
|
||||
pSql->retry++;
|
||||
ret = tscToSQLCmd(pSql, &SQLInfo);
|
||||
}
|
||||
SQLInfoDestroy(&SQLInfo);
|
||||
}
|
||||
|
||||
if (ret == TSDB_CODE_SUCCESS) {
|
||||
pSql->retry = 0;
|
||||
}
|
||||
|
||||
/*
|
||||
* the pRes->code may be modified or released by another thread in tscTableMetaCallBack function,
|
||||
* so do NOT use pRes->code to determine if the getTableMeta function
|
||||
|
|
|
|||
|
|
@ -276,8 +276,6 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
|
|||
}
|
||||
}
|
||||
|
||||
STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
|
||||
|
||||
int32_t cmd = pCmd->command;
|
||||
if ((cmd == TSDB_SQL_SELECT || cmd == TSDB_SQL_FETCH || cmd == TSDB_SQL_INSERT || cmd == TSDB_SQL_UPDATE_TAGS_VAL) &&
|
||||
(rpcMsg->code == TSDB_CODE_TDB_INVALID_TABLE_ID ||
|
||||
|
|
@ -302,7 +300,7 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
|
|||
taosMsleep(duration);
|
||||
}
|
||||
|
||||
rpcMsg->code = tscRenewTableMeta(pSql, pTableMetaInfo->name);
|
||||
rpcMsg->code = tscRenewTableMeta(pSql, 0);
|
||||
|
||||
// if there is an error occurring, proceed to the following error handling procedure.
|
||||
if (rpcMsg->code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
|
||||
|
|
@ -2202,14 +2200,14 @@ int tscGetMeterMetaEx(SSqlObj *pSql, STableMetaInfo *pTableMetaInfo, bool create
|
|||
/**
|
||||
* retrieve table meta from mnode, and update the local table meta cache.
|
||||
* @param pSql sql object
|
||||
* @param tableId table full name
|
||||
* @param tableIndex table index
|
||||
* @return status code
|
||||
*/
|
||||
int tscRenewTableMeta(SSqlObj *pSql, char *tableId) {
|
||||
int tscRenewTableMeta(SSqlObj *pSql, int32_t tableIndex) {
|
||||
SSqlCmd *pCmd = &pSql->cmd;
|
||||
|
||||
SQueryInfo * pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
|
||||
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
|
||||
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, tableIndex);
|
||||
|
||||
STableMeta* pTableMeta = pTableMetaInfo->pTableMeta;
|
||||
if (pTableMetaInfo->pTableMeta) {
|
||||
|
|
|
|||
|
|
@ -820,7 +820,7 @@ int taos_validate_sql(TAOS *taos, const char *sql) {
|
|||
|
||||
static int tscParseTblNameList(SSqlObj *pSql, const char *tblNameList, int32_t tblListLen) {
|
||||
// must before clean the sqlcmd object
|
||||
tscResetSqlCmdObj(&pSql->cmd);
|
||||
tscResetSqlCmdObj(&pSql->cmd, false);
|
||||
|
||||
SSqlCmd *pCmd = &pSql->cmd;
|
||||
|
||||
|
|
|
|||
|
|
@ -33,7 +33,7 @@
|
|||
static void freeQueryInfoImpl(SQueryInfo* pQueryInfo);
|
||||
static void clearAllTableMetaInfo(SQueryInfo* pQueryInfo, const char* address, bool removeFromCache);
|
||||
|
||||
SCond* tsGetSTableQueryCond(STagCond* pTagCond, uint64_t uid) {
|
||||
SCond* tsGetSTableQueryCond(STagCond* pTagCond, uint64_t uid) {
|
||||
if (pTagCond->pCond == NULL) {
|
||||
return NULL;
|
||||
}
|
||||
|
|
@ -294,7 +294,7 @@ void tscDestroyResPointerInfo(SSqlRes* pRes) {
|
|||
pRes->data = NULL; // pRes->data points to the buffer of pRsp, no need to free
|
||||
}
|
||||
|
||||
static void tscFreeQueryInfo(SSqlCmd* pCmd) {
|
||||
static void tscFreeQueryInfo(SSqlCmd* pCmd, bool removeFromCache) {
|
||||
if (pCmd == NULL || pCmd->numOfClause == 0) {
|
||||
return;
|
||||
}
|
||||
|
|
@ -304,7 +304,7 @@ static void tscFreeQueryInfo(SSqlCmd* pCmd) {
|
|||
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, i);
|
||||
|
||||
freeQueryInfoImpl(pQueryInfo);
|
||||
clearAllTableMetaInfo(pQueryInfo, (const char*)addr, false);
|
||||
clearAllTableMetaInfo(pQueryInfo, (const char*)addr, removeFromCache);
|
||||
taosTFree(pQueryInfo);
|
||||
}
|
||||
|
||||
|
|
@ -312,7 +312,7 @@ static void tscFreeQueryInfo(SSqlCmd* pCmd) {
|
|||
taosTFree(pCmd->pQueryInfo);
|
||||
}
|
||||
|
||||
void tscResetSqlCmdObj(SSqlCmd* pCmd) {
|
||||
void tscResetSqlCmdObj(SSqlCmd* pCmd, bool removeFromCache) {
|
||||
pCmd->command = 0;
|
||||
pCmd->numOfCols = 0;
|
||||
pCmd->count = 0;
|
||||
|
|
@ -326,7 +326,7 @@ void tscResetSqlCmdObj(SSqlCmd* pCmd) {
|
|||
|
||||
pCmd->pDataBlocks = tscDestroyBlockArrayList(pCmd->pDataBlocks);
|
||||
|
||||
tscFreeQueryInfo(pCmd);
|
||||
tscFreeQueryInfo(pCmd, removeFromCache);
|
||||
}
|
||||
|
||||
void tscFreeSqlResult(SSqlObj* pSql) {
|
||||
|
|
@ -364,7 +364,7 @@ void tscPartiallyFreeSqlObj(SSqlObj* pSql) {
|
|||
taosTFree(pSql->pSubs);
|
||||
pSql->numOfSubs = 0;
|
||||
|
||||
tscResetSqlCmdObj(pCmd);
|
||||
tscResetSqlCmdObj(pCmd, false);
|
||||
}
|
||||
|
||||
void tscFreeSqlObj(SSqlObj* pSql) {
|
||||
|
|
|
|||
|
|
@ -69,6 +69,15 @@ extern "C" {
|
|||
#define TSDB_FUNC_AVG_IRATE 33
|
||||
|
||||
#define TSDB_FUNC_TID_TAG 34
|
||||
#define TSDB_FUNC_HISTOGRAM 35
|
||||
#define TSDB_FUNC_HLL 36
|
||||
#define TSDB_FUNC_MODE 37
|
||||
#define TSDB_FUNC_SAMPLE 38
|
||||
#define TSDB_FUNC_CEIL 39
|
||||
#define TSDB_FUNC_FLOOR 40
|
||||
#define TSDB_FUNC_ROUND 41
|
||||
#define TSDB_FUNC_MAVG 42
|
||||
#define TSDB_FUNC_CSUM 43
|
||||
|
||||
#define TSDB_FUNCSTATE_SO 0x1u // single output
|
||||
#define TSDB_FUNCSTATE_MO 0x2u // dynamic number of output, not multinumber of output e.g., TOP/BOTTOM
|
||||
|
|
|
|||
|
|
@ -35,8 +35,6 @@
|
|||
* forced to load primary column explicitly.
|
||||
*/
|
||||
#define Q_STATUS_EQUAL(p, s) (((p) & (s)) != 0)
|
||||
|
||||
|
||||
#define QUERY_IS_ASC_QUERY(q) (GET_FORWARD_DIRECTION_FACTOR((q)->order.order) == QUERY_ASC_FORWARD_STEP)
|
||||
|
||||
#define IS_MASTER_SCAN(runtime) ((runtime)->scanFlag == MASTER_SCAN)
|
||||
|
|
@ -1602,11 +1600,11 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int16_t order
|
|||
SColIndex* pIndex = &pSqlFuncMsg->colInfo;
|
||||
|
||||
if (TSDB_COL_REQ_NULL(pIndex->flag)) {
|
||||
pCtx->requireNull = true;
|
||||
pIndex->flag &= ~(TSDB_COL_NULL);
|
||||
pCtx->requireNull = true;
|
||||
pIndex->flag &= ~(TSDB_COL_NULL);
|
||||
} else {
|
||||
pCtx->requireNull = false;
|
||||
}
|
||||
pCtx->requireNull = false;
|
||||
}
|
||||
|
||||
int32_t index = pSqlFuncMsg->colInfo.colIndex;
|
||||
if (TSDB_COL_IS_TAG(pIndex->flag)) {
|
||||
|
|
@ -1927,24 +1925,24 @@ static bool onlyFirstQuery(SQuery *pQuery) { return onlyOneQueryType(pQuery, TSD
|
|||
static bool onlyLastQuery(SQuery *pQuery) { return onlyOneQueryType(pQuery, TSDB_FUNC_LAST, TSDB_FUNC_LAST_DST); }
|
||||
|
||||
// todo refactor, add iterator
|
||||
static void doExchangeTimeWindow(SQInfo* pQInfo) {
|
||||
size_t t = GET_NUM_OF_TABLEGROUP(pQInfo);
|
||||
static void doExchangeTimeWindow(SQInfo* pQInfo, STimeWindow* win) {
|
||||
size_t t = taosArrayGetSize(pQInfo->tableGroupInfo.pGroupList);
|
||||
for(int32_t i = 0; i < t; ++i) {
|
||||
SArray* p1 = GET_TABLEGROUP(pQInfo, i);
|
||||
SArray* p1 = taosArrayGetP(pQInfo->tableGroupInfo.pGroupList, i);
|
||||
|
||||
SArray* tableKeyGroup = taosArrayGetP(pQInfo->tableGroupInfo.pGroupList, i);
|
||||
size_t len = taosArrayGetSize(p1);
|
||||
for(int32_t j = 0; j < len; ++j) {
|
||||
STableQueryInfo* pTableQueryInfo = (STableQueryInfo*) taosArrayGetP(p1, j);
|
||||
SWAP(pTableQueryInfo->win.skey, pTableQueryInfo->win.ekey, TSKEY);
|
||||
STableKeyInfo* pInfo = taosArrayGet(p1, j);
|
||||
|
||||
STableKeyInfo* pInfo = taosArrayGet(tableKeyGroup, j);
|
||||
pInfo->lastKey = pTableQueryInfo->win.skey;
|
||||
// update the new lastkey if it is equalled to the value of the old skey
|
||||
if (pInfo->lastKey == win->ekey) {
|
||||
pInfo->lastKey = win->skey;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
static void changeExecuteScanOrder(SQInfo *pQInfo, bool stableQuery) {
|
||||
static void changeExecuteScanOrder(SQInfo *pQInfo, SQueryTableMsg* pQueryMsg, bool stableQuery) {
|
||||
SQuery* pQuery = pQInfo->runtimeEnv.pQuery;
|
||||
|
||||
// in case of point-interpolation query, use asc order scan
|
||||
|
|
@ -1953,34 +1951,36 @@ static void changeExecuteScanOrder(SQInfo *pQInfo, bool stableQuery) {
|
|||
|
||||
// todo handle the case the the order irrelevant query type mixed up with order critical query type
|
||||
// descending order query for last_row query
|
||||
if (isFirstLastRowQuery(pQuery)) {
|
||||
if (isFirstLastRowQuery(pQuery) && !QUERY_IS_ASC_QUERY(pQuery)) {
|
||||
qDebug("QInfo:%p scan order changed for last_row query, old:%d, new:%d", GET_QINFO_ADDR(pQuery),
|
||||
pQuery->order.order, TSDB_ORDER_ASC);
|
||||
|
||||
SWAP(pQuery->window.skey, pQuery->window.ekey, TSKEY);
|
||||
pQuery->order.order = TSDB_ORDER_ASC;
|
||||
if (pQuery->window.skey > pQuery->window.ekey) {
|
||||
SWAP(pQuery->window.skey, pQuery->window.ekey, TSKEY);
|
||||
}
|
||||
assert (pQuery->window.skey <= pQuery->window.ekey);
|
||||
|
||||
doExchangeTimeWindow(pQInfo, &pQuery->window);
|
||||
return;
|
||||
}
|
||||
|
||||
if (isGroupbyNormalCol(pQuery->pGroupbyExpr)) {
|
||||
if (isGroupbyNormalCol(pQuery->pGroupbyExpr) && !QUERY_IS_ASC_QUERY(pQuery)) {
|
||||
pQuery->order.order = TSDB_ORDER_ASC;
|
||||
if (pQuery->window.skey > pQuery->window.ekey) {
|
||||
SWAP(pQuery->window.skey, pQuery->window.ekey, TSKEY);
|
||||
}
|
||||
SWAP(pQuery->window.skey, pQuery->window.ekey, TSKEY);
|
||||
assert (pQuery->window.skey <= pQuery->window.ekey);
|
||||
|
||||
doExchangeTimeWindow(pQInfo, &pQuery->window);
|
||||
return;
|
||||
}
|
||||
|
||||
if (isPointInterpoQuery(pQuery) && pQuery->intervalTime == 0) {
|
||||
if (!QUERY_IS_ASC_QUERY(pQuery)) {
|
||||
qDebug(msg, GET_QINFO_ADDR(pQuery), "interp", pQuery->order.order, TSDB_ORDER_ASC, pQuery->window.skey,
|
||||
pQuery->window.ekey, pQuery->window.ekey, pQuery->window.skey);
|
||||
SWAP(pQuery->window.skey, pQuery->window.ekey, TSKEY);
|
||||
}
|
||||
if (isPointInterpoQuery(pQuery) && (pQuery->intervalTime == 0) && !QUERY_IS_ASC_QUERY(pQuery)) {
|
||||
qDebug(msg, GET_QINFO_ADDR(pQuery), "interp", pQuery->order.order, TSDB_ORDER_ASC, pQuery->window.skey,
|
||||
pQuery->window.ekey, pQuery->window.ekey, pQuery->window.skey);
|
||||
SWAP(pQuery->window.skey, pQuery->window.ekey, TSKEY);
|
||||
|
||||
pQuery->order.order = TSDB_ORDER_ASC;
|
||||
|
||||
assert (pQuery->window.skey <= pQuery->window.ekey);
|
||||
doExchangeTimeWindow(pQInfo, &pQuery->window);
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
@ -1991,7 +1991,7 @@ static void changeExecuteScanOrder(SQInfo *pQInfo, bool stableQuery) {
|
|||
pQuery->window.ekey, pQuery->window.ekey, pQuery->window.skey);
|
||||
|
||||
SWAP(pQuery->window.skey, pQuery->window.ekey, TSKEY);
|
||||
doExchangeTimeWindow(pQInfo);
|
||||
doExchangeTimeWindow(pQInfo, &pQuery->window);
|
||||
}
|
||||
|
||||
pQuery->order.order = TSDB_ORDER_ASC;
|
||||
|
|
@ -2001,7 +2001,7 @@ static void changeExecuteScanOrder(SQInfo *pQInfo, bool stableQuery) {
|
|||
pQuery->window.ekey, pQuery->window.ekey, pQuery->window.skey);
|
||||
|
||||
SWAP(pQuery->window.skey, pQuery->window.ekey, TSKEY);
|
||||
doExchangeTimeWindow(pQInfo);
|
||||
doExchangeTimeWindow(pQInfo, &pQuery->window);
|
||||
}
|
||||
|
||||
pQuery->order.order = TSDB_ORDER_DESC;
|
||||
|
|
@ -2015,6 +2015,7 @@ static void changeExecuteScanOrder(SQInfo *pQInfo, bool stableQuery) {
|
|||
pQuery->window.skey, pQuery->window.ekey, pQuery->window.ekey, pQuery->window.skey);
|
||||
|
||||
SWAP(pQuery->window.skey, pQuery->window.ekey, TSKEY);
|
||||
doExchangeTimeWindow(pQInfo, &pQuery->window);
|
||||
}
|
||||
|
||||
pQuery->order.order = TSDB_ORDER_ASC;
|
||||
|
|
@ -2024,6 +2025,7 @@ static void changeExecuteScanOrder(SQInfo *pQInfo, bool stableQuery) {
|
|||
pQuery->window.skey, pQuery->window.ekey, pQuery->window.ekey, pQuery->window.skey);
|
||||
|
||||
SWAP(pQuery->window.skey, pQuery->window.ekey, TSKEY);
|
||||
doExchangeTimeWindow(pQInfo, &pQuery->window);
|
||||
}
|
||||
|
||||
pQuery->order.order = TSDB_ORDER_DESC;
|
||||
|
|
@ -2918,11 +2920,11 @@ int32_t mergeIntoGroupResultImpl(SQInfo *pQInfo, SArray *pGroup) {
|
|||
STableQueryInfo *item = taosArrayGetP(pGroup, i);
|
||||
|
||||
SIDList list = getDataBufPagesIdList(pRuntimeEnv->pResultBuf, TSDB_TABLEID(item->pTable)->tid);
|
||||
pageList = list;
|
||||
tid = TSDB_TABLEID(item->pTable)->tid;
|
||||
|
||||
if (taosArrayGetSize(list) > 0 && item->windowResInfo.size > 0) {
|
||||
pTableList[numOfTables++] = item;
|
||||
tid = TSDB_TABLEID(item->pTable)->tid;
|
||||
pageList = list;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -4352,6 +4354,32 @@ static bool skipTimeInterval(SQueryRuntimeEnv *pRuntimeEnv, TSKEY* start) {
|
|||
return true;
|
||||
}
|
||||
|
||||
static void freeTableQueryInfo(STableGroupInfo* pTableGroupInfo) {
|
||||
if (pTableGroupInfo->pGroupList == NULL) {
|
||||
assert(pTableGroupInfo->numOfTables == 0);
|
||||
} else {
|
||||
size_t numOfGroups = taosArrayGetSize(pTableGroupInfo->pGroupList);
|
||||
for (int32_t i = 0; i < numOfGroups; ++i) {
|
||||
SArray *p = taosArrayGetP(pTableGroupInfo->pGroupList, i);
|
||||
|
||||
size_t num = taosArrayGetSize(p);
|
||||
for(int32_t j = 0; j < num; ++j) {
|
||||
STableQueryInfo* item = taosArrayGetP(p, j);
|
||||
destroyTableQueryInfo(item);
|
||||
}
|
||||
|
||||
taosArrayDestroy(p);
|
||||
}
|
||||
|
||||
taosArrayDestroy(pTableGroupInfo->pGroupList);
|
||||
pTableGroupInfo->pGroupList = NULL;
|
||||
pTableGroupInfo->numOfTables = 0;
|
||||
}
|
||||
|
||||
taosHashCleanup(pTableGroupInfo->map);
|
||||
pTableGroupInfo->map = NULL;
|
||||
}
|
||||
|
||||
static int32_t setupQueryHandle(void* tsdb, SQInfo* pQInfo, bool isSTableQuery) {
|
||||
SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv;
|
||||
SQuery *pQuery = pQInfo->runtimeEnv.pQuery;
|
||||
|
|
@ -4387,20 +4415,22 @@ static int32_t setupQueryHandle(void* tsdb, SQInfo* pQInfo, bool isSTableQuery)
|
|||
terrno = TSDB_CODE_SUCCESS;
|
||||
if (isFirstLastRowQuery(pQuery)) {
|
||||
pRuntimeEnv->pQueryHandle = tsdbQueryLastRow(tsdb, &cond, &pQInfo->tableGroupInfo, pQInfo);
|
||||
if (pRuntimeEnv->pQueryHandle == NULL) { // no data in current stable, clear all
|
||||
freeTableQueryInfo(&pQInfo->tableqinfoGroupInfo);
|
||||
} else { // update the query time window
|
||||
pQuery->window = cond.twindow;
|
||||
|
||||
// update the query time window
|
||||
pQuery->window = cond.twindow;
|
||||
size_t numOfGroups = GET_NUM_OF_TABLEGROUP(pQInfo);
|
||||
for (int32_t i = 0; i < numOfGroups; ++i) {
|
||||
SArray *group = GET_TABLEGROUP(pQInfo, i);
|
||||
|
||||
size_t numOfGroups = GET_NUM_OF_TABLEGROUP(pQInfo);
|
||||
for(int32_t i = 0; i < numOfGroups; ++i) {
|
||||
SArray *group = GET_TABLEGROUP(pQInfo, i);
|
||||
size_t t = taosArrayGetSize(group);
|
||||
for (int32_t j = 0; j < t; ++j) {
|
||||
STableQueryInfo *pCheckInfo = taosArrayGetP(group, j);
|
||||
|
||||
size_t t = taosArrayGetSize(group);
|
||||
for (int32_t j = 0; j < t; ++j) {
|
||||
STableQueryInfo *pCheckInfo = taosArrayGetP(group, j);
|
||||
|
||||
pCheckInfo->win = pQuery->window;
|
||||
pCheckInfo->lastKey = pCheckInfo->win.skey;
|
||||
pCheckInfo->win = pQuery->window;
|
||||
pCheckInfo->lastKey = pCheckInfo->win.skey;
|
||||
}
|
||||
}
|
||||
}
|
||||
} else if (isPointInterpoQuery(pQuery)) {
|
||||
|
|
@ -4449,15 +4479,17 @@ int32_t doInitQInfo(SQInfo *pQInfo, STSBuf *pTsBuf, void *tsdb, int32_t vgId, bo
|
|||
|
||||
setScanLimitationByResultBuffer(pQuery);
|
||||
|
||||
// NOTE: pTableCheckInfo need to update the query time range and the lastKey info
|
||||
// TODO fixme
|
||||
changeExecuteScanOrder(pQInfo, isSTableQuery);
|
||||
|
||||
code = setupQueryHandle(tsdb, pQInfo, isSTableQuery);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
return code;
|
||||
}
|
||||
|
||||
if (pQInfo->tableqinfoGroupInfo.numOfTables == 0) {
|
||||
qDebug("QInfo:%p no table qualified for tag filter, abort query", pQInfo);
|
||||
setQueryStatus(pQuery, QUERY_COMPLETED);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
pQInfo->tsdb = tsdb;
|
||||
pQInfo->vgId = vgId;
|
||||
|
||||
|
|
@ -6022,14 +6054,6 @@ static void doUpdateExprColumnIndex(SQuery *pQuery) {
|
|||
}
|
||||
}
|
||||
|
||||
static int compareTableIdInfo(const void* a, const void* b) {
|
||||
const STableIdInfo* x = (const STableIdInfo*)a;
|
||||
const STableIdInfo* y = (const STableIdInfo*)b;
|
||||
if (x->uid > y->uid) return 1;
|
||||
if (x->uid < y->uid) return -1;
|
||||
return 0;
|
||||
}
|
||||
|
||||
static void freeQInfo(SQInfo *pQInfo);
|
||||
|
||||
static void calResultBufSize(SQuery* pQuery) {
|
||||
|
|
@ -6051,8 +6075,8 @@ static void calResultBufSize(SQuery* pQuery) {
|
|||
}
|
||||
}
|
||||
|
||||
static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SArray* pTableIdList, SSqlGroupbyExpr *pGroupbyExpr, SExprInfo *pExprs,
|
||||
STableGroupInfo *pTableGroupInfo, SColumnInfo* pTagCols) {
|
||||
static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SSqlGroupbyExpr *pGroupbyExpr, SExprInfo *pExprs,
|
||||
STableGroupInfo *pTableGroupInfo, SColumnInfo* pTagCols, bool stableQuery) {
|
||||
int16_t numOfCols = pQueryMsg->numOfCols;
|
||||
int16_t numOfOutput = pQueryMsg->numOfOutput;
|
||||
|
||||
|
|
@ -6151,8 +6175,6 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SArray* pTableIdList,
|
|||
}
|
||||
|
||||
int tableIndex = 0;
|
||||
STimeWindow window = pQueryMsg->window;
|
||||
taosArraySort(pTableIdList, compareTableIdInfo);
|
||||
|
||||
pQInfo->runtimeEnv.interBufSize = getOutputInterResultBufSize(pQuery);
|
||||
pQInfo->pBuf = calloc(pTableGroupInfo->numOfTables, sizeof(STableQueryInfo));
|
||||
|
|
@ -6161,12 +6183,20 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SArray* pTableIdList,
|
|||
}
|
||||
|
||||
// NOTE: pTableCheckInfo need to update the query time range and the lastKey info
|
||||
// changeExecuteScanOrder(pQInfo, stableQuery);
|
||||
pQInfo->arrTableIdInfo = taosArrayInit(tableIndex, sizeof(STableIdInfo));
|
||||
pQInfo->dataReady = QUERY_RESULT_NOT_READY;
|
||||
pthread_mutex_init(&pQInfo->lock, NULL);
|
||||
|
||||
pQuery->pos = -1;
|
||||
pQuery->window = pQueryMsg->window;
|
||||
changeExecuteScanOrder(pQInfo, pQueryMsg, stableQuery);
|
||||
|
||||
STimeWindow window = pQuery->window;
|
||||
|
||||
int32_t index = 0;
|
||||
|
||||
for(int32_t i = 0; i < numOfGroups; ++i) {
|
||||
SArray* pa = taosArrayGetP(pTableGroupInfo->pGroupList, i);
|
||||
SArray* pa = taosArrayGetP(pQInfo->tableGroupInfo.pGroupList, i);
|
||||
|
||||
size_t s = taosArrayGetSize(pa);
|
||||
SArray* p1 = taosArrayInit(s, POINTER_BYTES);
|
||||
|
|
@ -6179,12 +6209,9 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SArray* pTableIdList,
|
|||
for(int32_t j = 0; j < s; ++j) {
|
||||
STableKeyInfo* info = taosArrayGet(pa, j);
|
||||
|
||||
STableId* id = TSDB_TABLEID(info->pTable);
|
||||
STableIdInfo* pTableId = taosArraySearch(pTableIdList, id, compareTableIdInfo);
|
||||
|
||||
window.skey = (pTableId != NULL)? pTableId->key:pQueryMsg->window.skey;
|
||||
void* buf = (char*)pQInfo->pBuf + index * sizeof(STableQueryInfo);
|
||||
|
||||
window.skey = info->lastKey;
|
||||
STableQueryInfo* item = createTableQueryInfo(&pQInfo->runtimeEnv, info->pTable, window, buf);
|
||||
if (item == NULL) {
|
||||
goto _cleanup;
|
||||
|
|
@ -6192,17 +6219,13 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SArray* pTableIdList,
|
|||
|
||||
item->groupIndex = i;
|
||||
taosArrayPush(p1, &item);
|
||||
|
||||
STableId* id = TSDB_TABLEID(info->pTable);
|
||||
taosHashPut(pQInfo->tableqinfoGroupInfo.map, &id->tid, sizeof(id->tid), &item, POINTER_BYTES);
|
||||
index += 1;
|
||||
}
|
||||
}
|
||||
|
||||
pQInfo->arrTableIdInfo = taosArrayInit(tableIndex, sizeof(STableIdInfo));
|
||||
pQInfo->dataReady = QUERY_RESULT_NOT_READY;
|
||||
pthread_mutex_init(&pQInfo->lock, NULL);
|
||||
|
||||
pQuery->pos = -1;
|
||||
pQuery->window = pQueryMsg->window;
|
||||
colIdCheck(pQuery);
|
||||
|
||||
qDebug("qmsg:%p QInfo:%p created", pQueryMsg, pQInfo);
|
||||
|
|
@ -6360,29 +6383,13 @@ static void freeQInfo(SQInfo *pQInfo) {
|
|||
taosTFree(pQuery);
|
||||
}
|
||||
|
||||
// todo refactor, extract method to destroytableDataInfo
|
||||
if (pQInfo->tableqinfoGroupInfo.pGroupList != NULL) {
|
||||
int32_t numOfGroups = (int32_t)(GET_NUM_OF_TABLEGROUP(pQInfo));
|
||||
for (int32_t i = 0; i < numOfGroups; ++i) {
|
||||
SArray *p = GET_TABLEGROUP(pQInfo, i);
|
||||
|
||||
size_t num = taosArrayGetSize(p);
|
||||
for(int32_t j = 0; j < num; ++j) {
|
||||
STableQueryInfo* item = taosArrayGetP(p, j);
|
||||
destroyTableQueryInfo(item);
|
||||
}
|
||||
|
||||
taosArrayDestroy(p);
|
||||
}
|
||||
}
|
||||
freeTableQueryInfo(&pQInfo->tableqinfoGroupInfo);
|
||||
|
||||
taosTFree(pQInfo->pBuf);
|
||||
taosArrayDestroy(pQInfo->tableqinfoGroupInfo.pGroupList);
|
||||
taosHashCleanup(pQInfo->tableqinfoGroupInfo.map);
|
||||
|
||||
tsdbDestroyTableGroup(&pQInfo->tableGroupInfo);
|
||||
taosArrayDestroy(pQInfo->arrTableIdInfo);
|
||||
|
||||
|
||||
pQInfo->signature = 0;
|
||||
|
||||
qDebug("QInfo:%p QInfo is freed", pQInfo);
|
||||
|
|
@ -6558,7 +6565,7 @@ int32_t qCreateQueryInfo(void* tsdb, int32_t vgId, SQueryTableMsg* pQueryMsg, qi
|
|||
assert(0);
|
||||
}
|
||||
|
||||
(*pQInfo) = createQInfoImpl(pQueryMsg, pTableIdList, pGroupbyExpr, pExprs, &tableGroupInfo, pTagColumnInfo);
|
||||
(*pQInfo) = createQInfoImpl(pQueryMsg, pGroupbyExpr, pExprs, &tableGroupInfo, pTagColumnInfo, isSTableQuery);
|
||||
pExprs = NULL;
|
||||
pGroupbyExpr = NULL;
|
||||
pTagColumnInfo = NULL;
|
||||
|
|
|
|||
|
|
@ -154,9 +154,14 @@ int32_t tBucketBigIntHash(tMemBucket *pBucket, const void *value) {
|
|||
|
||||
// todo refactor to more generic
|
||||
int32_t tBucketIntHash(tMemBucket *pBucket, const void *value) {
|
||||
int32_t v = *(int32_t *)value;
|
||||
int32_t index = -1;
|
||||
int32_t v = 0;
|
||||
switch(pBucket->type) {
|
||||
case TSDB_DATA_TYPE_SMALLINT: v = *(int16_t*) value; break;
|
||||
case TSDB_DATA_TYPE_TINYINT: v = *(int8_t*) value; break;
|
||||
default: v = *(int32_t*) value;break;
|
||||
}
|
||||
|
||||
int32_t index = -1;
|
||||
if (pBucket->range.iMaxVal == INT32_MIN) {
|
||||
/*
|
||||
* taking negative integer into consideration,
|
||||
|
|
|
|||
|
|
@ -295,9 +295,16 @@ out_of_memory:
|
|||
}
|
||||
|
||||
TsdbQueryHandleT tsdbQueryLastRow(TSDB_REPO_T *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupList, void* qinfo) {
|
||||
pCond->order = TSDB_ORDER_ASC;
|
||||
pCond->twindow = changeTableGroupByLastrow(groupList);
|
||||
|
||||
// no qualified table
|
||||
if (groupList->numOfTables == 0) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
STsdbQueryHandle *pQueryHandle = (STsdbQueryHandle*) tsdbQueryTables(tsdb, pCond, groupList, qinfo);
|
||||
|
||||
assert(pCond->order == TSDB_ORDER_ASC && pCond->twindow.skey <= pCond->twindow.ekey);
|
||||
return pQueryHandle;
|
||||
}
|
||||
|
||||
|
|
@ -1981,8 +1988,9 @@ bool tsdbNextDataBlock(TsdbQueryHandleT* pHandle) {
|
|||
STimeWindow changeTableGroupByLastrow(STableGroupInfo *groupList) {
|
||||
STimeWindow window = {INT64_MAX, INT64_MIN};
|
||||
|
||||
int32_t totalNumOfTable = 0;
|
||||
|
||||
// NOTE: starts from the buffer in case of descending timestamp order check data blocks
|
||||
// todo consider the query time window, current last_row does not apply the query time window
|
||||
size_t numOfGroups = taosArrayGetSize(groupList->pGroupList);
|
||||
for(int32_t j = 0; j < numOfGroups; ++j) {
|
||||
SArray* pGroup = taosArrayGetP(groupList->pGroupList, j);
|
||||
|
|
@ -1993,8 +2001,9 @@ STimeWindow changeTableGroupByLastrow(STableGroupInfo *groupList) {
|
|||
size_t numOfTables = taosArrayGetSize(pGroup);
|
||||
for(int32_t i = 0; i < numOfTables; ++i) {
|
||||
STableKeyInfo* pKeyInfo = (STableKeyInfo*) taosArrayGet(pGroup, i);
|
||||
TSKEY lastKey = ((STable*)(pKeyInfo->pTable))->lastKey;
|
||||
|
||||
// if the lastKey equals to INT64_MIN, there is no data in this table
|
||||
TSKEY lastKey = ((STable*)(pKeyInfo->pTable))->lastKey;
|
||||
if (key < lastKey) {
|
||||
key = lastKey;
|
||||
|
||||
|
|
@ -2012,13 +2021,23 @@ STimeWindow changeTableGroupByLastrow(STableGroupInfo *groupList) {
|
|||
}
|
||||
}
|
||||
|
||||
// clear current group
|
||||
taosArrayClear(pGroup);
|
||||
|
||||
// more than one table in each group, only one table left for each group
|
||||
if (numOfTables > 1) {
|
||||
taosArrayClear(pGroup);
|
||||
if (keyInfo.pTable != NULL) {
|
||||
totalNumOfTable++;
|
||||
taosArrayPush(pGroup, &keyInfo);
|
||||
}
|
||||
}
|
||||
|
||||
// window does not being updated, so set the original
|
||||
if (window.skey == INT64_MAX && window.ekey == INT64_MIN) {
|
||||
window = TSWINDOW_INITIALIZER;
|
||||
assert(totalNumOfTable == 0);
|
||||
}
|
||||
|
||||
groupList->numOfTables = totalNumOfTable;
|
||||
return window;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -378,40 +378,43 @@ void taosCacheRelease(SCacheObj *pCacheObj, void **data, bool _remove) {
|
|||
} else {
|
||||
// NOTE: remove it from hash in the first place, otherwise, the pNode may have been released by other thread
|
||||
// when reaches here.
|
||||
SCacheDataNode* p = NULL;
|
||||
int32_t ret = taosHashRemoveWithData(pCacheObj->pHashTable, pNode->key, pNode->keySize, &p, sizeof(void*));
|
||||
SCacheDataNode *p = NULL;
|
||||
int32_t ret = taosHashRemoveWithData(pCacheObj->pHashTable, pNode->key, pNode->keySize, &p, sizeof(void *));
|
||||
ref = T_REF_DEC(pNode);
|
||||
|
||||
// successfully remove from hash table, if failed, this node must have been move to trash already, do nothing.
|
||||
// note that the remove operation can be executed only once.
|
||||
if (ret == 0) {
|
||||
if (p != pNode) {
|
||||
uDebug("cache:%s, key:%p, successfully removed a new entry:%p, refcnt:%d, prev entry:%p has been removed by others already", pCacheObj->name, pNode->key, p->data, T_REF_VAL_GET(p), pNode->data);
|
||||
assert(p->pTNodeHeader == NULL);
|
||||
taosAddToTrash(pCacheObj, p);
|
||||
} else {
|
||||
if (p != pNode) {
|
||||
uDebug( "cache:%s, key:%p, successfully removed a new entry:%p, refcnt:%d, prev entry:%p has been removed by "
|
||||
"others already", pCacheObj->name, pNode->key, p->data, T_REF_VAL_GET(p), pNode->data);
|
||||
|
||||
uDebug("cache:%s, key:%p, %p successfully removed from hash table, refcnt:%d", pCacheObj->name, pNode->key, pNode->data, ref);
|
||||
if (ref > 0) {
|
||||
assert(pNode->pTNodeHeader == NULL);
|
||||
assert(p->pTNodeHeader == NULL);
|
||||
taosAddToTrash(pCacheObj, p);
|
||||
} else {
|
||||
uDebug("cache:%s, key:%p, %p successfully removed from hash table, refcnt:%d", pCacheObj->name, pNode->key,
|
||||
pNode->data, ref);
|
||||
if (ref > 0) {
|
||||
assert(pNode->pTNodeHeader == NULL);
|
||||
|
||||
taosAddToTrash(pCacheObj, pNode);
|
||||
} else { // ref == 0
|
||||
atomic_sub_fetch_64(&pCacheObj->totalSize, pNode->size);
|
||||
taosAddToTrash(pCacheObj, pNode);
|
||||
} else { // ref == 0
|
||||
atomic_sub_fetch_64(&pCacheObj->totalSize, pNode->size);
|
||||
|
||||
int32_t size = (int32_t)taosHashGetSize(pCacheObj->pHashTable);
|
||||
uDebug("cache:%s, key:%p, %p is destroyed from cache, size:%dbytes, num:%d size:%" PRId64 "bytes",
|
||||
pCacheObj->name, pNode->key, pNode->data, pNode->size, size, pCacheObj->totalSize);
|
||||
int32_t size = (int32_t)taosHashGetSize(pCacheObj->pHashTable);
|
||||
uDebug("cache:%s, key:%p, %p is destroyed from cache, size:%dbytes, num:%d size:%" PRId64 "bytes",
|
||||
pCacheObj->name, pNode->key, pNode->data, pNode->size, size, pCacheObj->totalSize);
|
||||
|
||||
if (pCacheObj->freeFp) {
|
||||
pCacheObj->freeFp(pNode->data);
|
||||
if (pCacheObj->freeFp) {
|
||||
pCacheObj->freeFp(pNode->data);
|
||||
}
|
||||
|
||||
free(pNode);
|
||||
}
|
||||
|
||||
free(pNode);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
uDebug("cache:%s, key:%p, %p has been removed from hash table by other thread already, refcnt:%d", pCacheObj->name, pNode->key, pNode->data, ref);
|
||||
uDebug("cache:%s, key:%p, %p has been removed from hash table by other thread already, refcnt:%d",
|
||||
pCacheObj->name, pNode->key, pNode->data, ref);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -513,7 +516,7 @@ void taosAddToTrash(SCacheObj *pCacheObj, SCacheDataNode *pNode) {
|
|||
pCacheObj->numOfElemsInTrash++;
|
||||
__cache_unlock(pCacheObj);
|
||||
|
||||
uDebug("%s key:%p, %p move to trash, numOfElem in trash:%d", pCacheObj->name, pNode->key, pNode->data,
|
||||
uDebug("cache:%s key:%p, %p move to trash, numOfElem in trash:%d", pCacheObj->name, pNode->key, pNode->data,
|
||||
pCacheObj->numOfElemsInTrash);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -10,6 +10,6 @@ IF (HEADER_GTEST_INCLUDE_DIR AND LIB_GTEST_STATIC_DIR)
|
|||
INCLUDE_DIRECTORIES(${HEADER_GTEST_INCLUDE_DIR})
|
||||
AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR} SOURCE_LIST)
|
||||
|
||||
ADD_EXECUTABLE(utilTest ./cacheTest.cpp ./hashTest.cpp)
|
||||
ADD_EXECUTABLE(utilTest ${SOURCE_LIST})
|
||||
TARGET_LINK_LIBRARIES(utilTest tutil common osdetail gtest pthread gcov)
|
||||
ENDIF()
|
||||
|
|
|
|||
|
|
@ -1,16 +1,9 @@
|
|||
#include "os.h"
|
||||
#include <iostream>
|
||||
#include <gtest/gtest.h>
|
||||
#include <sys/time.h>
|
||||
|
||||
#include "taos.h"
|
||||
//#include "tsdb.h"
|
||||
|
||||
//#include "testCommon.h"
|
||||
#include "tstoken.h"
|
||||
#include "tutil.h"
|
||||
#include "tcache.h"
|
||||
#include "ttimer.h"
|
||||
|
||||
namespace {
|
||||
int32_t tsMaxMgmtConnections = 10000;
|
||||
|
|
|
|||
|
|
@ -693,7 +693,7 @@ class DbConnRest(DbConn):
|
|||
def __init__(self):
|
||||
super().__init__()
|
||||
self._type = self.TYPE_REST
|
||||
self._url = "http://localhost:6020/rest/sql" # fixed for now
|
||||
self._url = "http://localhost:6041/rest/sql" # fixed for now
|
||||
self._result = None
|
||||
|
||||
def openByType(self): # Open connection
|
||||
|
|
@ -1306,6 +1306,7 @@ class DbManager():
|
|||
"Cannot establish DB connection, please re-run script without parameter, and follow the instructions.")
|
||||
sys.exit(2)
|
||||
else:
|
||||
print("Failed to connect to DB, errno = {}, msg: {}".format(Helper.convertErrno(err.errno), err.msg))
|
||||
raise
|
||||
except BaseException:
|
||||
print("[=] Unexpected exception")
|
||||
|
|
@ -1910,10 +1911,19 @@ class TaskReadData(StateTransitionTask):
|
|||
# 'twa(speed)', # TODO: this one REQUIRES a where statement, not reasonable
|
||||
'sum(speed)',
|
||||
'stddev(speed)',
|
||||
# SELECTOR functions
|
||||
'min(speed)',
|
||||
'max(speed)',
|
||||
'first(speed)',
|
||||
'last(speed)']) # TODO: add more from 'top'
|
||||
'last(speed)',
|
||||
# 'top(speed)', # TODO: not supported?
|
||||
# 'bottom(speed)', # TODO: not supported?
|
||||
# 'percentile(speed, 10)', # TODO: TD-1316
|
||||
'last_row(speed)',
|
||||
# Transformation Functions
|
||||
# 'diff(speed)', # TODO: no supported?!
|
||||
'spread(speed)'
|
||||
]) # TODO: add more from 'top'
|
||||
filterExpr = Dice.choice([ # TODO: add various kind of WHERE conditions
|
||||
None
|
||||
])
|
||||
|
|
@ -2768,7 +2778,7 @@ class MainExec:
|
|||
try:
|
||||
ret = self._clientMgr.run(self._svcMgr) # stop TAOS service inside
|
||||
except requests.exceptions.ConnectionError as err:
|
||||
logger.warning("Failed to open REST connection to DB")
|
||||
logger.warning("Failed to open REST connection to DB: {}".format(err.getMessage()))
|
||||
# don't raise
|
||||
return ret
|
||||
|
||||
|
|
|
|||
|
|
@ -347,6 +347,8 @@ if $rows != 3 then
|
|||
return -1
|
||||
endi
|
||||
|
||||
print ======================udc with normal column group by
|
||||
|
||||
sql_error select from t1
|
||||
sql_error select abc from t1
|
||||
sql_error select abc as tu from t1
|
||||
|
|
|
|||
|
|
@ -152,3 +152,23 @@ sql select t1,t1,count(*),t1,t1 from lr_stb0 where ts>'2018-09-24 00:00:00.000'
|
|||
if $rows != 46 then
|
||||
return -1
|
||||
endi
|
||||
|
||||
print ========>td-1317, empty table last_row query crashed
|
||||
sql create table m1(ts timestamp, k int) tags (a int);
|
||||
sql create table t1 using m1 tags(1);
|
||||
sql create table t2 using m1 tags(2);
|
||||
|
||||
sql select last_row(*) from t1
|
||||
if $rows != 0 then
|
||||
return -1
|
||||
endi
|
||||
|
||||
sql select last_row(*) from m1
|
||||
if $rows != 0 then
|
||||
return -1
|
||||
endi
|
||||
|
||||
sql select last_row(*) from m1 where tbname in ('t1')
|
||||
if $rows != 0 then
|
||||
return -1
|
||||
endi
|
||||
|
|
|
|||
|
|
@ -99,6 +99,8 @@ run general/parser/union.sim
|
|||
sleep 2000
|
||||
run general/parser/constCol.sim
|
||||
sleep 2000
|
||||
run general/parser/timestamp.sim
|
||||
sleep 2000
|
||||
run general/parser/sliding.sim
|
||||
|
||||
#sleep 2000
|
||||
|
|
|
|||
|
|
@ -20,7 +20,7 @@ $db = $dbPrefix . $i
|
|||
$stb = $stbPrefix . $i
|
||||
|
||||
sql drop database if exists $db
|
||||
sql create database $db maxrows 200 cache 1024 tblocks 200 maxTables 4
|
||||
sql create database $db maxrows 200 maxTables 4
|
||||
print ====== create tables
|
||||
sql use $db
|
||||
sql create table $stb (ts timestamp, c1 timestamp, c2 int) tags(t1 binary(20))
|
||||
|
|
|
|||
|
|
@ -22,12 +22,29 @@ $tsu = $tsu - $delta
|
|||
$tsu = $tsu + $ts0
|
||||
|
||||
##### select from supertable
|
||||
|
||||
$tb = $tbPrefix . 0
|
||||
sql select first(c1), last(c1) from $tb where ts >= $ts0 and ts < $tsu interval(5m) fill(value, -1)
|
||||
sql select first(c1), last(c1), (1537325400 - 1537146000)/(5*60) v from $tb where ts >= $ts0 and ts < $tsu interval(5m) fill(value, -1)
|
||||
$res = $rowNum * 2
|
||||
$res = $res - 1
|
||||
if $rows != $res then
|
||||
$n = $res - 2
|
||||
print ============>$n
|
||||
if $rows != $n then
|
||||
print expect $n, actual $rows
|
||||
return -1
|
||||
endi
|
||||
|
||||
if $data03 != 598.000000000 then
|
||||
print expect 598.000000000, actual $data03
|
||||
return -1
|
||||
endi
|
||||
|
||||
|
||||
if $data13 != 598.000000000 then
|
||||
print expect 598.000000000, actual $data03
|
||||
return -1
|
||||
endi
|
||||
|
||||
sql select first(c1), last(c1), (1537325400 - 1537146000)/(5*60) v from $tb where ts >= $ts0 and ts < $tsu interval(5m) fill(value, NULL)
|
||||
if $data13 != 598.000000000 then
|
||||
print expect 598.000000000, actual $data03
|
||||
return -1
|
||||
endi
|
||||
|
|
@ -118,4 +118,42 @@ if $data21 != 2.10000 then
|
|||
return -1
|
||||
endi
|
||||
|
||||
print =====================td-1302 case
|
||||
sql create database t1 keep 36500;
|
||||
sql use t1;
|
||||
sql create table test(ts timestamp, k int);
|
||||
sql insert into test values(29999, 1)(70000, 2)(80000, 3)
|
||||
|
||||
print ================== restart server to commit data into disk
|
||||
system sh/exec.sh -n dnode1 -s stop -x SIGINT
|
||||
sleep 5000
|
||||
system sh/exec.sh -n dnode1 -s start
|
||||
print ================== server restart completed
|
||||
sql connect
|
||||
sleep 3000
|
||||
|
||||
sql select count(*) from t1.test where ts>10000 and ts<90000 interval(5000a)
|
||||
if $rows != 3 then
|
||||
return -1
|
||||
endi
|
||||
|
||||
print =========>td-1308
|
||||
sql create database db;
|
||||
sql use db;
|
||||
|
||||
sql create table stb (ts timestamp, c1 int, c2 binary(10)) tags(t1 binary(10));
|
||||
sql create table tb1 using stb tags('a1');
|
||||
|
||||
sql insert into tb1 values('2020-09-03 15:30:48.812', 0, 'tb1');
|
||||
sql select count(*) from stb where ts > '2020-09-03 15:30:44' interval(4s);
|
||||
if $rows != 1 then
|
||||
return -1
|
||||
endi
|
||||
|
||||
sql create table tb4 using stb tags('a4');
|
||||
sql select count(*) from stb where ts > '2020-09-03 15:30:44' interval(4s);
|
||||
if $rows != 1 then
|
||||
return -1
|
||||
endi
|
||||
|
||||
system sh/exec.sh -n dnode1 -s stop -x SIGINT
|
||||
Loading…
Reference in New Issue