提交 bb1b3ae7 编写于 作者: W wpan

Merge branch 'feature/m2d7' into feature/TD-5739

...@@ -143,7 +143,7 @@ taosd -C ...@@ -143,7 +143,7 @@ taosd -C
TDengine集群中加入一个新的dnode时,涉及集群相关的一些参数必须与已有集群的配置相同,否则不能成功加入到集群中。会进行校验的参数如下: TDengine集群中加入一个新的dnode时,涉及集群相关的一些参数必须与已有集群的配置相同,否则不能成功加入到集群中。会进行校验的参数如下:
- numOfMnodes:系统中管理节点个数。默认值:3。 - numOfMnodes:系统中管理节点个数。默认值:3。(2.0 版本从 2.0.20.11 开始、2.1 及以上版本从 2.1.6.0 开始,numOfMnodes 默认值改为 1。)
- mnodeEqualVnodeNum: 一个mnode等同于vnode消耗的个数。默认值:4。 - mnodeEqualVnodeNum: 一个mnode等同于vnode消耗的个数。默认值:4。
- offlineThreshold: dnode离线阈值,超过该时间将导致该dnode从集群中删除。单位为秒,默认值:86400*10(即10天)。 - offlineThreshold: dnode离线阈值,超过该时间将导致该dnode从集群中删除。单位为秒,默认值:86400*10(即10天)。
- statusInterval: dnode向mnode报告状态时长。单位为秒,默认值:1。 - statusInterval: dnode向mnode报告状态时长。单位为秒,默认值:1。
......
...@@ -34,7 +34,7 @@ taos> DESCRIBE meters; ...@@ -34,7 +34,7 @@ taos> DESCRIBE meters;
- 时间格式为 ```YYYY-MM-DD HH:mm:ss.MS```,默认时间分辨率为毫秒。比如:```2017-08-12 18:25:58.128``` - 时间格式为 ```YYYY-MM-DD HH:mm:ss.MS```,默认时间分辨率为毫秒。比如:```2017-08-12 18:25:58.128```
- 内部函数 now 是客户端的当前时间 - 内部函数 now 是客户端的当前时间
- 插入记录时,如果时间戳为 now,插入数据时使用提交这条记录的客户端的当前时间 - 插入记录时,如果时间戳为 now,插入数据时使用提交这条记录的客户端的当前时间
- Epoch Time:时间戳也可以是一个长整数,表示从 1970-01-01 08:00:00.000 开始的毫秒数 - Epoch Time:时间戳也可以是一个长整数,表示从格林威治时间 1970-01-01 00:00:00.000 (UTC/GMT) 开始的毫秒数(相应地,如果所在 Database 的时间精度设置为“微秒”,则长整型格式的时间戳含义也就对应于从格林威治时间 1970-01-01 00:00:00.000 (UTC/GMT) 开始的微秒数)
- 时间可以加减,比如 now-2h,表明查询时刻向前推 2 个小时(最近 2 小时)。数字后面的时间单位可以是 u(微秒)、a(毫秒)、s(秒)、m(分)、h(小时)、d(天)、w(周)。 比如 `select * from t1 where ts > now-2w and ts <= now-1w`,表示查询两周前整整一周的数据。在指定降频操作(down sampling)的时间窗口(interval)时,时间单位还可以使用 n(自然月) 和 y(自然年)。 - 时间可以加减,比如 now-2h,表明查询时刻向前推 2 个小时(最近 2 小时)。数字后面的时间单位可以是 u(微秒)、a(毫秒)、s(秒)、m(分)、h(小时)、d(天)、w(周)。 比如 `select * from t1 where ts > now-2w and ts <= now-1w`,表示查询两周前整整一周的数据。在指定降频操作(down sampling)的时间窗口(interval)时,时间单位还可以使用 n(自然月) 和 y(自然年)。
TDengine 缺省的时间戳是毫秒精度,但通过在 CREATE DATABASE 时传递的 PRECISION 参数就可以支持微秒。 TDengine 缺省的时间戳是毫秒精度,但通过在 CREATE DATABASE 时传递的 PRECISION 参数就可以支持微秒。
......
...@@ -132,7 +132,7 @@ The SQL creates a database demo, each data file stores 10 days of data, the memo ...@@ -132,7 +132,7 @@ The SQL creates a database demo, each data file stores 10 days of data, the memo
When adding a new dnode to the TDengine cluster, some parameters related to the cluster must be the same as the configuration of the existing cluster, otherwise it cannot be successfully added to the cluster. The parameters that will be verified are as follows: When adding a new dnode to the TDengine cluster, some parameters related to the cluster must be the same as the configuration of the existing cluster, otherwise it cannot be successfully added to the cluster. The parameters that will be verified are as follows:
- numOfMnodes: the number of management nodes in the system. Default: 3. - numOfMnodes: the number of management nodes in the system. Default: 3. (Since version 2.0.20.11 and version 2.1.6.0, the default value of "numOfMnodes" has been changed to 1.)
- balance: whether to enable load balancing. 0: No, 1: Yes. Default: 1. - balance: whether to enable load balancing. 0: No, 1: Yes. Default: 1.
- mnodeEqualVnodeNum: an mnode is equal to the number of vnodes consumed. Default: 4. - mnodeEqualVnodeNum: an mnode is equal to the number of vnodes consumed. Default: 4.
- offlineThreshold: the threshold for a dnode to be offline, exceed which the dnode will be removed from the cluster. The unit is seconds, and the default value is 86400*10 (that is, 10 days). - offlineThreshold: the threshold for a dnode to be offline, exceed which the dnode will be removed from the cluster. The unit is seconds, and the default value is 86400*10 (that is, 10 days).
......
...@@ -344,6 +344,7 @@ int32_t tscCreateTableMetaFromSTableMeta(STableMeta* pChild, const char* name, v ...@@ -344,6 +344,7 @@ int32_t tscCreateTableMetaFromSTableMeta(STableMeta* pChild, const char* name, v
STableMeta* tscTableMetaDup(STableMeta* pTableMeta); STableMeta* tscTableMetaDup(STableMeta* pTableMeta);
SVgroupsInfo* tscVgroupsInfoDup(SVgroupsInfo* pVgroupsInfo); SVgroupsInfo* tscVgroupsInfoDup(SVgroupsInfo* pVgroupsInfo);
int32_t tscGetTagFilterSerializeLen(SQueryInfo* pQueryInfo);
int32_t tscGetColFilterSerializeLen(SQueryInfo* pQueryInfo); int32_t tscGetColFilterSerializeLen(SQueryInfo* pQueryInfo);
int32_t tscCreateQueryFromQueryInfo(SQueryInfo* pQueryInfo, SQueryAttr* pQueryAttr, void* addr); int32_t tscCreateQueryFromQueryInfo(SQueryInfo* pQueryInfo, SQueryAttr* pQueryAttr, void* addr);
void* createQInfoFromQueryNode(SQueryInfo* pQueryInfo, STableGroupInfo* pTableGroupInfo, SOperatorInfo* pOperator, char* sql, void* addr, int32_t stage, uint64_t qId); void* createQInfoFromQueryNode(SQueryInfo* pQueryInfo, STableGroupInfo* pTableGroupInfo, SOperatorInfo* pOperator, char* sql, void* addr, int32_t stage, uint64_t qId);
......
...@@ -992,9 +992,7 @@ int32_t tsParseValues(char **str, STableDataBlocks *pDataBlock, int maxRows, SIn ...@@ -992,9 +992,7 @@ int32_t tsParseValues(char **str, STableDataBlocks *pDataBlock, int maxRows, SIn
index = 0; index = 0;
sToken = tStrGetToken(*str, &index, false); sToken = tStrGetToken(*str, &index, false);
if (sToken.n == 0 || sToken.type != TK_RP) { if (sToken.n == 0 || sToken.type != TK_RP) {
tscSQLSyntaxErrMsg(pInsertParam->msg, ") expected", *str); return tscSQLSyntaxErrMsg(pCmd->payload, ") expected", *str);
code = TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
return code;
} }
*str += index; *str += index;
......
...@@ -900,6 +900,7 @@ int32_t tscValidateSqlInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) { ...@@ -900,6 +900,7 @@ int32_t tscValidateSqlInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
SSqlNode* pSqlNode = taosArrayGetP(pInfo->list, i); SSqlNode* pSqlNode = taosArrayGetP(pInfo->list, i);
tscTrace("0x%"PRIx64" start to parse the %dth subclause, total:%"PRIzu, pSql->self, i, size); tscTrace("0x%"PRIx64" start to parse the %dth subclause, total:%"PRIzu, pSql->self, i, size);
// normalizeSqlNode(pSqlNode); // normalize the column name in each function // normalizeSqlNode(pSqlNode); // normalize the column name in each function
if ((code = validateSqlNode(pSql, pSqlNode, pQueryInfo)) != TSDB_CODE_SUCCESS) { if ((code = validateSqlNode(pSql, pSqlNode, pQueryInfo)) != TSDB_CODE_SUCCESS) {
return code; return code;
...@@ -2028,7 +2029,6 @@ static SUdfInfo* isValidUdf(SArray* pUdfInfo, const char* name, int32_t len) { ...@@ -2028,7 +2029,6 @@ static SUdfInfo* isValidUdf(SArray* pUdfInfo, const char* name, int32_t len) {
tscError("udfinfo is null"); tscError("udfinfo is null");
return NULL; return NULL;
} }
size_t t = taosArrayGetSize(pUdfInfo); size_t t = taosArrayGetSize(pUdfInfo);
for(int32_t i = 0; i < t; ++i) { for(int32_t i = 0; i < t; ++i) {
SUdfInfo* pUdf = taosArrayGet(pUdfInfo, i); SUdfInfo* pUdf = taosArrayGet(pUdfInfo, i);
...@@ -6078,10 +6078,12 @@ int32_t setAlterTableInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) { ...@@ -6078,10 +6078,12 @@ int32_t setAlterTableInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
SSchema* pSchema = tscGetTableTagSchema(pTableMetaInfo->pTableMeta); SSchema* pSchema = tscGetTableTagSchema(pTableMetaInfo->pTableMeta);
int16_t numOfTags = tscGetNumOfTags(pTableMetaInfo->pTableMeta); int16_t numOfTags = tscGetNumOfTags(pTableMetaInfo->pTableMeta);
int16_t i; int32_t numOfCols = tscGetNumOfColumns(pTableMetaInfo->pTableMeta);
int32_t tagIndex = columnIndex.columnIndex - numOfCols;
assert(tagIndex>=0);
uint32_t nLen = 0; uint32_t nLen = 0;
for (i = 0; i < numOfTags; ++i) { for (int i = 0; i < numOfTags; ++i) {
nLen += (i != columnIndex.columnIndex) ? pSchema[i].bytes : pItem->bytes; nLen += (i != tagIndex) ? pSchema[i].bytes : pItem->bytes;
} }
if (nLen >= TSDB_MAX_TAGS_LEN) { if (nLen >= TSDB_MAX_TAGS_LEN) {
return invalidOperationMsg(pMsg, msg24); return invalidOperationMsg(pMsg, msg24);
...@@ -8455,6 +8457,7 @@ int32_t validateSqlNode(SSqlObj* pSql, SSqlNode* pSqlNode, SQueryInfo* pQueryInf ...@@ -8455,6 +8457,7 @@ int32_t validateSqlNode(SSqlObj* pSql, SSqlNode* pSqlNode, SQueryInfo* pQueryInf
const char* msg6 = "not support stddev/percentile/interp in the outer query yet"; const char* msg6 = "not support stddev/percentile/interp in the outer query yet";
const char* msg7 = "derivative/twa/irate requires timestamp column exists in subquery"; const char* msg7 = "derivative/twa/irate requires timestamp column exists in subquery";
const char* msg8 = "condition missing for join query"; const char* msg8 = "condition missing for join query";
const char* msg9 = "not support 3 level select";
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
...@@ -8485,6 +8488,13 @@ int32_t validateSqlNode(SSqlObj* pSql, SSqlNode* pSqlNode, SQueryInfo* pQueryInf ...@@ -8485,6 +8488,13 @@ int32_t validateSqlNode(SSqlObj* pSql, SSqlNode* pSqlNode, SQueryInfo* pQueryInf
// parse the subquery in the first place // parse the subquery in the first place
int32_t numOfSub = (int32_t)taosArrayGetSize(pSqlNode->from->list); int32_t numOfSub = (int32_t)taosArrayGetSize(pSqlNode->from->list);
for (int32_t i = 0; i < numOfSub; ++i) { for (int32_t i = 0; i < numOfSub; ++i) {
// check if there is 3 level select
SRelElementPair* subInfo = taosArrayGet(pSqlNode->from->list, i);
SSqlNode* p = taosArrayGetP(subInfo->pSubquery, 0);
if (p->from->type == SQL_NODE_FROM_SUBQUERY){
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg9);
}
code = doValidateSubquery(pSqlNode, i, pSql, pQueryInfo, tscGetErrorMsgPayload(pCmd)); code = doValidateSubquery(pSqlNode, i, pSql, pQueryInfo, tscGetErrorMsgPayload(pCmd));
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
return code; return code;
......
...@@ -678,6 +678,8 @@ static int32_t tscEstimateQueryMsgSize(SSqlObj *pSql) { ...@@ -678,6 +678,8 @@ static int32_t tscEstimateQueryMsgSize(SSqlObj *pSql) {
int32_t srcColListSize = (int32_t)(taosArrayGetSize(pQueryInfo->colList) * sizeof(SColumnInfo)); int32_t srcColListSize = (int32_t)(taosArrayGetSize(pQueryInfo->colList) * sizeof(SColumnInfo));
int32_t srcColFilterSize = tscGetColFilterSerializeLen(pQueryInfo); int32_t srcColFilterSize = tscGetColFilterSerializeLen(pQueryInfo);
int32_t srcTagFilterSize = tscGetTagFilterSerializeLen(pQueryInfo);
size_t numOfExprs = tscNumOfExprs(pQueryInfo); size_t numOfExprs = tscNumOfExprs(pQueryInfo);
int32_t exprSize = (int32_t)(sizeof(SSqlExpr) * numOfExprs * 2); int32_t exprSize = (int32_t)(sizeof(SSqlExpr) * numOfExprs * 2);
...@@ -698,8 +700,8 @@ static int32_t tscEstimateQueryMsgSize(SSqlObj *pSql) { ...@@ -698,8 +700,8 @@ static int32_t tscEstimateQueryMsgSize(SSqlObj *pSql) {
tableSerialize = totalTables * sizeof(STableIdInfo); tableSerialize = totalTables * sizeof(STableIdInfo);
} }
return MIN_QUERY_MSG_PKT_SIZE + minMsgSize() + sizeof(SQueryTableMsg) + srcColListSize + srcColFilterSize + exprSize + tsBufSize + return MIN_QUERY_MSG_PKT_SIZE + minMsgSize() + sizeof(SQueryTableMsg) + srcColListSize + srcColFilterSize + srcTagFilterSize +
tableSerialize + sqlLen + 4096 + pQueryInfo->bufLen; exprSize + tsBufSize + tableSerialize + sqlLen + 4096 + pQueryInfo->bufLen;
} }
static char *doSerializeTableInfo(SQueryTableMsg *pQueryMsg, SSqlObj *pSql, STableMetaInfo *pTableMetaInfo, char *pMsg, static char *doSerializeTableInfo(SQueryTableMsg *pQueryMsg, SSqlObj *pSql, STableMetaInfo *pTableMetaInfo, char *pMsg,
...@@ -1035,7 +1037,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) { ...@@ -1035,7 +1037,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
SCond *pCond = tsGetSTableQueryCond(pTagCond, pTableMeta->id.uid); SCond *pCond = tsGetSTableQueryCond(pTagCond, pTableMeta->id.uid);
if (pCond != NULL && pCond->cond != NULL) { if (pCond != NULL && pCond->cond != NULL) {
pQueryMsg->tagCondLen = htons(pCond->len); pQueryMsg->tagCondLen = htonl(pCond->len);
memcpy(pMsg, pCond->cond, pCond->len); memcpy(pMsg, pCond->cond, pCond->len);
pMsg += pCond->len; pMsg += pCond->len;
......
...@@ -2790,7 +2790,8 @@ void tscDequoteAndTrimToken(SStrToken* pToken) { ...@@ -2790,7 +2790,8 @@ void tscDequoteAndTrimToken(SStrToken* pToken) {
} }
int32_t tscValidateName(SStrToken* pToken) { int32_t tscValidateName(SStrToken* pToken) {
if (pToken->type != TK_STRING && pToken->type != TK_ID) { if (pToken == NULL || pToken->z == NULL ||
(pToken->type != TK_STRING && pToken->type != TK_ID)) {
return TSDB_CODE_TSC_INVALID_OPERATION; return TSDB_CODE_TSC_INVALID_OPERATION;
} }
...@@ -4684,6 +4685,21 @@ int32_t tscGetColFilterSerializeLen(SQueryInfo* pQueryInfo) { ...@@ -4684,6 +4685,21 @@ int32_t tscGetColFilterSerializeLen(SQueryInfo* pQueryInfo) {
return len; return len;
} }
int32_t tscGetTagFilterSerializeLen(SQueryInfo* pQueryInfo) {
// serialize tag column query condition
if (pQueryInfo->tagCond.pCond != NULL && taosArrayGetSize(pQueryInfo->tagCond.pCond) > 0) {
STagCond* pTagCond = &pQueryInfo->tagCond;
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
STableMeta * pTableMeta = pTableMetaInfo->pTableMeta;
SCond *pCond = tsGetSTableQueryCond(pTagCond, pTableMeta->id.uid);
if (pCond != NULL && pCond->cond != NULL) {
return pCond->len;
}
}
return 0;
}
int32_t tscCreateQueryFromQueryInfo(SQueryInfo* pQueryInfo, SQueryAttr* pQueryAttr, void* addr) { int32_t tscCreateQueryFromQueryInfo(SQueryInfo* pQueryInfo, SQueryAttr* pQueryAttr, void* addr) {
memset(pQueryAttr, 0, sizeof(SQueryAttr)); memset(pQueryAttr, 0, sizeof(SQueryAttr));
......
...@@ -122,6 +122,7 @@ ...@@ -122,6 +122,7 @@
<exclude>**/TSDBJNIConnectorTest.java</exclude> <exclude>**/TSDBJNIConnectorTest.java</exclude>
<exclude>**/TaosInfoMonitorTest.java</exclude> <exclude>**/TaosInfoMonitorTest.java</exclude>
<exclude>**/UnsignedNumberJniTest.java</exclude> <exclude>**/UnsignedNumberJniTest.java</exclude>
<exclude>**/TimeZoneTest.java</exclude>
</excludes> </excludes>
<testFailureIgnore>true</testFailureIgnore> <testFailureIgnore>true</testFailureIgnore>
</configuration> </configuration>
......
...@@ -80,7 +80,8 @@ public class TSDBJNIConnector { ...@@ -80,7 +80,8 @@ public class TSDBJNIConnector {
this.taos = this.connectImp(host, port, dbName, user, password); this.taos = this.connectImp(host, port, dbName, user, password);
if (this.taos == TSDBConstants.JNI_NULL_POINTER) { if (this.taos == TSDBConstants.JNI_NULL_POINTER) {
throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_JNI_CONNECTION_NULL); String errMsg = this.getErrMsg(0);
throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_JNI_CONNECTION_NULL, errMsg);
} }
// invoke connectImp only here // invoke connectImp only here
taosInfo.conn_open_increment(); taosInfo.conn_open_increment();
......
...@@ -34,9 +34,8 @@ public class QueryDataTest { ...@@ -34,9 +34,8 @@ public class QueryDataTest {
String createTableSql = "create table " + stbName + "(ts timestamp, name binary(64))"; String createTableSql = "create table " + stbName + "(ts timestamp, name binary(64))";
statement.executeUpdate(createTableSql); statement.executeUpdate(createTableSql);
} catch (SQLException e) { } catch (SQLException e) {
return; e.printStackTrace();
} }
} }
......
package com.taosdata.jdbc.cases;
import com.taosdata.jdbc.TSDBDriver;
import org.junit.Test;
import java.sql.*;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.Properties;
public class TimeZoneTest {
private String url = "jdbc:TAOS://127.0.0.1:6030/?user=root&password=taosdata";
@Test
public void javaTimeZone() {
LocalDateTime localDateTime = LocalDateTime.of(1970, 1, 1, 0, 0, 0);
Instant instant = localDateTime.atZone(ZoneId.of("UTC-8")).toInstant();
System.out.println("UTC-8: " + instant.getEpochSecond() + "," + instant);
instant = localDateTime.atZone(ZoneId.of("UT")).toInstant();
System.out.println("UTC: " + instant.getEpochSecond() + "," + instant);
instant = localDateTime.atZone(ZoneId.of("UTC+8")).toInstant();
System.out.println("UTC+8: " + instant.getEpochSecond() + "," + instant);
}
@Test
public void taosTimeZone() {
// given
Properties props = new Properties();
props.setProperty(TSDBDriver.PROPERTY_KEY_TIME_ZONE, "UTC-8");
// when and then
try (Connection connection = DriverManager.getConnection(url, props)) {
Statement stmt = connection.createStatement();
stmt.execute("drop database if exists timezone_test");
stmt.execute("create database if not exists timezone_test keep 365000");
stmt.execute("use timezone_test");
stmt.execute("create table weather(ts timestamp, temperature float)");
stmt.execute("insert into timezone_test.weather(ts, temperature) values('1970-01-01 00:00:00', 1.0)");
ResultSet rs = stmt.executeQuery("select * from timezone_test.weather");
while (rs.next()) {
Timestamp ts = rs.getTimestamp("ts");
System.out.println("ts: " + ts.getTime() + "," + ts);
}
stmt.execute("insert into timezone_test.weather(ts, temperature, humidity) values('1970-01-02 00:00:00', 1.0, 2.0)");
rs = stmt.executeQuery("select * from timezone_test.weather");
while (rs.next()) {
Timestamp ts = rs.getTimestamp("ts");
System.out.println("ts: " + ts.getTime() + "," + ts);
}
stmt.execute("drop database if exists timezone_test");
stmt.close();
} catch (SQLException e) {
e.printStackTrace();
}
}
}
\ No newline at end of file
...@@ -490,7 +490,7 @@ typedef struct { ...@@ -490,7 +490,7 @@ typedef struct {
int16_t numOfCols; // the number of columns will be load from vnode int16_t numOfCols; // the number of columns will be load from vnode
SInterval interval; SInterval interval;
SSessionWindow sw; // session window SSessionWindow sw; // session window
uint16_t tagCondLen; // tag length in current query uint32_t tagCondLen; // tag length in current query
uint32_t tbnameCondLen; // table name filter condition string length uint32_t tbnameCondLen; // table name filter condition string length
int16_t numOfGroupCols; // num of group by columns int16_t numOfGroupCols; // num of group by columns
int16_t orderByIdx; int16_t orderByIdx;
......
...@@ -94,7 +94,7 @@ STsdbRepo *tsdbOpenRepo(STsdbCfg *pCfg, STsdbAppH *pAppH); ...@@ -94,7 +94,7 @@ STsdbRepo *tsdbOpenRepo(STsdbCfg *pCfg, STsdbAppH *pAppH);
int tsdbCloseRepo(STsdbRepo *repo, int toCommit); int tsdbCloseRepo(STsdbRepo *repo, int toCommit);
int32_t tsdbConfigRepo(STsdbRepo *repo, STsdbCfg *pCfg); int32_t tsdbConfigRepo(STsdbRepo *repo, STsdbCfg *pCfg);
int tsdbGetState(STsdbRepo *repo); int tsdbGetState(STsdbRepo *repo);
bool tsdbInCompact(STsdbRepo *repo); int8_t tsdbGetCompactState(STsdbRepo *repo);
// --------- TSDB TABLE DEFINITION // --------- TSDB TABLE DEFINITION
typedef struct { typedef struct {
uint64_t uid; // the unique table ID uint64_t uid; // the unique table ID
......
...@@ -5095,7 +5095,9 @@ static int getRowDataFromSample( ...@@ -5095,7 +5095,9 @@ static int getRowDataFromSample(
static int64_t generateStbRowData( static int64_t generateStbRowData(
SSuperTable* stbInfo, SSuperTable* stbInfo,
char* recBuf, int64_t timestamp) char* recBuf,
int64_t remainderBufLen,
int64_t timestamp)
{ {
int64_t dataLen = 0; int64_t dataLen = 0;
char *pstr = recBuf; char *pstr = recBuf;
...@@ -5123,6 +5125,7 @@ static int64_t generateStbRowData( ...@@ -5123,6 +5125,7 @@ static int64_t generateStbRowData(
rand_string(buf, stbInfo->columns[i].dataLen); rand_string(buf, stbInfo->columns[i].dataLen);
dataLen += snprintf(pstr + dataLen, maxLen - dataLen, "\'%s\',", buf); dataLen += snprintf(pstr + dataLen, maxLen - dataLen, "\'%s\',", buf);
tmfree(buf); tmfree(buf);
} else { } else {
char *tmp; char *tmp;
...@@ -5179,6 +5182,9 @@ static int64_t generateStbRowData( ...@@ -5179,6 +5182,9 @@ static int64_t generateStbRowData(
tstrncpy(pstr + dataLen, ",", 2); tstrncpy(pstr + dataLen, ",", 2);
dataLen += 1; dataLen += 1;
} }
if (dataLen > remainderBufLen)
return 0;
} }
dataLen -= 1; dataLen -= 1;
...@@ -5385,7 +5391,7 @@ static int32_t generateDataTailWithoutStb( ...@@ -5385,7 +5391,7 @@ static int32_t generateDataTailWithoutStb(
int32_t k = 0; int32_t k = 0;
for (k = 0; k < batch;) { for (k = 0; k < batch;) {
char data[MAX_DATA_SIZE]; char *data = pstr;
memset(data, 0, MAX_DATA_SIZE); memset(data, 0, MAX_DATA_SIZE);
int64_t retLen = 0; int64_t retLen = 0;
...@@ -5409,7 +5415,7 @@ static int32_t generateDataTailWithoutStb( ...@@ -5409,7 +5415,7 @@ static int32_t generateDataTailWithoutStb(
if (len > remainderBufLen) if (len > remainderBufLen)
break; break;
pstr += sprintf(pstr, "%s", data); pstr += retLen;
k++; k++;
len += retLen; len += retLen;
remainderBufLen -= retLen; remainderBufLen -= retLen;
...@@ -5465,14 +5471,14 @@ static int32_t generateStbDataTail( ...@@ -5465,14 +5471,14 @@ static int32_t generateStbDataTail(
int32_t k; int32_t k;
for (k = 0; k < batch;) { for (k = 0; k < batch;) {
char data[MAX_DATA_SIZE]; char *data = pstr;
memset(data, 0, MAX_DATA_SIZE);
int64_t lenOfRow = 0; int64_t lenOfRow = 0;
if (tsRand) { if (tsRand) {
if (superTblInfo->disorderRatio > 0) { if (superTblInfo->disorderRatio > 0) {
lenOfRow = generateStbRowData(superTblInfo, data, lenOfRow = generateStbRowData(superTblInfo, data,
remainderBufLen,
startTime + getTSRandTail( startTime + getTSRandTail(
superTblInfo->timeStampStep, k, superTblInfo->timeStampStep, k,
superTblInfo->disorderRatio, superTblInfo->disorderRatio,
...@@ -5480,6 +5486,7 @@ static int32_t generateStbDataTail( ...@@ -5480,6 +5486,7 @@ static int32_t generateStbDataTail(
); );
} else { } else {
lenOfRow = generateStbRowData(superTblInfo, data, lenOfRow = generateStbRowData(superTblInfo, data,
remainderBufLen,
startTime + superTblInfo->timeStampStep * k startTime + superTblInfo->timeStampStep * k
); );
} }
...@@ -5492,11 +5499,15 @@ static int32_t generateStbDataTail( ...@@ -5492,11 +5499,15 @@ static int32_t generateStbDataTail(
pSamplePos); pSamplePos);
} }
if (lenOfRow == 0) {
data[0] = '\0';
break;
}
if ((lenOfRow + 1) > remainderBufLen) { if ((lenOfRow + 1) > remainderBufLen) {
break; break;
} }
pstr += snprintf(pstr , lenOfRow + 1, "%s", data); pstr += lenOfRow;
k++; k++;
len += lenOfRow; len += lenOfRow;
remainderBufLen -= lenOfRow; remainderBufLen -= lenOfRow;
...@@ -6248,7 +6259,7 @@ static int32_t generateStbProgressiveData( ...@@ -6248,7 +6259,7 @@ static int32_t generateStbProgressiveData(
assert(buffer != NULL); assert(buffer != NULL);
char *pstr = buffer; char *pstr = buffer;
memset(buffer, 0, *pRemainderBufLen); memset(pstr, 0, *pRemainderBufLen);
int64_t headLen = generateStbSQLHead( int64_t headLen = generateStbSQLHead(
superTblInfo, superTblInfo,
...@@ -6642,7 +6653,7 @@ static void* syncWriteProgressive(threadInfo *pThreadInfo) { ...@@ -6642,7 +6653,7 @@ static void* syncWriteProgressive(threadInfo *pThreadInfo) {
return NULL; return NULL;
} }
int64_t remainderBufLen = maxSqlLen; int64_t remainderBufLen = maxSqlLen - 2000;
char *pstr = pThreadInfo->buffer; char *pstr = pThreadInfo->buffer;
int len = snprintf(pstr, int len = snprintf(pstr,
...@@ -6824,10 +6835,14 @@ static void callBack(void *param, TAOS_RES *res, int code) { ...@@ -6824,10 +6835,14 @@ static void callBack(void *param, TAOS_RES *res, int code) {
&& rand_num < pThreadInfo->superTblInfo->disorderRatio) { && rand_num < pThreadInfo->superTblInfo->disorderRatio) {
int64_t d = pThreadInfo->lastTs int64_t d = pThreadInfo->lastTs
- (taosRandom() % pThreadInfo->superTblInfo->disorderRange + 1); - (taosRandom() % pThreadInfo->superTblInfo->disorderRange + 1);
generateStbRowData(pThreadInfo->superTblInfo, data, d); generateStbRowData(pThreadInfo->superTblInfo, data,
MAX_DATA_SIZE,
d);
} else { } else {
generateStbRowData(pThreadInfo->superTblInfo, generateStbRowData(pThreadInfo->superTblInfo,
data, pThreadInfo->lastTs += 1000); data,
MAX_DATA_SIZE,
pThreadInfo->lastTs += 1000);
} }
pstr += sprintf(pstr, "%s", data); pstr += sprintf(pstr, "%s", data);
pThreadInfo->counter++; pThreadInfo->counter++;
...@@ -7052,6 +7067,7 @@ static void startMultiThreadInsertData(int threads, char* db_name, ...@@ -7052,6 +7067,7 @@ static void startMultiThreadInsertData(int threads, char* db_name,
for (int i = 0; i < threads; i++) { for (int i = 0; i < threads; i++) {
threadInfo *pThreadInfo = infos + i; threadInfo *pThreadInfo = infos + i;
pThreadInfo->threadID = i; pThreadInfo->threadID = i;
tstrncpy(pThreadInfo->db_name, db_name, TSDB_DB_NAME_LEN); tstrncpy(pThreadInfo->db_name, db_name, TSDB_DB_NAME_LEN);
pThreadInfo->time_precision = timePrec; pThreadInfo->time_precision = timePrec;
pThreadInfo->superTblInfo = superTblInfo; pThreadInfo->superTblInfo = superTblInfo;
......
...@@ -253,6 +253,10 @@ static int32_t mnodeProcessHeartBeatMsg(SMnodeMsg *pMsg) { ...@@ -253,6 +253,10 @@ static int32_t mnodeProcessHeartBeatMsg(SMnodeMsg *pMsg) {
int32_t connId = htonl(pHBMsg->connId); int32_t connId = htonl(pHBMsg->connId);
SConnObj *pConn = mnodeAccquireConn(connId, connInfo.user, connInfo.clientIp, connInfo.clientPort); SConnObj *pConn = mnodeAccquireConn(connId, connInfo.user, connInfo.clientIp, connInfo.clientPort);
if (pConn == NULL) {
pHBMsg->pid = htonl(pHBMsg->pid);
pConn = mnodeCreateConn(connInfo.user, connInfo.clientIp, connInfo.clientPort, pHBMsg->pid, pHBMsg->appName);
}
if (pConn == NULL) { if (pConn == NULL) {
// do not close existing links, otherwise // do not close existing links, otherwise
......
...@@ -7015,19 +7015,20 @@ static SSDataBlock* hashDistinct(void* param, bool* newgroup) { ...@@ -7015,19 +7015,20 @@ static SSDataBlock* hashDistinct(void* param, bool* newgroup) {
if (isNull(val, type)) { if (isNull(val, type)) {
continue; continue;
} }
char* p = val;
size_t keyLen = 0; size_t keyLen = 0;
if (IS_VAR_DATA_TYPE(pOperator->pExpr->base.colType)) { if (IS_VAR_DATA_TYPE(pOperator->pExpr->base.colType)) {
tstr* var = (tstr*)(val); tstr* var = (tstr*)(val);
p = var->data;
keyLen = varDataLen(var); keyLen = varDataLen(var);
} else { } else {
keyLen = bytes; keyLen = bytes;
} }
int dummy; int dummy;
void* res = taosHashGet(pInfo->pSet, val, keyLen); void* res = taosHashGet(pInfo->pSet, p, keyLen);
if (res == NULL) { if (res == NULL) {
taosHashPut(pInfo->pSet, val, keyLen, &dummy, sizeof(dummy)); taosHashPut(pInfo->pSet, p, keyLen, &dummy, sizeof(dummy));
char* start = pResultColInfoData->pData + bytes * pInfo->pRes->info.rows; char* start = pResultColInfoData->pData + bytes * pInfo->pRes->info.rows;
memcpy(start, val, bytes); memcpy(start, val, bytes);
pRes->info.rows += 1; pRes->info.rows += 1;
...@@ -7252,7 +7253,7 @@ int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SQueryParam* param) { ...@@ -7252,7 +7253,7 @@ int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SQueryParam* param) {
pQueryMsg->numOfCols = htons(pQueryMsg->numOfCols); pQueryMsg->numOfCols = htons(pQueryMsg->numOfCols);
pQueryMsg->numOfOutput = htons(pQueryMsg->numOfOutput); pQueryMsg->numOfOutput = htons(pQueryMsg->numOfOutput);
pQueryMsg->numOfGroupCols = htons(pQueryMsg->numOfGroupCols); pQueryMsg->numOfGroupCols = htons(pQueryMsg->numOfGroupCols);
pQueryMsg->tagCondLen = htons(pQueryMsg->tagCondLen); pQueryMsg->tagCondLen = htonl(pQueryMsg->tagCondLen);
pQueryMsg->tsBuf.tsOffset = htonl(pQueryMsg->tsBuf.tsOffset); pQueryMsg->tsBuf.tsOffset = htonl(pQueryMsg->tsBuf.tsOffset);
pQueryMsg->tsBuf.tsLen = htonl(pQueryMsg->tsBuf.tsLen); pQueryMsg->tsBuf.tsLen = htonl(pQueryMsg->tsBuf.tsLen);
pQueryMsg->tsBuf.tsNumOfBlocks = htonl(pQueryMsg->tsBuf.tsNumOfBlocks); pQueryMsg->tsBuf.tsNumOfBlocks = htonl(pQueryMsg->tsBuf.tsNumOfBlocks);
......
...@@ -94,8 +94,9 @@ struct STsdbRepo { ...@@ -94,8 +94,9 @@ struct STsdbRepo {
pthread_mutex_t mutex; pthread_mutex_t mutex;
bool repoLocked; bool repoLocked;
int32_t code; // Commit code int32_t code; // Commit code
SMergeBuf mergeBuf; //used when update=2 SMergeBuf mergeBuf; //used when update=2
bool inCompact; // is in compact process? int8_t compactState; // compact state: inCompact/noCompact/waitingCompact?
}; };
#define REPO_ID(r) (r)->config.tsdbId #define REPO_ID(r) (r)->config.tsdbId
......
...@@ -58,6 +58,7 @@ static int tsdbCompactFSetImpl(SCompactH *pComph); ...@@ -58,6 +58,7 @@ static int tsdbCompactFSetImpl(SCompactH *pComph);
static int tsdbWriteBlockToRightFile(SCompactH *pComph, STable *pTable, SDataCols *pDataCols, void **ppBuf, static int tsdbWriteBlockToRightFile(SCompactH *pComph, STable *pTable, SDataCols *pDataCols, void **ppBuf,
void **ppCBuf); void **ppCBuf);
enum { TSDB_NO_COMPACT, TSDB_IN_COMPACT, TSDB_WAITING_COMPACT};
int tsdbCompact(STsdbRepo *pRepo) { return tsdbAsyncCompact(pRepo); } int tsdbCompact(STsdbRepo *pRepo) { return tsdbAsyncCompact(pRepo); }
void *tsdbCompactImpl(STsdbRepo *pRepo) { void *tsdbCompactImpl(STsdbRepo *pRepo) {
...@@ -89,16 +90,21 @@ _err: ...@@ -89,16 +90,21 @@ _err:
} }
static int tsdbAsyncCompact(STsdbRepo *pRepo) { static int tsdbAsyncCompact(STsdbRepo *pRepo) {
if (pRepo->compactState != TSDB_NO_COMPACT) {
tsdbInfo("vgId:%d not compact tsdb again ", REPO_ID(pRepo));
return 0;
}
pRepo->compactState = TSDB_WAITING_COMPACT;
tsem_wait(&(pRepo->readyToCommit)); tsem_wait(&(pRepo->readyToCommit));
return tsdbScheduleCommit(pRepo, COMPACT_REQ); return tsdbScheduleCommit(pRepo, COMPACT_REQ);
} }
static void tsdbStartCompact(STsdbRepo *pRepo) { static void tsdbStartCompact(STsdbRepo *pRepo) {
ASSERT(!pRepo->inCompact); assert(pRepo->compactState != TSDB_IN_COMPACT);
tsdbInfo("vgId:%d start to compact!", REPO_ID(pRepo)); tsdbInfo("vgId:%d start to compact!", REPO_ID(pRepo));
tsdbStartFSTxn(pRepo, 0, 0); tsdbStartFSTxn(pRepo, 0, 0);
pRepo->code = TSDB_CODE_SUCCESS; pRepo->code = TSDB_CODE_SUCCESS;
pRepo->inCompact = true; pRepo->compactState = TSDB_IN_COMPACT;
} }
static void tsdbEndCompact(STsdbRepo *pRepo, int eno) { static void tsdbEndCompact(STsdbRepo *pRepo, int eno) {
...@@ -107,7 +113,7 @@ static void tsdbEndCompact(STsdbRepo *pRepo, int eno) { ...@@ -107,7 +113,7 @@ static void tsdbEndCompact(STsdbRepo *pRepo, int eno) {
} else { } else {
tsdbEndFSTxn(pRepo); tsdbEndFSTxn(pRepo);
} }
pRepo->inCompact = false; pRepo->compactState = TSDB_NO_COMPACT;
tsdbInfo("vgId:%d compact over, %s", REPO_ID(pRepo), (eno == TSDB_CODE_SUCCESS) ? "succeed" : "failed"); tsdbInfo("vgId:%d compact over, %s", REPO_ID(pRepo), (eno == TSDB_CODE_SUCCESS) ? "succeed" : "failed");
tsem_post(&(pRepo->readyToCommit)); tsem_post(&(pRepo->readyToCommit));
} }
......
...@@ -200,7 +200,7 @@ STsdbRepoInfo *tsdbGetStatus(STsdbRepo *pRepo) { return NULL; } ...@@ -200,7 +200,7 @@ STsdbRepoInfo *tsdbGetStatus(STsdbRepo *pRepo) { return NULL; }
int tsdbGetState(STsdbRepo *repo) { return repo->state; } int tsdbGetState(STsdbRepo *repo) { return repo->state; }
bool tsdbInCompact(STsdbRepo *repo) { return repo->inCompact; } int8_t tsdbGetCompactState(STsdbRepo *repo) { return (int8_t)(repo->compactState); }
void tsdbReportStat(void *repo, int64_t *totalPoints, int64_t *totalStorage, int64_t *compStorage) { void tsdbReportStat(void *repo, int64_t *totalPoints, int64_t *totalStorage, int64_t *compStorage) {
ASSERT(repo != NULL); ASSERT(repo != NULL);
...@@ -541,7 +541,7 @@ static STsdbRepo *tsdbNewRepo(STsdbCfg *pCfg, STsdbAppH *pAppH) { ...@@ -541,7 +541,7 @@ static STsdbRepo *tsdbNewRepo(STsdbCfg *pCfg, STsdbAppH *pAppH) {
pRepo->state = TSDB_STATE_OK; pRepo->state = TSDB_STATE_OK;
pRepo->code = TSDB_CODE_SUCCESS; pRepo->code = TSDB_CODE_SUCCESS;
pRepo->inCompact = false; pRepo->compactState = 0;
pRepo->config = *pCfg; pRepo->config = *pCfg;
if (pAppH) { if (pAppH) {
pRepo->appH = *pAppH; pRepo->appH = *pAppH;
......
...@@ -2708,6 +2708,7 @@ static bool loadBlockOfActiveTable(STsdbQueryHandle* pQueryHandle) { ...@@ -2708,6 +2708,7 @@ static bool loadBlockOfActiveTable(STsdbQueryHandle* pQueryHandle) {
} }
if (exists) { if (exists) {
tsdbRetrieveDataBlock((TsdbQueryHandleT) pQueryHandle, NULL);
if (pQueryHandle->currentLoadExternalRows && pQueryHandle->window.skey == pQueryHandle->window.ekey) { if (pQueryHandle->currentLoadExternalRows && pQueryHandle->window.skey == pQueryHandle->window.ekey) {
SColumnInfoData* pColInfo = taosArrayGet(pQueryHandle->pColumns, 0); SColumnInfoData* pColInfo = taosArrayGet(pQueryHandle->pColumns, 0);
assert(*(int64_t*)pColInfo->pData == pQueryHandle->window.skey); assert(*(int64_t*)pColInfo->pData == pQueryHandle->window.skey);
......
...@@ -162,7 +162,7 @@ static void vnodeBuildVloadMsg(SVnodeObj *pVnode, SStatusMsg *pStatus) { ...@@ -162,7 +162,7 @@ static void vnodeBuildVloadMsg(SVnodeObj *pVnode, SStatusMsg *pStatus) {
pLoad->status = pVnode->status; pLoad->status = pVnode->status;
pLoad->role = pVnode->role; pLoad->role = pVnode->role;
pLoad->replica = pVnode->syncCfg.replica; pLoad->replica = pVnode->syncCfg.replica;
pLoad->compact = (pVnode->tsdb != NULL) && tsdbInCompact(pVnode->tsdb) ? 1 : 0; pLoad->compact = (pVnode->tsdb != NULL) ? tsdbGetCompactState(pVnode->tsdb) : 0;
} }
int32_t vnodeGetVnodeList(int32_t vnodeList[], int32_t *numOfVnodes) { int32_t vnodeGetVnodeList(int32_t vnodeList[], int32_t *numOfVnodes) {
......
此差异已折叠。
...@@ -35,6 +35,9 @@ sql alter table tb1 set tag name = "" ...@@ -35,6 +35,9 @@ sql alter table tb1 set tag name = ""
sql alter table tb1 set tag name = "shenzhen" sql alter table tb1 set tag name = "shenzhen"
sql alter table tb1 set tag len = 379 sql alter table tb1 set tag len = 379
# case TD-5594
sql create stable st5520(ts timestamp, f int) tags(t0 bool, t1 nchar(4093), t2 nchar(1))
sql_error alter stable st5520 modify tag t2 nchar(2);
# test end # test end
sql drop database $db sql drop database $db
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册