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

Merge branch '3.0' of https://github.com/taosdata/TDengine into 3.0

This commit is contained in:
chenhaoran 2022-10-18 12:23:27 +00:00
parent 4697a3080a c77bc2b9b9
commit ae01bba8c6
50 changed files with 696 additions and 686 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cmake/taostools_CMakeLists.txt.in
View File

@ -2,7 +2,7 @@
# taos-tools
ExternalProject_Add(taos-tools
GIT_REPOSITORY https://github.com/taosdata/taos-tools.git
GIT_TAG 4d02980
GIT_TAG c64858f
SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools"
BINARY_DIR ""
#BUILD_IN_SOURCE TRUE

58
docs/en/14-reference/03-connector/05-go.mdx
View File

@ -7,7 +7,6 @@ title: TDengine Go Connector
import Tabs from '@theme/Tabs';
import TabItem from '@theme/TabItem';
import Preparition from "./_preparation.mdx"
import GoInsert from "../../07-develop/03-insert-data/_go_sql.mdx"
import GoInfluxLine from "../../07-develop/03-insert-data/_go_line.mdx"
import GoOpenTSDBTelnet from "../../07-develop/03-insert-data/_go_opts_telnet.mdx"
@ -176,6 +175,37 @@ func main() {
}
```
</TabItem>
<TabItem value="WebSocket" label="WebSocket connection">
_taosRestful_ implements Go's `database/sql/driver` interface via `http client`. You can use the [`database/sql`](https://golang.org/pkg/database/sql/) interface by simply introducing the driver (driver-go minimum version 3.0.2).
Use `taosWS` as `driverName` and use a correct [DSN](#DSN) as `dataSourceName` with the following parameters supported by the DSN.
* `writeTimeout` The timeout to send data via WebSocket.
* `readTimeout` The timeout to receive response data via WebSocket.
For example:
```go
package main
import (
"database/sql"
"fmt"
_ "github.com/taosdata/driver-go/v3/taosWS"
)
func main() {
var taosUri = "root:taosdata@ws(localhost:6041)/"
taos, err := sql.Open("taosWS", taosUri)
if err != nil {
fmt.Println("failed to connect TDengine, err:", err)
return
}
}
```
</TabItem>
</Tabs>
## Usage examples
@ -331,7 +361,7 @@ Creates consumer group.
* `func (c *Consumer) Subscribe(topics []string) error`
Subscribes to a topic.
Subscribes to topics.
* `func (c *Consumer) Poll(timeout time.Duration) (*Result, error)`
@ -409,6 +439,30 @@ Close consumer.
Closes the parameter binding.
### Subscribe via WebSocket
* `func NewConsumer(config *Config) (*Consumer, error)`
Creates consumer group.
* `func (c *Consumer) Subscribe(topic []string) error`
Subscribes to topics.
* `func (c *Consumer) Poll(timeout time.Duration) (*Result, error)`
Polling information.
* `func (c *Consumer) Commit(messageID uint64) error`
Commit information.
* `func (c *Consumer) Close() error`
Close consumer.
For a complete example see [GitHub sample file](https://github.com/taosdata/driver-go/blob/3.0/examples/tmqoverws/main.go)
## API Reference
Full API see [driver-go documentation](https://pkg.go.dev/github.com/taosdata/driver-go/v3)

55
docs/zh/08-connector/20-go.mdx
View File

@ -177,6 +177,37 @@ func main() {
}
```
</TabItem>
<TabItem value="WebSocket" label="WebSocket 连接">
_taosWS_ 通过 `WebSocket` 实现了 Go 的 `database/sql/driver` 接口。只需要引入驱动（driver-go 最低版本 3.0.2）就可以使用[`database/sql`](https://golang.org/pkg/database/sql/)的接口。
使用 `taosWS` 作为 `driverName` 并且使用一个正确的 [DSN](#DSN) 作为 `dataSourceName`，DSN 支持的参数：
* `writeTimeout` 通过 WebSocket 发送数据的超时时间。
* `readTimeout` 通过 WebSocket 接收响应数据的超时时间。
示例：
```go
package main
import (
"database/sql"
"fmt"
_ "github.com/taosdata/driver-go/v3/taosWS"
)
func main() {
var taosUri = "root:taosdata@ws(localhost:6041)/"
taos, err := sql.Open("taosWS", taosUri)
if err != nil {
fmt.Println("failed to connect TDengine, err:", err)
return
}
}
```
</TabItem>
</Tabs>
## 使用示例
@ -410,6 +441,30 @@ func main() {
结束参数绑定。
### 通过 WebSocket 订阅
* `func NewConsumer(config *Config) (*Consumer, error)`
创建消费者。
* `func (c *Consumer) Subscribe(topic []string) error`
订阅主题。
* `func (c *Consumer) Poll(timeout time.Duration) (*Result, error)`
轮询消息。
* `func (c *Consumer) Commit(messageID uint64) error`
提交消息。
* `func (c *Consumer) Close() error`
关闭消费者。
完整订阅示例参见 [GitHub 示例文件](https://github.com/taosdata/driver-go/blob/3.0/examples/tmqoverws/main.go)
## API 参考
全部 API 见 [driver-go 文档](https://pkg.go.dev/github.com/taosdata/driver-go/v3)

1
include/client/taos.h
View File

@ -198,6 +198,7 @@ DLL_EXPORT const void *taos_get_raw_block(TAOS_RES *res);
DLL_EXPORT int taos_load_table_info(TAOS *taos, const char *tableNameList);
DLL_EXPORT TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int protocol, int precision);
DLL_EXPORT TAOS_RES *taos_schemaless_insert_raw(TAOS* taos, char* lines, int len, int32_t *totalRows, int protocol, int precision);
/* --------------------------TMQ INTERFACE------------------------------- */

4
source/client/src/clientMsgHandler.c
View File

@ -194,6 +194,10 @@ int32_t processUseDbRsp(void* param, SDataBuf* pMsg, int32_t code) {
SUseDbRsp usedbRsp = {0};
tDeserializeSUseDbRsp(pMsg->pData, pMsg->len, &usedbRsp);
if(strlen(usedbRsp.db) == 0){
return TSDB_CODE_MND_DB_NOT_EXIST;
}
SName name = {0};
tNameFromString(&name, usedbRsp.db, T_NAME_ACCT | T_NAME_DB);

230
source/client/src/clientSml.c
View File

@ -28,8 +28,8 @@
#define QUOTE '"'
#define SLASH '\\'
#define JUMP_SPACE(sql) \
while (*sql != '\0') { \
#define JUMP_SPACE(sql, sqlEnd) \
while (sql < sqlEnd) { \
if (*sql == SPACE) \
sql++; \
else \
@ -917,16 +917,17 @@ static int32_t smlParseValue(SSmlKv *pVal, SSmlMsgBuf *msg) {
return TSDB_CODE_TSC_INVALID_VALUE;
}
static int32_t smlParseInfluxString(const char *sql, SSmlLineInfo *elements, SSmlMsgBuf *msg) {
static int32_t smlParseInfluxString(const char *sql, const char *sqlEnd, SSmlLineInfo *elements, SSmlMsgBuf *msg) {
if (!sql) return TSDB_CODE_SML_INVALID_DATA;
JUMP_SPACE(sql)
JUMP_SPACE(sql, sqlEnd)
if (*sql == COMMA) return TSDB_CODE_SML_INVALID_DATA;
elements->measure = sql;
// parse measure
while (*sql != '\0') {
while (sql < sqlEnd) {
if ((sql != elements->measure) && IS_SLASH_LETTER(sql)) {
MOVE_FORWARD_ONE(sql, strlen(sql) + 1);
MOVE_FORWARD_ONE(sql, sqlEnd - sql);
sqlEnd--;
continue;
}
if (IS_COMMA(sql)) {
@ -950,7 +951,7 @@ static int32_t smlParseInfluxString(const char *sql, SSmlLineInfo *elements, SSm
} else {
if (*sql == COMMA) sql++;
elements->tags = sql;
while (*sql != '\0') {
while (sql < sqlEnd) {
if (IS_SPACE(sql)) {
break;
}
@ -961,10 +962,10 @@ static int32_t smlParseInfluxString(const char *sql, SSmlLineInfo *elements, SSm
elements->measureTagsLen = sql - elements->measure;
// parse cols
JUMP_SPACE(sql)
JUMP_SPACE(sql, sqlEnd)
elements->cols = sql;
bool isInQuote = false;
while (*sql != '\0') {
while (sql < sqlEnd) {
if (IS_QUOTE(sql)) {
isInQuote = !isInQuote;
}
@ -984,10 +985,10 @@ static int32_t smlParseInfluxString(const char *sql, SSmlLineInfo *elements, SSm
}
// parse timestamp
JUMP_SPACE(sql)
JUMP_SPACE(sql, sqlEnd)
elements->timestamp = sql;
while (*sql != '\0') {
if (*sql == SPACE) {
while (sql < sqlEnd) {
if (isspace(*sql)) {
break;
}
sql++;
@ -997,8 +998,8 @@ static int32_t smlParseInfluxString(const char *sql, SSmlLineInfo *elements, SSm
return TSDB_CODE_SUCCESS;
}
static void smlParseTelnetElement(const char **sql, const char **data, int32_t *len) {
while (**sql != '\0') {
static void smlParseTelnetElement(const char **sql, const char *sqlEnd, const char **data, int32_t *len) {
while (*sql < sqlEnd) {
if (**sql != SPACE && !(*data)) {
*data = *sql;
} else if (**sql == SPACE && *data) {
@ -1009,20 +1010,20 @@ static void smlParseTelnetElement(const char **sql, const char **data, int32_t *
}
}
static int32_t smlParseTelnetTags(const char *data, SArray *cols, char *childTableName, SHashObj *dumplicateKey,
static int32_t smlParseTelnetTags(const char *data, const char *sqlEnd, SArray *cols, char *childTableName, SHashObj *dumplicateKey,
SSmlMsgBuf *msg) {
if(!cols) return TSDB_CODE_OUT_OF_MEMORY;
const char *sql = data;
size_t childTableNameLen = strlen(tsSmlChildTableName);
while (*sql != '\0') {
JUMP_SPACE(sql)
while (sql < sqlEnd) {
JUMP_SPACE(sql, sqlEnd)
if (*sql == '\0') break;
const char *key = sql;
int32_t keyLen = 0;
// parse key
while (*sql != '\0') {
while (sql < sqlEnd) {
if (*sql == SPACE) {
smlBuildInvalidDataMsg(msg, "invalid data", sql);
return TSDB_CODE_SML_INVALID_DATA;
@ -1047,7 +1048,7 @@ static int32_t smlParseTelnetTags(const char *data, SArray *cols, char *childTab
// parse value
const char *value = sql;
int32_t valueLen = 0;
while (*sql != '\0') {
while (sql < sqlEnd) {
// parse value
if (*sql == SPACE) {
break;
@ -1092,11 +1093,11 @@ static int32_t smlParseTelnetTags(const char *data, SArray *cols, char *childTab
}
// format: <metric> <timestamp> <value> <tagk_1>=<tagv_1>[ <tagk_n>=<tagv_n>]
static int32_t smlParseTelnetString(SSmlHandle *info, const char *sql, SSmlTableInfo *tinfo, SArray *cols) {
static int32_t smlParseTelnetString(SSmlHandle *info, const char *sql, const char *sqlEnd, SSmlTableInfo *tinfo, SArray *cols) {
if (!sql) return TSDB_CODE_SML_INVALID_DATA;
// parse metric
smlParseTelnetElement(&sql, &tinfo->sTableName, &tinfo->sTableNameLen);
smlParseTelnetElement(&sql, sqlEnd, &tinfo->sTableName, &tinfo->sTableNameLen);
if (!(tinfo->sTableName) || IS_INVALID_TABLE_LEN(tinfo->sTableNameLen)) {
smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", sql);
return TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH;
@ -1105,7 +1106,7 @@ static int32_t smlParseTelnetString(SSmlHandle *info, const char *sql, SSmlTable
// parse timestamp
const char *timestamp = NULL;
int32_t tLen = 0;
smlParseTelnetElement(&sql, &timestamp, &tLen);
smlParseTelnetElement(&sql, sqlEnd, &timestamp, &tLen);
if (!timestamp || tLen == 0) {
smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", sql);
return TSDB_CODE_SML_INVALID_DATA;
@ -1120,7 +1121,7 @@ static int32_t smlParseTelnetString(SSmlHandle *info, const char *sql, SSmlTable
// parse value
const char *value = NULL;
int32_t valueLen = 0;
smlParseTelnetElement(&sql, &value, &valueLen);
smlParseTelnetElement(&sql, sqlEnd, &value, &valueLen);
if (!value || valueLen == 0) {
smlBuildInvalidDataMsg(&info->msgBuf, "invalid value", sql);
return TSDB_CODE_TSC_INVALID_VALUE;
@ -1138,7 +1139,7 @@ static int32_t smlParseTelnetString(SSmlHandle *info, const char *sql, SSmlTable
}
// parse tags
ret = smlParseTelnetTags(sql, tinfo->tags, tinfo->childTableName, info->dumplicateKey, &info->msgBuf);
ret = smlParseTelnetTags(sql, sqlEnd, tinfo->tags, tinfo->childTableName, info->dumplicateKey, &info->msgBuf);
if (ret != TSDB_CODE_SUCCESS) {
smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", sql);
return ret;
@ -2073,11 +2074,11 @@ static int32_t smlParseJSONString(SSmlHandle *info, cJSON *root, SSmlTableInfo *
}
/************* TSDB_SML_JSON_PROTOCOL function end **************/
static int32_t smlParseInfluxLine(SSmlHandle *info, const char *sql) {
static int32_t smlParseInfluxLine(SSmlHandle *info, const char *sql, const int len) {
SSmlLineInfo elements = {0};
uDebug("SML:0x%" PRIx64 " smlParseInfluxLine sql:%s, hello", info->id, sql);
int ret = smlParseInfluxString(sql, &elements, &info->msgBuf);
int ret = smlParseInfluxString(sql, sql + len, &elements, &info->msgBuf);
if (ret != TSDB_CODE_SUCCESS) {
uError("SML:0x%" PRIx64 " smlParseInfluxLine failed", info->id);
return ret;
@ -2184,7 +2185,7 @@ static int32_t smlParseInfluxLine(SSmlHandle *info, const char *sql) {
return TSDB_CODE_SUCCESS;
}
static int32_t smlParseTelnetLine(SSmlHandle *info, void *data) {
static int32_t smlParseTelnetLine(SSmlHandle *info, void *data, const int len) {
int ret = TSDB_CODE_SUCCESS;
SSmlTableInfo *tinfo = smlBuildTableInfo();
if (!tinfo) {
@ -2198,7 +2199,7 @@ static int32_t smlParseTelnetLine(SSmlHandle *info, void *data) {
}
if (info->protocol == TSDB_SML_TELNET_PROTOCOL) {
ret = smlParseTelnetString(info, (const char *)data, tinfo, cols);
ret = smlParseTelnetString(info, (const char *)data, (char*)data + len, tinfo, cols);
} else if (info->protocol == TSDB_SML_JSON_PROTOCOL) {
ret = smlParseJSONString(info, (cJSON *)data, tinfo, cols);
} else {
@ -2289,7 +2290,7 @@ static int32_t smlParseJSON(SSmlHandle *info, char *payload) {
for (int32_t i = 0; i < payloadNum; ++i) {
cJSON *dataPoint = (payloadNum == 1 && cJSON_IsObject(root)) ? root : cJSON_GetArrayItem(root, i);
ret = smlParseTelnetLine(info, dataPoint);
ret = smlParseTelnetLine(info, dataPoint, -1);
if (ret != TSDB_CODE_SUCCESS) {
uError("SML:0x%" PRIx64 " Invalid JSON Payload", info->id);
goto end;
@ -2378,10 +2379,14 @@ static void smlPrintStatisticInfo(SSmlHandle *info) {
info->cost.endTime - info->cost.insertRpcTime, info->cost.endTime - info->cost.parseTime);
}
static int32_t smlParseLine(SSmlHandle *info, char *lines[], int numLines) {
static int32_t smlParseLine(SSmlHandle *info, char *lines[], char* rawLine, char* rawLineEnd, int numLines) {
int32_t code = TSDB_CODE_SUCCESS;
if (info->protocol == TSDB_SML_JSON_PROTOCOL) {
code = smlParseJSON(info, *lines);
if(lines){
code = smlParseJSON(info, *lines);
}else if(rawLine){
code = smlParseJSON(info, rawLine);
}
if (code != TSDB_CODE_SUCCESS) {
uError("SML:0x%" PRIx64 " smlParseJSON failed:%s", info->id, *lines);
return code;
@ -2390,28 +2395,46 @@ static int32_t smlParseLine(SSmlHandle *info, char *lines[], int numLines) {
}
for (int32_t i = 0; i < numLines; ++i) {
char *tmp = NULL;
int len = 0;
if(lines){
tmp = lines[i];
len = strlen(tmp);
}else if(rawLine){
tmp = rawLine;
while(rawLine < rawLineEnd){
if(*(rawLine++) == '\n'){
break;
}
len++;
}
if(info->protocol == TSDB_SML_LINE_PROTOCOL && tmp[0] == '#'){ // this line is comment
continue;
}
}
if (info->protocol == TSDB_SML_LINE_PROTOCOL) {
code = smlParseInfluxLine(info, lines[i]);
code = smlParseInfluxLine(info, tmp, len);
} else if (info->protocol == TSDB_SML_TELNET_PROTOCOL) {
code = smlParseTelnetLine(info, lines[i]);
code = smlParseTelnetLine(info, tmp, len);
} else {
ASSERT(0);
}
if (code != TSDB_CODE_SUCCESS) {
uError("SML:0x%" PRIx64 " smlParseLine failed. line %d : %s", info->id, i, lines[i]);
uError("SML:0x%" PRIx64 " smlParseLine failed. line %d : %s", info->id, i, tmp);
return code;
}
}
return code;
}
static int smlProcess(SSmlHandle *info, char *lines[], int numLines) {
static int smlProcess(SSmlHandle *info, char *lines[], char* rawLine, char* rawLineEnd, int numLines) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t retryNum = 0;
info->cost.parseTime = taosGetTimestampUs();
code = smlParseLine(info, lines, numLines);
code = smlParseLine(info, lines, rawLine, rawLineEnd, numLines);
if (code != 0) {
uError("SML:0x%" PRIx64 " smlParseLine error : %s", info->id, tstrerror(code));
return code;
@ -2504,39 +2527,8 @@ static void smlInsertCallback(void *param, void *res, int32_t code) {
smlDestroyInfo(info);
}
/**
* taos_schemaless_insert() parse and insert data points into database according to
* different protocol.
*
* @param $lines input array may contain multiple lines, each line indicates a data point.
* If protocol=2 is used input array should contain single JSON
* string(e.g. char *lines[] = {"$JSON_string"}). If need to insert
* multiple data points in JSON format, should include them in $JSON_string
* as a JSON array.
* @param $numLines indicates how many data points in $lines.
* If protocol = 2 is used this param will be ignored as $lines should
* contain single JSON string.
* @param $protocol indicates which protocol to use for parsing:
* 0 - influxDB line protocol
* 1 - OpenTSDB telnet line protocol
* 2 - OpenTSDB JSON format protocol
* @return return zero for successful insertion. Otherwise return none-zero error code of
* failure reason.
*
*/
TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int protocol, int precision) {
if (NULL == taos) {
terrno = TSDB_CODE_TSC_DISCONNECTED;
return NULL;
}
SRequestObj *request = (SRequestObj *)createRequest(*(int64_t *)taos, TSDB_SQL_INSERT);
if (!request) {
uError("SML:taos_schemaless_insert error request is null");
return NULL;
}
TAOS_RES *taos_schemaless_insert_inner(SRequestObj *request, char *lines[], char *rawLine, char *rawLineEnd, int numLines, int protocol, int precision) {
int batchs = 0;
STscObj *pTscObj = request->pTscObj;
@ -2560,12 +2552,6 @@ TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int pr
goto end;
}
if (!lines) {
request->code = TSDB_CODE_SML_INVALID_DATA;
smlBuildInvalidDataMsg(&msg, "lines is null", NULL);
goto end;
}
if (protocol < TSDB_SML_LINE_PROTOCOL || protocol > TSDB_SML_JSON_PROTOCOL) {
request->code = TSDB_CODE_SML_INVALID_PROTOCOL_TYPE;
smlBuildInvalidDataMsg(&msg, "protocol invalidate", NULL);
@ -2616,15 +2602,28 @@ TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int pr
info->affectedRows = perBatch;
info->pRequest->body.queryFp = smlInsertCallback;
info->pRequest->body.param = info;
int32_t code = smlProcess(info, lines, perBatch);
lines += perBatch;
int32_t code = smlProcess(info, lines, rawLine, rawLineEnd, perBatch);
if(lines){
lines += perBatch;
}
if(rawLine){
int num = 0;
while(rawLine < rawLineEnd){
if(*(rawLine++) == '\n'){
num++;
}
if(num == perBatch){
break;
}
}
}
if (code != TSDB_CODE_SUCCESS) {
info->pRequest->body.queryFp(info, req, code);
}
}
tsem_wait(&params.sem);
end:
end:
taosThreadSpinDestroy(&params.lock);
tsem_destroy(&params.sem);
// ((STscObj *)taos)->schemalessType = 0;
@ -2632,3 +2631,80 @@ end:
uDebug("resultend:%s", request->msgBuf);
return (TAOS_RES *)request;
}
/**
* taos_schemaless_insert() parse and insert data points into database according to
* different protocol.
*
* @param $lines input array may contain multiple lines, each line indicates a data point.
* If protocol=2 is used input array should contain single JSON
* string(e.g. char *lines[] = {"$JSON_string"}). If need to insert
* multiple data points in JSON format, should include them in $JSON_string
* as a JSON array.
* @param $numLines indicates how many data points in $lines.
* If protocol = 2 is used this param will be ignored as $lines should
* contain single JSON string.
* @param $protocol indicates which protocol to use for parsing:
* 0 - influxDB line protocol
* 1 - OpenTSDB telnet line protocol
* 2 - OpenTSDB JSON format protocol
* @return return zero for successful insertion. Otherwise return none-zero error code of
* failure reason.
*
*/
TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int protocol, int precision) {
if (NULL == taos) {
terrno = TSDB_CODE_TSC_DISCONNECTED;
return NULL;
}
SRequestObj *request = (SRequestObj *)createRequest(*(int64_t *)taos, TSDB_SQL_INSERT);
if (!request) {
uError("SML:taos_schemaless_insert error request is null");
return NULL;
}
if (!lines) {
SSmlMsgBuf msg = {ERROR_MSG_BUF_DEFAULT_SIZE, request->msgBuf};
request->code = TSDB_CODE_SML_INVALID_DATA;
smlBuildInvalidDataMsg(&msg, "lines is null", NULL);
return (TAOS_RES *)request;
}
return taos_schemaless_insert_inner(request, lines, NULL, NULL, numLines, protocol, precision);
}
TAOS_RES *taos_schemaless_insert_raw(TAOS* taos, char* lines, int len, int32_t *totalRows, int protocol, int precision){
if (NULL == taos) {
terrno = TSDB_CODE_TSC_DISCONNECTED;
return NULL;
}
SRequestObj *request = (SRequestObj *)createRequest(*(int64_t *)taos, TSDB_SQL_INSERT);
if (!request) {
uError("SML:taos_schemaless_insert error request is null");
return NULL;
}
if (!lines || len <= 0) {
SSmlMsgBuf msg = {ERROR_MSG_BUF_DEFAULT_SIZE, request->msgBuf};
request->code = TSDB_CODE_SML_INVALID_DATA;
smlBuildInvalidDataMsg(&msg, "lines is null", NULL);
return (TAOS_RES *)request;
}
int numLines = 0;
*totalRows = 0;
char *tmp = lines;
for(int i = 0; i < len; i++){
if(lines[i] == '\n' || i == len - 1){
numLines++;
if(tmp[0] != '#' || protocol != TSDB_SML_LINE_PROTOCOL){ //ignore comment
(*totalRows)++;
}
tmp = lines + i + 1;
}
}
return taos_schemaless_insert_inner(request, NULL, lines, lines + len, numLines, protocol, precision);
}

28
source/client/test/smlTest.cpp
View File

@ -44,7 +44,7 @@ TEST(testCase, smlParseInfluxString_Test) {
char *tmp = "\\,st,t1=3,t2=4,t3=t3 c1=3i64,c3=\"passit hello,c1=2\",c2=false,c4=4f64 1626006833639000000 ,32,c=3";
char *sql = (char *)taosMemoryCalloc(256, 1);
memcpy(sql, tmp, strlen(tmp) + 1);
int ret = smlParseInfluxString(sql, &elements, &msgBuf);
int ret = smlParseInfluxString(sql, sql + strlen(sql), &elements, &msgBuf);
ASSERT_EQ(ret, 0);
ASSERT_EQ(elements.measure, sql);
ASSERT_EQ(elements.measureLen, strlen(",st"));
@ -63,14 +63,14 @@ TEST(testCase, smlParseInfluxString_Test) {
tmp = "st,t1=3,t2=4,t3=t3 c1=3i64,c3=\"passit hello,c1=2,c2=false,c4=4f64 1626006833639000000";
memcpy(sql, tmp, strlen(tmp) + 1);
memset(&elements, 0, sizeof(SSmlLineInfo));
ret = smlParseInfluxString(sql, &elements, &msgBuf);
ret = smlParseInfluxString(sql, sql + strlen(sql), &elements, &msgBuf);
ASSERT_NE(ret, 0);
// case 3 false
tmp = "st, t1=3,t2=4,t3=t3 c1=3i64,c3=\"passit hello,c1=2,c2=false,c4=4f64 1626006833639000000";
memcpy(sql, tmp, strlen(tmp) + 1);
memset(&elements, 0, sizeof(SSmlLineInfo));
ret = smlParseInfluxString(sql, &elements, &msgBuf);
ret = smlParseInfluxString(sql, sql + strlen(sql), &elements, &msgBuf);
ASSERT_EQ(ret, 0);
ASSERT_EQ(elements.cols, sql + elements.measureTagsLen + 1);
ASSERT_EQ(elements.colsLen, strlen("t1=3,t2=4,t3=t3"));
@ -79,7 +79,7 @@ TEST(testCase, smlParseInfluxString_Test) {
tmp = "st, c1=3i64,c3=\"passit hello,c1=2\",c2=false,c4=4f64 1626006833639000000";
memcpy(sql, tmp, strlen(tmp) + 1);
memset(&elements, 0, sizeof(SSmlLineInfo));
ret = smlParseInfluxString(sql, &elements, &msgBuf);
ret = smlParseInfluxString(sql, sql + strlen(sql), &elements, &msgBuf);
ASSERT_EQ(ret, 0);
ASSERT_EQ(elements.measure, sql);
ASSERT_EQ(elements.measureLen, strlen("st"));
@ -98,7 +98,7 @@ TEST(testCase, smlParseInfluxString_Test) {
tmp = " st c1=3i64,c3=\"passit hello,c1=2\",c2=false,c4=4f64 1626006833639000000 ";
memcpy(sql, tmp, strlen(tmp) + 1);
memset(&elements, 0, sizeof(SSmlLineInfo));
ret = smlParseInfluxString(sql, &elements, &msgBuf);
ret = smlParseInfluxString(sql, sql + strlen(sql), &elements, &msgBuf);
ASSERT_EQ(ret, 0);
ASSERT_EQ(elements.measure, sql + 1);
ASSERT_EQ(elements.measureLen, strlen("st"));
@ -116,21 +116,21 @@ TEST(testCase, smlParseInfluxString_Test) {
tmp = " st c1=3i64,c3=\"passit hello,c1=2\",c2=false,c4=4f64 ";
memcpy(sql, tmp, strlen(tmp) + 1);
memset(&elements, 0, sizeof(SSmlLineInfo));
ret = smlParseInfluxString(sql, &elements, &msgBuf);
ret = smlParseInfluxString(sql, sql + strlen(sql), &elements, &msgBuf);
ASSERT_EQ(ret, 0);
// case 7
tmp = " st , ";
memcpy(sql, tmp, strlen(tmp) + 1);
memset(&elements, 0, sizeof(SSmlLineInfo));
ret = smlParseInfluxString(sql, &elements, &msgBuf);
ret = smlParseInfluxString(sql, sql + strlen(sql), &elements, &msgBuf);
ASSERT_EQ(ret, 0);
// case 8 false
tmp = ", st , ";
memcpy(sql, tmp, strlen(tmp) + 1);
memset(&elements, 0, sizeof(SSmlLineInfo));
ret = smlParseInfluxString(sql, &elements, &msgBuf);
ret = smlParseInfluxString(sql, sql + strlen(sql), &elements, &msgBuf);
ASSERT_NE(ret, 0);
taosMemoryFree(sql);
}
@ -542,7 +542,7 @@ TEST(testCase, smlParseTelnetLine_error_Test) {
"sys.procs.running 1479496100 42 host= web01",
};
for (int i = 0; i < sizeof(sql) / sizeof(sql[0]); i++) {
int ret = smlParseTelnetLine(info, (void *)sql[i]);
int ret = smlParseTelnetLine(info, (void *)sql[i], strlen(sql[i]));
ASSERT_NE(ret, 0);
}
@ -561,7 +561,7 @@ TEST(testCase, smlParseTelnetLine_diff_type_Test) {
int ret = TSDB_CODE_SUCCESS;
for (int i = 0; i < sizeof(sql) / sizeof(sql[0]); i++) {
ret = smlParseTelnetLine(info, (void *)sql[i]);
ret = smlParseTelnetLine(info, (void *)sql[i], strlen(sql[i]));
if (ret != TSDB_CODE_SUCCESS) break;
}
ASSERT_NE(ret, 0);
@ -617,7 +617,7 @@ TEST(testCase, smlParseTelnetLine_json_error_Test) {
int ret = TSDB_CODE_SUCCESS;
for (int i = 0; i < sizeof(sql) / sizeof(sql[0]); i++) {
ret = smlParseTelnetLine(info, (void *)sql[i]);
ret = smlParseTelnetLine(info, (void *)sql[i], strlen(sql[i]));
ASSERT_NE(ret, 0);
}
@ -653,7 +653,7 @@ TEST(testCase, smlParseTelnetLine_diff_json_type1_Test) {
int ret = TSDB_CODE_SUCCESS;
for (int i = 0; i < sizeof(sql) / sizeof(sql[0]); i++) {
ret = smlParseTelnetLine(info, (void *)sql[i]);
ret = smlParseTelnetLine(info, (void *)sql[i], strlen(sql[i]));
if (ret != TSDB_CODE_SUCCESS) break;
}
ASSERT_NE(ret, 0);
@ -688,7 +688,7 @@ TEST(testCase, smlParseTelnetLine_diff_json_type2_Test) {
};
int ret = TSDB_CODE_SUCCESS;
for (int i = 0; i < sizeof(sql) / sizeof(sql[0]); i++) {
ret = smlParseTelnetLine(info, (void *)sql[i]);
ret = smlParseTelnetLine(info, (void *)sql[i], strlen(sql[i]));
if (ret != TSDB_CODE_SUCCESS) break;
}
ASSERT_NE(ret, 0);
@ -1002,7 +1002,7 @@ TEST(testCase, sml_col_4096_Test) {
int ret = TSDB_CODE_SUCCESS;
for (int i = 0; i < sizeof(sql) / sizeof(sql[0]); i++) {
ret = smlParseInfluxLine(info, sql[i]);
ret = smlParseInfluxLine(info, sql[i], strlen(sql[i]));
if (ret != TSDB_CODE_SUCCESS) break;
}
ASSERT_NE(ret, 0);

20
source/dnode/mgmt/node_mgmt/src/dmEnv.c
View File

@ -51,6 +51,7 @@ static int32_t dmInitMonitor() {
static bool dmCheckDiskSpace() {
osUpdate();
// sufficiency
if (!osDataSpaceSufficient()) {
dWarn("free data disk size: %f GB, not sufficient, expected %f GB at least", (double)tsDataSpace.size.avail / 1024.0 / 1024.0 / 1024.0, (double)tsDataSpace.reserved / 1024.0 / 1024.0 / 1024.0);
}
@ -60,7 +61,24 @@ static bool dmCheckDiskSpace() {
if (!osTempSpaceSufficient()) {
dWarn("free temp disk size: %f GB, not sufficient, expected %f GB at least", (double)tsTempSpace.size.avail / 1024.0 / 1024.0 / 1024.0, (double)tsTempSpace.reserved / 1024.0 / 1024.0 / 1024.0);
}
return true;
// availability
bool ret = true;
if (!osDataSpaceAvailable()) {
dError("data disk space unavailable, i.e. %s", tsDataDir);
terrno = TSDB_CODE_VND_NO_DISKSPACE;
ret = false;
}
if (!osLogSpaceAvailable()) {
dError("log disk space unavailable, i.e. %s", tsLogDir);
terrno = TSDB_CODE_VND_NO_DISKSPACE;
ret = false;
}
if (!osTempSpaceAvailable()) {
dError("temp disk space unavailable, i.e. %s", tsTempDir);
terrno = TSDB_CODE_VND_NO_DISKSPACE;
ret = false;
}
return ret;
}
static bool dmCheckDataDirVersion() {

2
source/dnode/mgmt/test/sut/src/sut.cpp
View File

@ -43,9 +43,7 @@ void Testbase::InitLog(const char* path) {
}
void Testbase::Init(const char* path, int16_t port) {
#ifdef _TD_DARWIN_64
osDefaultInit();
#endif
tsServerPort = port;
strcpy(tsLocalFqdn, "localhost");
snprintf(tsLocalEp, TSDB_EP_LEN, "%s:%u", tsLocalFqdn, tsServerPort);

19
source/dnode/mnode/impl/src/mndMain.c
View File

@ -603,22 +603,27 @@ static int32_t mndCheckMnodeState(SRpcMsg *pMsg) {
return 0;
}
if (mndAcquireRpc(pMsg->info.node) == 0) return 0;
SMnode *pMnode = pMsg->info.node;
const char *role = syncGetMyRoleStr(pMnode->syncMgmt.sync);
bool restored = syncIsRestoreFinish(pMnode->syncMgmt.sync);
if (pMsg->msgType == TDMT_MND_MQ_TIMER || pMsg->msgType == TDMT_MND_TELEM_TIMER ||
pMsg->msgType == TDMT_MND_TRANS_TIMER || pMsg->msgType == TDMT_MND_TTL_TIMER ||
pMsg->msgType == TDMT_MND_UPTIME_TIMER) {
mTrace("timer not process since mnode restored:%d stopped:%d, sync restored:%d role:%s ", pMnode->restored,
pMnode->stopped, restored, role);
return -1;
}
SEpSet epSet = {0};
SMnode *pMnode = pMsg->info.node;
const STraceId *trace = &pMsg->info.traceId;
SEpSet epSet = {0};
mndGetMnodeEpSet(pMnode, &epSet);
const STraceId *trace = &pMsg->info.traceId;
mDebug(
"msg:%p, failed to check mnode state since %s, mnode restored:%d stopped:%d, sync restored:%d role:%s type:%s "
"numOfEps:%d inUse:%d",
pMsg, terrstr(), pMnode->restored, pMnode->stopped, syncIsRestoreFinish(pMnode->syncMgmt.sync),
syncGetMyRoleStr(pMnode->syncMgmt.sync), TMSG_INFO(pMsg->msgType), epSet.numOfEps, epSet.inUse);
"msg:%p, type:%s failed to process since %s, mnode restored:%d stopped:%d, sync restored:%d "
"role:%s, redirect numOfEps:%d inUse:%d",
pMsg, TMSG_INFO(pMsg->msgType), terrstr(), pMnode->restored, pMnode->stopped, restored, role, epSet.numOfEps,
epSet.inUse);
if (epSet.numOfEps > 0) {
for (int32_t i = 0; i < epSet.numOfEps; ++i) {

36
source/dnode/mnode/impl/src/mndMnode.c
View File

@ -394,7 +394,6 @@ static int32_t mndCreateMnode(SMnode *pMnode, SRpcMsg *pReq, SDnodeObj *pDnode,
if (mndSetCreateMnodeRedoActions(pMnode, pTrans, pDnode, &mnodeObj) != 0) goto _OVER;
if (mndSetCreateMnodeRedoLogs(pMnode, pTrans, &mnodeObj) != 0) goto _OVER;
if (mndSetCreateMnodeCommitLogs(pMnode, pTrans, &mnodeObj) != 0) goto _OVER;
if (mndTransAppendNullLog(pTrans) != 0) goto _OVER;
if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
code = 0;
@ -478,7 +477,6 @@ static int32_t mndSetDropMnodeCommitLogs(SMnode *pMnode, STrans *pTrans, SMnodeO
static int32_t mndSetDropMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnodeObj *pDnode, SMnodeObj *pObj) {
SSdb *pSdb = pMnode->pSdb;
void *pIter = NULL;
int32_t numOfReplicas = 0;
SDDropMnodeReq dropReq = {0};
SEpSet dropEpSet = {0};
@ -505,9 +503,8 @@ static int32_t mndSetDropMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnode
int32_t mndSetDropMnodeInfoToTrans(SMnode *pMnode, STrans *pTrans, SMnodeObj *pObj) {
if (pObj == NULL) return 0;
if (mndSetDropMnodeCommitLogs(pMnode, pTrans, pObj) != 0) return -1;
if (mndSetDropMnodeRedoActions(pMnode, pTrans, pObj->pDnode, pObj) != 0) return -1;
if (mndTransAppendNullLog(pTrans) != 0) return -1;
if (mndSetDropMnodeCommitLogs(pMnode, pTrans, pObj) != 0) return -1;
return 0;
}
@ -715,7 +712,8 @@ static void mndReloadSyncConfig(SMnode *pMnode) {
SMnodeObj *pObj = NULL;
ESdbStatus objStatus = 0;
void *pIter = NULL;
bool hasUpdatingMnode = false;
int32_t updatingMnodes = 0;
int32_t readyMnodes = 0;
SSyncCfg cfg = {.myIndex = -1};
while (1) {
@ -723,7 +721,11 @@ static void mndReloadSyncConfig(SMnode *pMnode) {
if (pIter == NULL) break;
if (objStatus == SDB_STATUS_CREATING || objStatus == SDB_STATUS_DROPPING) {
mInfo("vgId:1, has updating mnode:%d, status:%s", pObj->id, sdbStatusName(objStatus));
hasUpdatingMnode = true;
updatingMnodes++;
}
if (objStatus == SDB_STATUS_READY) {
mInfo("vgId:1, has ready mnode:%d, status:%s", pObj->id, sdbStatusName(objStatus));
readyMnodes++;
}
if (objStatus == SDB_STATUS_READY || objStatus == SDB_STATUS_CREATING) {
@ -739,18 +741,24 @@ static void mndReloadSyncConfig(SMnode *pMnode) {
sdbReleaseLock(pSdb, pObj, false);
}
if (readyMnodes <= 0 || updatingMnodes <= 0) {
mInfo("vgId:1, mnode sync not reconfig since readyMnodes:%d updatingMnodes:%d", readyMnodes, updatingMnodes);
return;
}
if (cfg.myIndex == -1) {
mInfo("vgId:1, mnode not reload since selfIndex is -1");
#if 1
mInfo("vgId:1, mnode sync not reconfig since selfIndex is -1");
#else
// cannot reconfig because the leader may fail to elect after reboot
mInfo("vgId:1, mnode sync not reconfig since selfIndex is -1, do sync stop oper");
syncStop(pMnode->syncMgmt.sync);
#endif
return;
}
if (!mndGetRestored(pMnode)) {
mInfo("vgId:1, mnode not reload since restore not finished");
return;
}
if (hasUpdatingMnode) {
mInfo("vgId:1, start to reload mnode sync, replica:%d myIndex:%d", cfg.replicaNum, cfg.myIndex);
if (updatingMnodes > 0) {
mInfo("vgId:1, mnode sync reconfig, replica:%d myIndex:%d", cfg.replicaNum, cfg.myIndex);
for (int32_t i = 0; i < cfg.replicaNum; ++i) {
SNodeInfo *pNode = &cfg.nodeInfo[i];
mInfo("vgId:1, index:%d, fqdn:%s port:%d", i, pNode->nodeFqdn, pNode->nodePort);

4
source/dnode/mnode/sdb/inc/sdb.h
View File

@ -403,6 +403,10 @@ const char *sdbStatusName(ESdbStatus status);
void sdbPrintOper(SSdb *pSdb, SSdbRow *pRow, const char *oper);
int32_t sdbGetIdFromRaw(SSdb *pSdb, SSdbRaw *pRaw);
void sdbWriteLock(SSdb *pSdb, int32_t type);
void sdbReadLock(SSdb *pSdb, int32_t type);
void sdbUnLock(SSdb *pSdb, int32_t type);
#ifdef __cplusplus
}
#endif

20
source/dnode/mnode/sdb/src/sdb.c
View File

@ -181,3 +181,23 @@ void sdbGetCommitInfo(SSdb *pSdb, int64_t *index, int64_t *term, int64_t *config
pSdb->applyIndex, pSdb->applyTerm, pSdb->applyConfig, *index, *term, *config);
#endif
}
void sdbWriteLock(SSdb *pSdb, int32_t type) {
TdThreadRwlock *pLock = &pSdb->locks[type];
// mTrace("sdb table:%d start write lock:%p", type, pLock);
taosThreadRwlockWrlock(pLock);
// mTrace("sdb table:%d stop write lock:%p", type, pLock);
}
void sdbReadLock(SSdb *pSdb, int32_t type) {
TdThreadRwlock *pLock = &pSdb->locks[type];
// mTrace("sdb table:%d start read lock:%p", type, pLock);
taosThreadRwlockRdlock(pLock);
// mTrace("sdb table:%d stop read lock:%p", type, pLock);
}
void sdbUnLock(SSdb *pSdb, int32_t type) {
TdThreadRwlock *pLock = &pSdb->locks[type];
// mTrace("sdb table:%d unlock:%p", type, pLock);
taosThreadRwlockUnlock(pLock);
}

7
source/dnode/mnode/sdb/src/sdbFile.c
View File

@ -363,9 +363,8 @@ static int32_t sdbWriteFileImp(SSdb *pSdb) {
mInfo("write %s to sdb file, total %d rows", sdbTableName(i), sdbGetSize(pSdb, i));
SHashObj *hash = pSdb->hashObjs[i];
TdThreadRwlock *pLock = &pSdb->locks[i];
taosThreadRwlockWrlock(pLock);
SHashObj *hash = pSdb->hashObjs[i];
sdbWriteLock(pSdb, i);
SSdbRow **ppRow = taosHashIterate(hash, NULL);
while (ppRow != NULL) {
@ -410,7 +409,7 @@ static int32_t sdbWriteFileImp(SSdb *pSdb) {
sdbFreeRaw(pRaw);
ppRow = taosHashIterate(hash, ppRow);
}
taosThreadRwlockUnlock(pLock);
sdbUnLock(pSdb, i);
}
if (code == 0) {

83
source/dnode/mnode/sdb/src/sdbHash.c
View File

@ -133,12 +133,12 @@ static int32_t sdbGetkeySize(SSdb *pSdb, ESdbType type, const void *pKey) {
}
static int32_t sdbInsertRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *pRow, int32_t keySize) {
TdThreadRwlock *pLock = &pSdb->locks[pRow->type];
taosThreadRwlockWrlock(pLock);
int32_t type = pRow->type;
sdbWriteLock(pSdb, type);
SSdbRow *pOldRow = taosHashGet(hash, pRow->pObj, keySize);
if (pOldRow != NULL) {
taosThreadRwlockUnlock(pLock);
sdbUnLock(pSdb, type);
sdbFreeRow(pSdb, pRow, false);
terrno = TSDB_CODE_SDB_OBJ_ALREADY_THERE;
return terrno;
@ -149,7 +149,7 @@ static int32_t sdbInsertRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
sdbPrintOper(pSdb, pRow, "insert");
if (taosHashPut(hash, pRow->pObj, keySize, &pRow, sizeof(void *)) != 0) {
taosThreadRwlockUnlock(pLock);
sdbUnLock(pSdb, type);
sdbFreeRow(pSdb, pRow, false);
terrno = TSDB_CODE_OUT_OF_MEMORY;
return terrno;
@ -164,12 +164,12 @@ static int32_t sdbInsertRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
taosHashRemove(hash, pRow->pObj, keySize);
sdbFreeRow(pSdb, pRow, false);
terrno = code;
taosThreadRwlockUnlock(pLock);
sdbUnLock(pSdb, type);
return terrno;
}
}
taosThreadRwlockUnlock(pLock);
sdbUnLock(pSdb, type);
if (pSdb->keyTypes[pRow->type] == SDB_KEY_INT32) {
pSdb->maxId[pRow->type] = TMAX(pSdb->maxId[pRow->type], *((int32_t *)pRow->pObj));
@ -183,26 +183,27 @@ static int32_t sdbInsertRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
}
static int32_t sdbUpdateRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *pNewRow, int32_t keySize) {
TdThreadRwlock *pLock = &pSdb->locks[pNewRow->type];
taosThreadRwlockWrlock(pLock);
int32_t type = pNewRow->type;
sdbWriteLock(pSdb, type);
SSdbRow **ppOldRow = taosHashGet(hash, pNewRow->pObj, keySize);
if (ppOldRow == NULL || *ppOldRow == NULL) {
taosThreadRwlockUnlock(pLock);
sdbUnLock(pSdb, type);
return sdbInsertRow(pSdb, hash, pRaw, pNewRow, keySize);
}
SSdbRow *pOldRow = *ppOldRow;
pOldRow->status = pRaw->status;
sdbPrintOper(pSdb, pOldRow, "update");
sdbUnLock(pSdb, type);
int32_t code = 0;
SdbUpdateFp updateFp = pSdb->updateFps[pNewRow->type];
SdbUpdateFp updateFp = pSdb->updateFps[type];
if (updateFp != NULL) {
code = (*updateFp)(pSdb, pOldRow->pObj, pNewRow->pObj);
}
taosThreadRwlockUnlock(pLock);
// sdbUnLock(pSdb, type);
sdbFreeRow(pSdb, pNewRow, false);
pSdb->tableVer[pOldRow->type]++;
@ -210,12 +211,12 @@ static int32_t sdbUpdateRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
}
static int32_t sdbDeleteRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *pRow, int32_t keySize) {
TdThreadRwlock *pLock = &pSdb->locks[pRow->type];
taosThreadRwlockWrlock(pLock);
int32_t type = pRow->type;
sdbWriteLock(pSdb, type);
SSdbRow **ppOldRow = taosHashGet(hash, pRow->pObj, keySize);
if (ppOldRow == NULL || *ppOldRow == NULL) {
taosThreadRwlockUnlock(pLock);
sdbUnLock(pSdb, type);
sdbFreeRow(pSdb, pRow, false);
terrno = TSDB_CODE_SDB_OBJ_NOT_THERE;
return terrno;
@ -228,7 +229,7 @@ static int32_t sdbDeleteRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
taosHashRemove(hash, pOldRow->pObj, keySize);
pSdb->tableVer[pOldRow->type]++;
taosThreadRwlockUnlock(pLock);
sdbUnLock(pSdb, type);
sdbFreeRow(pSdb, pRow, false);
@ -282,12 +283,11 @@ void *sdbAcquire(SSdb *pSdb, ESdbType type, const void *pKey) {
void *pRet = NULL;
int32_t keySize = sdbGetkeySize(pSdb, type, pKey);
TdThreadRwlock *pLock = &pSdb->locks[type];
taosThreadRwlockRdlock(pLock);
sdbReadLock(pSdb, type);
SSdbRow **ppRow = taosHashGet(hash, pKey, keySize);
if (ppRow == NULL || *ppRow == NULL) {
taosThreadRwlockUnlock(pLock);
sdbUnLock(pSdb, type);
terrno = TSDB_CODE_SDB_OBJ_NOT_THERE;
return NULL;
}
@ -310,13 +310,13 @@ void *sdbAcquire(SSdb *pSdb, ESdbType type, const void *pKey) {
break;
}
taosThreadRwlockUnlock(pLock);
sdbUnLock(pSdb, type);
return pRet;
}
static void sdbCheckRow(SSdb *pSdb, SSdbRow *pRow) {
TdThreadRwlock *pLock = &pSdb->locks[pRow->type];
taosThreadRwlockWrlock(pLock);
int32_t type = pRow->type;
sdbWriteLock(pSdb, type);
int32_t ref = atomic_sub_fetch_32(&pRow->refCount, 1);
sdbPrintOper(pSdb, pRow, "check");
@ -324,7 +324,7 @@ static void sdbCheckRow(SSdb *pSdb, SSdbRow *pRow) {
sdbFreeRow(pSdb, pRow, true);
}
taosThreadRwlockUnlock(pLock);
sdbUnLock(pSdb, type);
}
void sdbReleaseLock(SSdb *pSdb, void *pObj, bool lock) {
@ -333,9 +333,9 @@ void sdbReleaseLock(SSdb *pSdb, void *pObj, bool lock) {
SSdbRow *pRow = (SSdbRow *)((char *)pObj - sizeof(SSdbRow));
if (pRow->type >= SDB_MAX) return;
TdThreadRwlock *pLock = &pSdb->locks[pRow->type];
int32_t type = pRow->type;
if (lock) {
taosThreadRwlockWrlock(pLock);
sdbWriteLock(pSdb, type);
}
int32_t ref = atomic_sub_fetch_32(&pRow->refCount, 1);
@ -345,7 +345,7 @@ void sdbReleaseLock(SSdb *pSdb, void *pObj, bool lock) {
}
if (lock) {
taosThreadRwlockUnlock(pLock);
sdbUnLock(pSdb, type);
}
}
@ -357,8 +357,7 @@ void *sdbFetch(SSdb *pSdb, ESdbType type, void *pIter, void **ppObj) {
SHashObj *hash = sdbGetHash(pSdb, type);
if (hash == NULL) return NULL;
TdThreadRwlock *pLock = &pSdb->locks[type];
taosThreadRwlockRdlock(pLock);
sdbReadLock(pSdb, type);
SSdbRow **ppRow = taosHashIterate(hash, pIter);
while (ppRow != NULL) {
@ -373,7 +372,7 @@ void *sdbFetch(SSdb *pSdb, ESdbType type, void *pIter, void **ppObj) {
*ppObj = pRow->pObj;
break;
}
taosThreadRwlockUnlock(pLock);
sdbUnLock(pSdb, type);
return ppRow;
}
@ -384,9 +383,8 @@ void *sdbFetchAll(SSdb *pSdb, ESdbType type, void *pIter, void **ppObj, ESdbStat
SHashObj *hash = sdbGetHash(pSdb, type);
if (hash == NULL) return NULL;
TdThreadRwlock *pLock = &pSdb->locks[type];
if (lock) {
taosThreadRwlockRdlock(pLock);
sdbReadLock(pSdb, type);
}
SSdbRow **ppRow = taosHashIterate(hash, pIter);
@ -404,7 +402,7 @@ void *sdbFetchAll(SSdb *pSdb, ESdbType type, void *pIter, void **ppObj, ESdbStat
break;
}
if (lock) {
taosThreadRwlockUnlock(pLock);
sdbUnLock(pSdb, type);
}
return ppRow;
@ -416,18 +414,17 @@ void sdbCancelFetch(SSdb *pSdb, void *pIter) {
SHashObj *hash = sdbGetHash(pSdb, pRow->type);
if (hash == NULL) return;
TdThreadRwlock *pLock = &pSdb->locks[pRow->type];
taosThreadRwlockRdlock(pLock);
int32_t type = pRow->type;
sdbReadLock(pSdb, type);
taosHashCancelIterate(hash, pIter);
taosThreadRwlockUnlock(pLock);
sdbUnLock(pSdb, type);
}
void sdbTraverse(SSdb *pSdb, ESdbType type, sdbTraverseFp fp, void *p1, void *p2, void *p3) {
SHashObj *hash = sdbGetHash(pSdb, type);
if (hash == NULL) return;
TdThreadRwlock *pLock = &pSdb->locks[type];
taosThreadRwlockRdlock(pLock);
sdbReadLock(pSdb, type);
SSdbRow **ppRow = taosHashIterate(hash, NULL);
while (ppRow != NULL) {
@ -443,17 +440,16 @@ void sdbTraverse(SSdb *pSdb, ESdbType type, sdbTraverseFp fp, void *p1, void *p2
ppRow = taosHashIterate(hash, ppRow);
}
taosThreadRwlockUnlock(pLock);
sdbUnLock(pSdb, type);
}
int32_t sdbGetSize(SSdb *pSdb, ESdbType type) {
SHashObj *hash = sdbGetHash(pSdb, type);
if (hash == NULL) return 0;
TdThreadRwlock *pLock = &pSdb->locks[type];
taosThreadRwlockRdlock(pLock);
sdbReadLock(pSdb, type);
int32_t size = taosHashGetSize(hash);
taosThreadRwlockUnlock(pLock);
sdbUnLock(pSdb, type);
return size;
}
@ -465,9 +461,7 @@ int32_t sdbGetMaxId(SSdb *pSdb, ESdbType type) {
if (pSdb->keyTypes[type] != SDB_KEY_INT32) return -1;
int32_t maxId = 0;
TdThreadRwlock *pLock = &pSdb->locks[type];
taosThreadRwlockRdlock(pLock);
sdbReadLock(pSdb, type);
SSdbRow **ppRow = taosHashIterate(hash, NULL);
while (ppRow != NULL) {
@ -477,8 +471,7 @@ int32_t sdbGetMaxId(SSdb *pSdb, ESdbType type) {
ppRow = taosHashIterate(hash, ppRow);
}
taosThreadRwlockUnlock(pLock);
sdbUnLock(pSdb, type);
maxId = TMAX(maxId, pSdb->maxId[type]);
return maxId + 1;
}

2
source/dnode/vnode/src/inc/sma.h
View File

@ -258,6 +258,7 @@ enum {
TD_FTYPE_RSMA_QTASKINFO = 0,
};
#if 0
struct STFile {
uint8_t state;
STFInfo info;
@ -287,6 +288,7 @@ int32_t tdUpdateTFileHeader(STFile *pTFile);
void tdUpdateTFileMagic(STFile *pTFile, void *pCksm);
void tdCloseTFile(STFile *pTFile);
void tdDestroyTFile(STFile *pTFile);
#endif
void tdGetVndFileName(int32_t vgId, const char *pdname, const char *dname, const char *fname, int64_t version,
char *outputName);

32
source/dnode/vnode/src/meta/metaQuery.c
View File

@ -1059,7 +1059,7 @@ int32_t metaFilterTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
if (param->val == NULL) {
metaError("vgId:%d, failed to filter NULL data", TD_VID(pMeta->pVnode));
return -1;
goto END;
} else {
if (IS_VAR_DATA_TYPE(param->type)) {
tagData = varDataVal(param->val);
@ -1111,27 +1111,25 @@ int32_t metaFilterTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
break;
}
}
if (p->suid != pKey->suid) {
if (p == NULL || p->suid != pKey->suid) {
break;
}
first = false;
if (p != NULL) {
int32_t cmp = (*param->filterFunc)(p->data, pKey->data, pKey->type);
if (cmp == 0) {
// match
tb_uid_t tuid = 0;
if (IS_VAR_DATA_TYPE(pKey->type)) {
tuid = *(tb_uid_t *)(p->data + varDataTLen(p->data));
} else {
tuid = *(tb_uid_t *)(p->data + tDataTypes[pCursor->type].bytes);
}
taosArrayPush(pUids, &tuid);
} else if (cmp == 1) {
// not match but should continue to iter
int32_t cmp = (*param->filterFunc)(p->data, pKey->data, pKey->type);
if (cmp == 0) {
// match
tb_uid_t tuid = 0;
if (IS_VAR_DATA_TYPE(pKey->type)) {
tuid = *(tb_uid_t *)(p->data + varDataTLen(p->data));
} else {
// not match and no more result
break;
tuid = *(tb_uid_t *)(p->data + tDataTypes[pCursor->type].bytes);
}
taosArrayPush(pUids, &tuid);
} else if (cmp == 1) {
// not match but should continue to iter
} else {
// not match and no more result
break;
}
valid = param->reverse ? tdbTbcMoveToPrev(pCursor->pCur) : tdbTbcMoveToNext(pCursor->pCur);
if (valid < 0) {

7
source/dnode/vnode/src/meta/metaTable.c
View File

@ -51,7 +51,7 @@ static int metaUpdateMetaRsp(tb_uid_t uid, char *tbName, SSchemaWrapper *pSchema
return -1;
}
strncpy(pMetaRsp->tbName, tbName, TSDB_TABLE_NAME_LEN);
tstrncpy(pMetaRsp->tbName, tbName, TSDB_TABLE_NAME_LEN);
pMetaRsp->numOfColumns = pSchema->nCols;
pMetaRsp->tableType = TSDB_NORMAL_TABLE;
pMetaRsp->sversion = pSchema->version;
@ -116,9 +116,10 @@ static int metaSaveJsonVarToIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry, const
indexMultiTermAdd(terms, term);
}
}
taosArrayDestroy(pTagVals);
indexJsonPut(pMeta->pTagIvtIdx, terms, tuid);
indexMultiTermDestroy(terms);
taosArrayDestroy(pTagVals);
#endif
return 0;
}
@ -159,6 +160,7 @@ int metaDelJsonVarFromIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry, const SSche
memcpy(val, (uint16_t *)&len, VARSTR_HEADER_SIZE);
type = TSDB_DATA_TYPE_VARCHAR;
term = indexTermCreate(suid, DEL_VALUE, type, key, nKey, val, len);
taosMemoryFree(val);
} else if (pTagVal->nData == 0) {
term = indexTermCreate(suid, DEL_VALUE, TSDB_DATA_TYPE_VARCHAR, key, nKey, pTagVal->pData, 0);
}
@ -177,6 +179,7 @@ int metaDelJsonVarFromIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry, const SSche
}
indexJsonPut(pMeta->pTagIvtIdx, terms, tuid);
indexMultiTermDestroy(terms);
taosArrayDestroy(pTagVals);
#endif
return 0;
}

8
source/dnode/vnode/src/sma/smaCommit.c
View File

@ -17,14 +17,17 @@
extern SSmaMgmt smaMgmt;
#if 0
static int32_t tdProcessRSmaSyncPreCommitImpl(SSma *pSma);
static int32_t tdProcessRSmaSyncCommitImpl(SSma *pSma);
static int32_t tdProcessRSmaSyncPostCommitImpl(SSma *pSma);
#endif
static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma);
static int32_t tdProcessRSmaAsyncCommitImpl(SSma *pSma);
static int32_t tdProcessRSmaAsyncPostCommitImpl(SSma *pSma);
static int32_t tdUpdateQTaskInfoFiles(SSma *pSma, SRSmaStat *pRSmaStat);
#if 0
/**
* @brief Only applicable to Rollup SMA
*
@ -48,6 +51,7 @@ int32_t smaSyncCommit(SSma *pSma) { return tdProcessRSmaSyncCommitImpl(pSma); }
* @return int32_t
*/
int32_t smaSyncPostCommit(SSma *pSma) { return tdProcessRSmaSyncPostCommitImpl(pSma); }
#endif
/**
* @brief Only applicable to Rollup SMA
@ -108,6 +112,7 @@ int32_t smaBegin(SSma *pSma) {
return TSDB_CODE_SUCCESS;
}
#if 0
/**
* @brief pre-commit for rollup sma(sync commit).
* 1) set trigger stat of rsma timer TASK_TRIGGER_STAT_PAUSED.
@ -169,6 +174,7 @@ static int32_t tdProcessRSmaSyncCommitImpl(SSma *pSma) {
#endif
return TSDB_CODE_SUCCESS;
}
#endif
// SQTaskFile ======================================================
@ -230,6 +236,7 @@ static int32_t tdUpdateQTaskInfoFiles(SSma *pSma, SRSmaStat *pStat) {
return TSDB_CODE_SUCCESS;
}
#if 0
/**
* @brief post-commit for rollup sma
* 1) clean up the outdated qtaskinfo files
@ -249,6 +256,7 @@ static int32_t tdProcessRSmaSyncPostCommitImpl(SSma *pSma) {
return TSDB_CODE_SUCCESS;
}
#endif
/**
* @brief Rsma async commit implementation(only do some necessary light weighted task)

420
source/dnode/vnode/src/sma/smaRollup.c
View File

@ -15,8 +15,6 @@
#include "sma.h"
#define RSMA_QTASKINFO_BUFSIZE (32768) // size
#define RSMA_QTASKINFO_HEAD_LEN (sizeof(int32_t) + sizeof(int8_t) + sizeof(int64_t)) // len + type + suid
#define RSMA_QTASKEXEC_SMOOTH_SIZE (100) // cnt
#define RSMA_SUBMIT_BATCH_SIZE (1024) // cnt
#define RSMA_FETCH_DELAY_MAX (120000) // ms
@ -48,23 +46,10 @@ static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SR
static void tdRSmaFetchTrigger(void *param, void *tmrId);
static int32_t tdRSmaInfoClone(SSma *pSma, SRSmaInfo *pInfo);
static void tdRSmaQTaskInfoFree(qTaskInfo_t *taskHandle, int32_t vgId, int32_t level);
static int32_t tdRSmaQTaskInfoIterInit(SRSmaQTaskInfoIter *pIter, STFile *pTFile);
static int32_t tdRSmaQTaskInfoIterNextBlock(SRSmaQTaskInfoIter *pIter, bool *isFinish);
static int32_t tdRSmaQTaskInfoRestore(SSma *pSma, int8_t type, SRSmaQTaskInfoIter *pIter);
static int32_t tdRSmaQTaskInfoItemRestore(SSma *pSma, const SRSmaQTaskInfoItem *infoItem);
static int32_t tdRSmaRestoreQTaskInfoInit(SSma *pSma, int64_t *nTables);
static int32_t tdRSmaRestoreQTaskInfoReload(SSma *pSma, int8_t type, int64_t qTaskFileVer);
static int32_t tdRSmaRestoreTSDataReload(SSma *pSma);
static SRSmaInfo *tdGetRSmaInfoByItem(SRSmaInfoItem *pItem) {
// adapt accordingly if definition of SRSmaInfo update
SRSmaInfo *pResult = NULL;
ASSERT(pItem->level == TSDB_RETENTION_L1 || pItem->level == TSDB_RETENTION_L2);
pResult = (SRSmaInfo *)POINTER_SHIFT(pItem, -(sizeof(SRSmaInfoItem) * (pItem->level - 1) + RSMA_INFO_HEAD_LEN));
ASSERT(pResult->pTSchema->numOfCols > 1);
return pResult;
}
struct SRSmaQTaskInfoItem {
int32_t len;
int8_t type;
@ -104,12 +89,6 @@ void tdRSmaQTaskInfoGetFullPathEx(int32_t vgId, tb_uid_t suid, int8_t level, con
snprintf(outputName + rsmaLen, TSDB_FILENAME_LEN - rsmaLen, "%" PRIi64 "%s%" PRIi8, suid, TD_DIRSEP, level);
}
static FORCE_INLINE int32_t tdRSmaQTaskInfoContLen(int32_t lenWithHead) {
return lenWithHead - RSMA_QTASKINFO_HEAD_LEN;
}
static FORCE_INLINE void tdRSmaQTaskInfoIterDestroy(SRSmaQTaskInfoIter *pIter) { taosMemoryFreeClear(pIter->pBuf); }
static void tdRSmaQTaskInfoFree(qTaskInfo_t *taskHandle, int32_t vgId, int32_t level) {
// Note: free/kill may in RC
if (!taskHandle || !(*taskHandle)) return;
@ -803,6 +782,7 @@ static int32_t tdExecuteRSmaImplAsync(SSma *pSma, const void *pMsg, int32_t inpu
return TSDB_CODE_SUCCESS;
}
#if 0
static int32_t tdRsmaPrintSubmitReq(SSma *pSma, SSubmitReq *pReq) {
SSubmitMsgIter msgIter = {0};
SSubmitBlkIter blkIter = {0};
@ -820,6 +800,7 @@ static int32_t tdRsmaPrintSubmitReq(SSma *pSma, SSubmitReq *pReq) {
}
return 0;
}
#endif
/**
* @brief sync mode
@ -1189,65 +1170,6 @@ _err:
return TSDB_CODE_FAILED;
}
static int32_t tdRSmaRestoreQTaskInfoReload(SSma *pSma, int8_t type, int64_t qTaskFileVer) {
SVnode *pVnode = pSma->pVnode;
STFile tFile = {0};
char qTaskInfoFName[TSDB_FILENAME_LEN] = {0};
tdRSmaQTaskInfoGetFileName(TD_VID(pVnode), qTaskFileVer, qTaskInfoFName);
if (tdInitTFile(&tFile, tfsGetPrimaryPath(pVnode->pTfs), qTaskInfoFName) < 0) {
goto _err;
}
if (!taosCheckExistFile(TD_TFILE_FULL_NAME(&tFile))) {
if (qTaskFileVer > 0) {
smaWarn("vgId:%d, restore rsma task %" PRIi8 " for version %" PRIi64 ", not start as %s not exist",
TD_VID(pVnode), type, qTaskFileVer, TD_TFILE_FULL_NAME(&tFile));
} else {
smaDebug("vgId:%d, restore rsma task %" PRIi8 " for version %" PRIi64 ", no need as %s not exist", TD_VID(pVnode),
type, qTaskFileVer, TD_TFILE_FULL_NAME(&tFile));
}
return TSDB_CODE_SUCCESS;
}
if (tdOpenTFile(&tFile, TD_FILE_READ) < 0) {
goto _err;
}
STFInfo tFileInfo = {0};
if (tdLoadTFileHeader(&tFile, &tFileInfo) < 0) {
goto _err;
}
SRSmaQTaskInfoIter fIter = {0};
if (tdRSmaQTaskInfoIterInit(&fIter, &tFile) < 0) {
tdRSmaQTaskInfoIterDestroy(&fIter);
tdCloseTFile(&tFile);
tdDestroyTFile(&tFile);
goto _err;
}
if (tdRSmaQTaskInfoRestore(pSma, type, &fIter) < 0) {
tdRSmaQTaskInfoIterDestroy(&fIter);
tdCloseTFile(&tFile);
tdDestroyTFile(&tFile);
goto _err;
}
tdRSmaQTaskInfoIterDestroy(&fIter);
tdCloseTFile(&tFile);
tdDestroyTFile(&tFile);
// restored successfully from committed or sync
smaInfo("vgId:%d, restore rsma task %" PRIi8 " for version %" PRIi64 ", qtaskinfo reload succeed", TD_VID(pVnode),
type, qTaskFileVer);
return TSDB_CODE_SUCCESS;
_err:
smaError("vgId:%d, restore rsma task %" PRIi8 " for version %" PRIi64 ", qtaskinfo reload failed since %s",
TD_VID(pVnode), type, qTaskFileVer, terrstr());
return TSDB_CODE_FAILED;
}
/**
* @brief reload ts data from checkpoint
*
@ -1270,19 +1192,12 @@ int32_t tdRSmaProcessRestoreImpl(SSma *pSma, int8_t type, int64_t qtaskFileVer)
return TSDB_CODE_SUCCESS;
}
#if 0
// step 2: retrieve qtaskinfo items from the persistence file(rsma/qtaskinfo) and restore
if (tdRSmaRestoreQTaskInfoReload(pSma, type, qtaskFileVer) < 0) {
goto _err;
}
#endif
// step 3: reload ts data from checkpoint
// step 2: reload ts data from checkpoint
if (tdRSmaRestoreTSDataReload(pSma) < 0) {
goto _err;
}
// step 4: open SRSmaFS for qTaskFiles
// step 3: open SRSmaFS for qTaskFiles
if (tdRSmaFSOpen(pSma, qtaskFileVer) < 0) {
goto _err;
}
@ -1295,191 +1210,6 @@ _err:
return TSDB_CODE_FAILED;
}
/**
* @brief Restore from SRSmaQTaskInfoItem
*
* @param pSma
* @param pItem
* @return int32_t
*/
static int32_t tdRSmaQTaskInfoItemRestore(SSma *pSma, const SRSmaQTaskInfoItem *pItem) {
SRSmaInfo *pRSmaInfo = NULL;
void *qTaskInfo = NULL;
pRSmaInfo = tdAcquireRSmaInfoBySuid(pSma, pItem->suid);
if (!pRSmaInfo) {
smaDebug("vgId:%d, no restore as no rsma info for table:%" PRIu64, SMA_VID(pSma), pItem->suid);
return TSDB_CODE_SUCCESS;
}
if (pItem->type == TSDB_RETENTION_L1) {
qTaskInfo = RSMA_INFO_QTASK(pRSmaInfo, 0);
} else if (pItem->type == TSDB_RETENTION_L2) {
qTaskInfo = RSMA_INFO_QTASK(pRSmaInfo, 1);
} else {
ASSERT(0);
}
if (!qTaskInfo) {
tdReleaseRSmaInfo(pSma, pRSmaInfo);
smaDebug("vgId:%d, no restore as NULL rsma qTaskInfo for table:%" PRIu64, SMA_VID(pSma), pItem->suid);
return TSDB_CODE_SUCCESS;
}
if (qDeserializeTaskStatus(qTaskInfo, pItem->qTaskInfo, pItem->len) < 0) {
tdReleaseRSmaInfo(pSma, pRSmaInfo);
smaError("vgId:%d, restore rsma task failed for table:%" PRIi64 " level %d since %s", SMA_VID(pSma), pItem->suid,
pItem->type, terrstr());
return TSDB_CODE_FAILED;
}
smaDebug("vgId:%d, restore rsma task success for table:%" PRIi64 " level %d", SMA_VID(pSma), pItem->suid,
pItem->type);
tdReleaseRSmaInfo(pSma, pRSmaInfo);
return TSDB_CODE_SUCCESS;
}
static int32_t tdRSmaQTaskInfoIterInit(SRSmaQTaskInfoIter *pIter, STFile *pTFile) {
memset(pIter, 0, sizeof(*pIter));
pIter->pTFile = pTFile;
pIter->offset = TD_FILE_HEAD_SIZE;
if (tdGetTFileSize(pTFile, &pIter->fsize) < 0) {
return TSDB_CODE_FAILED;
}
if ((pIter->fsize - TD_FILE_HEAD_SIZE) < RSMA_QTASKINFO_BUFSIZE) {
pIter->nAlloc = pIter->fsize - TD_FILE_HEAD_SIZE;
} else {
pIter->nAlloc = RSMA_QTASKINFO_BUFSIZE;
}
if (pIter->nAlloc < TD_FILE_HEAD_SIZE) {
pIter->nAlloc = TD_FILE_HEAD_SIZE;
}
pIter->pBuf = taosMemoryMalloc(pIter->nAlloc);
if (!pIter->pBuf) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return TSDB_CODE_FAILED;
}
pIter->qBuf = pIter->pBuf;
return TSDB_CODE_SUCCESS;
}
static int32_t tdRSmaQTaskInfoIterNextBlock(SRSmaQTaskInfoIter *pIter, bool *isFinish) {
STFile *pTFile = pIter->pTFile;
int64_t nBytes = RSMA_QTASKINFO_BUFSIZE;
if (pIter->offset >= pIter->fsize) {
*isFinish = true;
return TSDB_CODE_SUCCESS;
}
if ((pIter->fsize - pIter->offset) < RSMA_QTASKINFO_BUFSIZE) {
nBytes = pIter->fsize - pIter->offset;
}
if (tdSeekTFile(pTFile, pIter->offset, SEEK_SET) < 0) {
return TSDB_CODE_FAILED;
}
if (tdReadTFile(pTFile, pIter->pBuf, nBytes) != nBytes) {
return TSDB_CODE_FAILED;
}
int32_t infoLen = 0;
taosDecodeFixedI32(pIter->pBuf, &infoLen);
if (infoLen > nBytes) {
if (infoLen <= RSMA_QTASKINFO_BUFSIZE) {
terrno = TSDB_CODE_RSMA_FILE_CORRUPTED;
smaError("iterate rsma qtaskinfo file %s failed since %s", TD_TFILE_FULL_NAME(pIter->pTFile), terrstr());
return TSDB_CODE_FAILED;
}
if (pIter->nAlloc < infoLen) {
pIter->nAlloc = infoLen;
void *pBuf = taosMemoryRealloc(pIter->pBuf, infoLen);
if (!pBuf) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return TSDB_CODE_FAILED;
}
pIter->pBuf = pBuf;
}
nBytes = infoLen;
if (tdSeekTFile(pTFile, pIter->offset, SEEK_SET) < 0) {
return TSDB_CODE_FAILED;
}
if (tdReadTFile(pTFile, pIter->pBuf, nBytes) != nBytes) {
return TSDB_CODE_FAILED;
}
}
pIter->qBuf = pIter->pBuf;
pIter->offset += nBytes;
pIter->nBytes = nBytes;
pIter->nBufPos = 0;
return TSDB_CODE_SUCCESS;
}
static int32_t tdRSmaQTaskInfoRestore(SSma *pSma, int8_t type, SRSmaQTaskInfoIter *pIter) {
while (1) {
// block iter
bool isFinish = false;
if (tdRSmaQTaskInfoIterNextBlock(pIter, &isFinish) < 0) {
return TSDB_CODE_FAILED;
}
if (isFinish) {
return TSDB_CODE_SUCCESS;
}
// consume the block
int32_t qTaskInfoLenWithHead = 0;
pIter->qBuf = taosDecodeFixedI32(pIter->qBuf, &qTaskInfoLenWithHead);
if (qTaskInfoLenWithHead < RSMA_QTASKINFO_HEAD_LEN) {
terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
smaError("vgId:%d, restore rsma task %" PRIi8 " from qtaskinfo file %s failed since %s", SMA_VID(pSma), type,
TD_TFILE_FULL_NAME(pIter->pTFile), terrstr());
return TSDB_CODE_FAILED;
}
while (1) {
if ((pIter->nBufPos + qTaskInfoLenWithHead) <= pIter->nBytes) {
SRSmaQTaskInfoItem infoItem = {0};
pIter->qBuf = taosDecodeFixedI8(pIter->qBuf, &infoItem.type);
pIter->qBuf = taosDecodeFixedI64(pIter->qBuf, &infoItem.suid);
infoItem.qTaskInfo = pIter->qBuf;
infoItem.len = tdRSmaQTaskInfoContLen(qTaskInfoLenWithHead);
// do the restore job
smaDebug("vgId:%d, restore rsma task %" PRIi8 " from qtaskinfo file %s offset:%" PRIi64 "\n", SMA_VID(pSma),
type, TD_TFILE_FULL_NAME(pIter->pTFile), pIter->offset - pIter->nBytes + pIter->nBufPos);
tdRSmaQTaskInfoItemRestore(pSma, &infoItem);
pIter->qBuf = POINTER_SHIFT(pIter->qBuf, infoItem.len);
pIter->nBufPos += qTaskInfoLenWithHead;
if ((pIter->nBufPos + RSMA_QTASKINFO_HEAD_LEN) >= pIter->nBytes) {
// prepare and load next block in the file
pIter->offset -= (pIter->nBytes - pIter->nBufPos);
break;
}
pIter->qBuf = taosDecodeFixedI32(pIter->qBuf, &qTaskInfoLenWithHead);
continue;
}
// prepare and load next block in the file
pIter->offset -= (pIter->nBytes - pIter->nBufPos);
break;
}
}
return TSDB_CODE_SUCCESS;
}
int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat, SHashObj *pInfoHash) {
SSma *pSma = pRSmaStat->pSma;
SVnode *pVnode = pSma->pVnode;
@ -1523,148 +1253,6 @@ _err:
return TSDB_CODE_FAILED;
}
#if 0
int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat, SHashObj *pInfoHash) {
SSma *pSma = pRSmaStat->pSma;
SVnode *pVnode = pSma->pVnode;
int32_t vid = SMA_VID(pSma);
int64_t toffset = 0;
bool isFileCreated = false;
if (taosHashGetSize(pInfoHash) <= 0) {
return TSDB_CODE_SUCCESS;
}
void *infoHash = taosHashIterate(pInfoHash, NULL);
if (!infoHash) {
return TSDB_CODE_SUCCESS;
}
int64_t fsMaxVer = tdRSmaFSMaxVer(pSma, pRSmaStat);
if (pRSmaStat->commitAppliedVer <= fsMaxVer) {
smaDebug("vgId:%d, rsma persist, no need as applied %" PRIi64 " not larger than fsMaxVer %" PRIi64, vid,
pRSmaStat->commitAppliedVer, fsMaxVer);
return TSDB_CODE_SUCCESS;
}
STFile tFile = {0};
#if 0
if (pRSmaStat->commitAppliedVer > 0) {
char qTaskInfoFName[TSDB_FILENAME_LEN];
tdRSmaQTaskInfoGetFileName(vid, pRSmaStat->commitAppliedVer, qTaskInfoFName);
if (tdInitTFile(&tFile, tfsGetPrimaryPath(pVnode->pTfs), qTaskInfoFName) < 0) {
smaError("vgId:%d, rsma persist, init %s failed since %s", vid, qTaskInfoFName, terrstr());
goto _err;
}
if (tdCreateTFile(&tFile, true, TD_FTYPE_RSMA_QTASKINFO) < 0) {
smaError("vgId:%d, rsma persist, create %s failed since %s", vid, TD_TFILE_FULL_NAME(&tFile), terrstr());
goto _err;
}
smaDebug("vgId:%d, rsma, serialize qTaskInfo, file %s created", vid, TD_TFILE_FULL_NAME(&tFile));
isFileCreated = true;
}
#endif
while (infoHash) {
SRSmaInfo *pRSmaInfo = *(SRSmaInfo **)infoHash;
if (RSMA_INFO_IS_DEL(pRSmaInfo)) {
infoHash = taosHashIterate(pInfoHash, infoHash);
continue;
}
for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
#if 0
qTaskInfo_t taskInfo = RSMA_INFO_IQTASK(pRSmaInfo, i);
#endif
qTaskInfo_t taskInfo = RSMA_INFO_QTASK(pRSmaInfo, i);
if (!taskInfo) {
smaDebug("vgId:%d, rsma, table %" PRIi64 " level %d qTaskInfo is NULL", vid, pRSmaInfo->suid, i + 1);
continue;
}
char *pOutput = NULL;
int32_t len = 0;
int8_t type = (int8_t)(i + 1);
if (qSerializeTaskStatus(taskInfo, &pOutput, &len) < 0) {
smaError("vgId:%d, rsma, table %" PRIi64 " level %d serialize qTaskInfo failed since %s", vid, pRSmaInfo->suid,
i + 1, terrstr());
goto _err;
}
if (!pOutput || len <= 0) {
smaDebug("vgId:%d, rsma, table %" PRIi64
" level %d serialize qTaskInfo success but no output(len %d), not persist",
vid, pRSmaInfo->suid, i + 1, len);
taosMemoryFreeClear(pOutput);
continue;
}
smaDebug("vgId:%d, rsma, table %" PRIi64 " level %d serialize qTaskInfo success with len %d, need persist", vid,
pRSmaInfo->suid, i + 1, len);
if (!isFileCreated) {
char qTaskInfoFName[TSDB_FILENAME_LEN];
tdRSmaQTaskInfoGetFileName(vid, pRSmaStat->commitAppliedVer, qTaskInfoFName);
if (tdInitTFile(&tFile, tfsGetPrimaryPath(pVnode->pTfs), qTaskInfoFName) < 0) {
smaError("vgId:%d, rsma persist, init %s failed since %s", vid, qTaskInfoFName, terrstr());
goto _err;
}
if (tdCreateTFile(&tFile, true, TD_FTYPE_RSMA_QTASKINFO) < 0) {
smaError("vgId:%d, rsma persist, create %s failed since %s", vid, TD_TFILE_FULL_NAME(&tFile), terrstr());
goto _err;
}
smaDebug("vgId:%d, rsma, table %" PRIi64 " serialize qTaskInfo, file %s created", vid, pRSmaInfo->suid,
TD_TFILE_FULL_NAME(&tFile));
isFileCreated = true;
}
char tmpBuf[RSMA_QTASKINFO_HEAD_LEN] = {0};
void *pTmpBuf = &tmpBuf;
int32_t headLen = 0;
headLen += taosEncodeFixedI32(&pTmpBuf, len + RSMA_QTASKINFO_HEAD_LEN);
headLen += taosEncodeFixedI8(&pTmpBuf, type);
headLen += taosEncodeFixedI64(&pTmpBuf, pRSmaInfo->suid);
ASSERT(headLen <= RSMA_QTASKINFO_HEAD_LEN);
tdAppendTFile(&tFile, (void *)&tmpBuf, headLen, &toffset);
smaDebug("vgId:%d, rsma, table %" PRIi64 " level %d head part(len:%d) appended to offset:%" PRIi64, vid,
pRSmaInfo->suid, i + 1, headLen, toffset);
tdAppendTFile(&tFile, pOutput, len, &toffset);
smaDebug("vgId:%d, rsma, table %" PRIi64 " level %d body part len:%d appended to offset:%" PRIi64, vid,
pRSmaInfo->suid, i + 1, len, toffset);
taosMemoryFree(pOutput);
}
infoHash = taosHashIterate(pInfoHash, infoHash);
}
if (isFileCreated) {
if (tdUpdateTFileHeader(&tFile) < 0) {
smaError("vgId:%d, rsma, failed to update tfile %s header since %s", vid, TD_TFILE_FULL_NAME(&tFile),
tstrerror(terrno));
goto _err;
} else {
smaDebug("vgId:%d, rsma, succeed to update tfile %s header", vid, TD_TFILE_FULL_NAME(&tFile));
}
tdCloseTFile(&tFile);
tdDestroyTFile(&tFile);
}
return TSDB_CODE_SUCCESS;
_err:
smaError("vgId:%d, rsma persist failed since %s", vid, terrstr());
if (isFileCreated) {
tdRemoveTFile(&tFile);
tdDestroyTFile(&tFile);
}
return TSDB_CODE_FAILED;
}
#endif
/**
* @brief trigger to get rsma result in async mode
*

7
source/dnode/vnode/src/sma/smaUtil.c
View File

@ -16,7 +16,7 @@
#include "sma.h"
// smaFileUtil ================
#if 0
#define TD_FILE_STATE_OK 0
#define TD_FILE_STATE_BAD 1
@ -182,6 +182,8 @@ void tdCloseTFile(STFile *pTFile) {
void tdDestroyTFile(STFile *pTFile) { taosMemoryFreeClear(TD_TFILE_FULL_NAME(pTFile)); }
#endif
void tdGetVndFileName(int32_t vgId, const char *pdname, const char *dname, const char *fname, int64_t version,
char *outputName) {
if (version < 0) {
@ -221,6 +223,7 @@ void tdGetVndDirName(int32_t vgId, const char *pdname, const char *dname, bool e
}
}
#if 0
int32_t tdInitTFile(STFile *pTFile, const char *dname, const char *fname) {
TD_TFILE_SET_STATE(pTFile, TD_FILE_STATE_OK);
TD_TFILE_SET_CLOSED(pTFile);
@ -286,6 +289,8 @@ int32_t tdRemoveTFile(STFile *pTFile) {
return 0;
}
#endif
// smaXXXUtil ================
void *tdAcquireSmaRef(int32_t rsetId, int64_t refId) {
void *pResult = taosAcquireRef(rsetId, refId);

3
source/libs/executor/src/timewindowoperator.c
View File

@ -1725,6 +1725,9 @@ static void destroyStateWindowOperatorInfo(void* param) {
cleanupBasicInfo(&pInfo->binfo);
taosMemoryFreeClear(pInfo->stateKey.pData);
cleanupExprSupp(&pInfo->scalarSup);
colDataDestroy(&pInfo->twAggSup.timeWindowData);
cleanupAggSup(&pInfo->aggSup);
cleanupGroupResInfo(&pInfo->groupResInfo);
taosMemoryFreeClear(param);
}

2
source/libs/index/inc/indexInt.h
View File

@ -138,7 +138,7 @@ void idxReleaseRef(int64_t ref);
#define IDX_TYPE_ADD_EXTERN_TYPE(ty, exTy) \
do { \
uint8_t oldTy = ty; \
ty = (ty >> 4) | exTy; \
ty = ((ty >> 4) & 0xFF) | exTy; \
ty = (ty << 4) | oldTy; \
} while (0)

2
source/libs/index/src/index.c
View File

@ -139,7 +139,7 @@ int indexOpen(SIndexOpts* opts, const char* path, SIndex** index) {
END:
if (idx != NULL) {
indexClose(idx);
indexDestroy(idx);
}
*index = NULL;
return ret;

12
source/libs/index/src/indexCache.c
View File

@ -538,7 +538,7 @@ int idxCachePut(void* cache, SIndexTerm* term, uint64_t uid) {
idxCacheRef(pCache);
// encode data
CacheTerm* ct = taosMemoryCalloc(1, sizeof(CacheTerm));
if (cache == NULL) {
if (ct == NULL) {
return -1;
}
// set up key
@ -730,15 +730,17 @@ static int32_t idxCacheJsonTermCompare(const void* l, const void* r) {
return cmp;
}
static MemTable* idxInternalCacheCreate(int8_t type) {
int ttype = IDX_TYPE_CONTAIN_EXTERN_TYPE(type, TSDB_DATA_TYPE_JSON) ? TSDB_DATA_TYPE_BINARY : TSDB_DATA_TYPE_BINARY;
// int ttype = IDX_TYPE_CONTAIN_EXTERN_TYPE(type, TSDB_DATA_TYPE_JSON) ? TSDB_DATA_TYPE_BINARY :
// TSDB_DATA_TYPE_BINARY;
int ttype = TSDB_DATA_TYPE_BINARY;
int32_t (*cmpFn)(const void* l, const void* r) =
IDX_TYPE_CONTAIN_EXTERN_TYPE(type, TSDB_DATA_TYPE_JSON) ? idxCacheJsonTermCompare : idxCacheTermCompare;
MemTable* tbl = taosMemoryCalloc(1, sizeof(MemTable));
idxMemRef(tbl);
if (ttype == TSDB_DATA_TYPE_BINARY || ttype == TSDB_DATA_TYPE_NCHAR) {
tbl->mem = tSkipListCreate(MAX_SKIP_LIST_LEVEL, ttype, MAX_INDEX_KEY_LEN, cmpFn, SL_ALLOW_DUP_KEY, idxCacheTermGet);
}
// if (ttype == TSDB_DATA_TYPE_BINARY || ttype == TSDB_DATA_TYPE_NCHAR) {
tbl->mem = tSkipListCreate(MAX_SKIP_LIST_LEVEL, ttype, MAX_INDEX_KEY_LEN, cmpFn, SL_ALLOW_DUP_KEY, idxCacheTermGet);
//}
return tbl;
}

1
source/libs/index/src/indexComm.c
View File

@ -427,6 +427,7 @@ int32_t idxConvertDataToStr(void* src, int8_t type, void** dst) {
*dst = taosMemoryCalloc(1, bufSize + 1);
idxInt2str(*(uint64_t*)src, *dst, 1);
tlen = strlen(*dst);
break;
case TSDB_DATA_TYPE_FLOAT:
*dst = taosMemoryCalloc(1, bufSize + 1);
sprintf(*dst, "%.9lf", *(float*)src);

28
source/libs/index/src/indexFilter.c
View File

@ -231,7 +231,9 @@ static int32_t sifInitParam(SNode *node, SIFParam *param, SIFCtx *ctx) {
SIF_ERR_RET(sifGetValueFromNode(node, &param->condValue));
param->colId = -1;
param->colValType = (uint8_t)(vn->node.resType.type);
memcpy(param->colName, vn->literal, strlen(vn->literal));
if (strlen(vn->literal) <= sizeof(param->colName)) {
memcpy(param->colName, vn->literal, strlen(vn->literal));
}
break;
}
case QUERY_NODE_COLUMN: {
@ -400,54 +402,52 @@ static FORCE_INLINE FilterFunc sifGetFilterFunc(EIndexQueryType type, bool *reve
static void sifSetFltParam(SIFParam *left, SIFParam *right, SDataTypeBuf *typedata, SMetaFltParam *param) {
int8_t ltype = left->colValType, rtype = right->colValType;
if (ltype == TSDB_DATA_TYPE_FLOAT) {
float f;
float f = 0;
SIF_DATA_CONVERT(rtype, right->condValue, f);
typedata->f = f;
param->val = &typedata->f;
} else if (ltype == TSDB_DATA_TYPE_DOUBLE) {
double d;
double d = 0;
SIF_DATA_CONVERT(rtype, right->condValue, d);
typedata->d = d;
param->val = &typedata->d;
} else if (ltype == TSDB_DATA_TYPE_BIGINT) {
int64_t i64;
int64_t i64 = 0;
SIF_DATA_CONVERT(rtype, right->condValue, i64);
typedata->i64 = i64;
param->val = &typedata->i64;
} else if (ltype == TSDB_DATA_TYPE_INT) {
int32_t i32;
int32_t i32 = 0;
SIF_DATA_CONVERT(rtype, right->condValue, i32);
typedata->i32 = i32;
param->val = &typedata->i32;
} else if (ltype == TSDB_DATA_TYPE_SMALLINT) {
int16_t i16;
int16_t i16 = 0;
SIF_DATA_CONVERT(rtype, right->condValue, i16);
typedata->i16 = i16;
param->val = &typedata->i16;
} else if (ltype == TSDB_DATA_TYPE_TINYINT) {
int8_t i8;
int8_t i8 = 0;
SIF_DATA_CONVERT(rtype, right->condValue, i8)
typedata->i8 = i8;
param->val = &typedata->i8;
} else if (ltype == TSDB_DATA_TYPE_UBIGINT) {
uint64_t u64;
uint64_t u64 = 0;
SIF_DATA_CONVERT(rtype, right->condValue, u64);
typedata->u64 = u64;
param->val = &typedata->u64;
} else if (ltype == TSDB_DATA_TYPE_UINT) {
uint32_t u32;
uint32_t u32 = 0;
SIF_DATA_CONVERT(rtype, right->condValue, u32);
typedata->u32 = u32;
param->val = &typedata->u32;
} else if (ltype == TSDB_DATA_TYPE_USMALLINT) {
uint16_t u16;
uint16_t u16 = 0;
SIF_DATA_CONVERT(rtype, right->condValue, u16);
typedata->u16 = u16;
param->val = &typedata->u16;
} else if (ltype == TSDB_DATA_TYPE_UTINYINT) {
uint8_t u8;
uint8_t u8 = 0;
SIF_DATA_CONVERT(rtype, right->condValue, u8);
typedata->u8 = u8;
param->val = &typedata->u8;
@ -663,7 +663,7 @@ static int32_t sifExecOper(SOperatorNode *node, SIFCtx *ctx, SIFParam *output) {
// ugly code, refactor later
if (nParam > 1 && params[1].status == SFLT_NOT_INDEX) {
output->status = SFLT_NOT_INDEX;
return code;
goto _return;
}
SIF_ERR_JRET(sifGetOperFn(node->opType, &operFn, &output->status));
}

2
source/libs/index/src/indexFst.c
View File

@ -338,7 +338,7 @@ uint8_t fstStateCommInput(FstState* s, bool* null) {
return v;
}
// 0 indicate that common_input is None
return v == 0 ? 0 : COMMON_INPUT(v);
return COMMON_INPUT(v);
}
// input_len

6
source/libs/index/src/indexFstFile.c
View File

@ -72,7 +72,8 @@ static int idxFileCtxDoReadFrom(IFileCtx* ctx, uint8_t* buf, int len, int32_t of
if (offset >= ctx->file.size) return 0;
do {
char key[128] = {0};
char key[1024] = {0};
assert(strlen(ctx->file.buf) + 1 + 64 < sizeof(key));
idxGenLRUKey(key, ctx->file.buf, blkId);
LRUHandle* h = taosLRUCacheLookup(ctx->lru, key, strlen(key));
@ -99,6 +100,7 @@ static int idxFileCtxDoReadFrom(IFileCtx* ctx, uint8_t* buf, int len, int32_t of
assert(blk->nread <= kBlockSize);
if (blk->nread < kBlockSize && blk->nread < len) {
taosMemoryFree(blk);
break;
}
@ -150,7 +152,7 @@ IFileCtx* idxFileCtxCreate(WriterType type, const char* path, bool readOnly, int
if (ctx->type == TFILE) {
// ugly code, refactor later
ctx->file.readOnly = readOnly;
memcpy(ctx->file.buf, path, strlen(path));
memcpy(ctx->file.buf, path, sizeof(ctx->file.buf));
if (readOnly == false) {
ctx->file.pFile = taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_APPEND);
taosFtruncateFile(ctx->file.pFile, 0);

12
source/libs/index/src/indexTfile.c
View File

@ -506,7 +506,9 @@ TFileWriter* tfileWriterOpen(char* path, uint64_t suid, int64_t version, const c
tfh.suid = suid;
tfh.version = version;
tfh.colType = colType;
memcpy(tfh.colName, colName, strlen(colName));
if (strlen(colName) <= sizeof(tfh.colName)) {
memcpy(tfh.colName, colName, strlen(colName));
}
return tfileWriterCreate(wcx, &tfh);
}
@ -580,8 +582,14 @@ int tfileWriterPut(TFileWriter* tw, void* data, bool order) {
if (cap < ttsz) {
cap = ttsz;
buf = (char*)taosMemoryRealloc(buf, cap);
char* t = (char*)taosMemoryRealloc(buf, cap);
if (t == NULL) {
taosMemoryFree(buf);
return -1;
}
buf = t;
}
char* p = buf;
tfileSerialTableIdsToBuf(p, v->tableId);
tw->ctx->write(tw->ctx, buf, ttsz);

31
source/libs/planner/src/planOptimizer.c
View File

@ -124,8 +124,8 @@ static void optSetParentOrder(SLogicNode* pNode, EOrder order) {
EDealRes scanPathOptHaveNormalColImpl(SNode* pNode, void* pContext) {
if (QUERY_NODE_COLUMN == nodeType(pNode)) {
*((bool*)pContext) =
(COLUMN_TYPE_TAG != ((SColumnNode*)pNode)->colType && COLUMN_TYPE_TBNAME != ((SColumnNode*)pNode)->colType);
*((bool*)pContext) =
(COLUMN_TYPE_TAG != ((SColumnNode*)pNode)->colType && COLUMN_TYPE_TBNAME != ((SColumnNode*)pNode)->colType);
return *((bool*)pContext) ? DEAL_RES_END : DEAL_RES_IGNORE_CHILD;
}
return DEAL_RES_CONTINUE;
@ -1520,11 +1520,15 @@ static bool partTagsHasIndefRowsSelectFunc(SNodeList* pFuncs) {
return false;
}
static int32_t partTagsRewriteGroupTagsToFuncs(SNodeList* pGroupTags, SNodeList* pAggFuncs) {
static int32_t partTagsRewriteGroupTagsToFuncs(SNodeList* pGroupTags, int32_t start, SNodeList* pAggFuncs) {
bool hasIndefRowsSelectFunc = partTagsHasIndefRowsSelectFunc(pAggFuncs);
int32_t code = TSDB_CODE_SUCCESS;
int32_t index = 0;
SNode* pNode = NULL;
FOREACH(pNode, pGroupTags) {
if (index++ < start) {
continue;
}
if (hasIndefRowsSelectFunc) {
code = nodesListStrictAppend(pAggFuncs, partTagsCreateWrapperFunc("_select_value", pNode));
} else {
@ -1559,20 +1563,35 @@ static int32_t partTagsOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLogicSub
}
} else {
SAggLogicNode* pAgg = (SAggLogicNode*)pNode;
int32_t start = -1;
SNode* pGroupKey = NULL;
FOREACH(pGroupKey, pAgg->pGroupKeys) {
code = nodesListMakeStrictAppend(
&pScan->pGroupTags, nodesCloneNode(nodesListGetNode(((SGroupingSetNode*)pGroupKey)->pParameterList, 0)));
SNode* pGroupExpr = nodesListGetNode(((SGroupingSetNode*)pGroupKey)->pParameterList, 0);
if (NULL != pScan->pGroupTags) {
SNode* pGroupTag = NULL;
FOREACH(pGroupTag, pScan->pGroupTags) {
if (nodesEqualNode(pGroupTag, pGroupExpr)) {
continue;
}
}
}
if (start < 0) {
start = LIST_LENGTH(pScan->pGroupTags);
}
code = nodesListMakeStrictAppend(&pScan->pGroupTags, nodesCloneNode(pGroupExpr));
if (TSDB_CODE_SUCCESS != code) {
break;
}
}
NODES_DESTORY_LIST(pAgg->pGroupKeys);
code = partTagsRewriteGroupTagsToFuncs(pScan->pGroupTags, pAgg->pAggFuncs);
if (TSDB_CODE_SUCCESS == code && start >= 0) {
code = partTagsRewriteGroupTagsToFuncs(pScan->pGroupTags, start, pAgg->pAggFuncs);
}
}
if (TSDB_CODE_SUCCESS == code) {
code = partTagsOptRebuildTbanme(pScan->pGroupTags);
}
pCxt->optimized = true;
return code;
}

2
source/libs/planner/test/planPartByTest.cpp
View File

@ -61,6 +61,8 @@ TEST_F(PlanPartitionByTest, withGroupBy) {
run("SELECT COUNT(*) FROM t1 PARTITION BY c1 GROUP BY c2");
run("SELECT TBNAME, c1 FROM st1 PARTITION BY TBNAME GROUP BY c1");
run("SELECT COUNT(*) FROM t1 PARTITION BY TBNAME GROUP BY TBNAME");
}
TEST_F(PlanPartitionByTest, withTimeLineFunc) {

14
source/libs/scalar/src/scalar.c
View File

@ -629,6 +629,8 @@ _return:
sclFreeParam(pWhen);
sclFreeParam(pThen);
taosMemoryFree(pWhen);
taosMemoryFree(pThen);
SCL_RET(code);
}
@ -664,6 +666,8 @@ int32_t sclWalkWhenList(SScalarCtx *ctx, SNodeList* pList, struct SListCell* pCe
sclFreeParam(pWhen);
sclFreeParam(pThen);
taosMemoryFreeClear(pWhen);
taosMemoryFreeClear(pThen);
}
if (pElse) {
@ -688,6 +692,8 @@ _return:
sclFreeParam(pWhen);
sclFreeParam(pThen);
taosMemoryFree(pWhen);
taosMemoryFree(pThen);
SCL_RET(code);
}
@ -929,6 +935,10 @@ int32_t sclExecCaseWhen(SCaseWhenNode *node, SScalarCtx *ctx, SScalarParam *outp
sclFreeParam(&comp);
sclFreeParam(pWhen);
sclFreeParam(pThen);
taosMemoryFree(pCase);
taosMemoryFree(pElse);
taosMemoryFree(pWhen);
taosMemoryFree(pThen);
return TSDB_CODE_SUCCESS;
@ -940,6 +950,10 @@ _return:
sclFreeParam(pWhen);
sclFreeParam(pThen);
sclFreeParam(output);
taosMemoryFree(pCase);
taosMemoryFree(pElse);
taosMemoryFree(pWhen);
taosMemoryFree(pThen);
SCL_RET(code);
}

2
source/libs/scalar/src/sclvector.c
View File

@ -654,7 +654,7 @@ int32_t vectorConvertSingleColImpl(const SScalarParam* pIn, SScalarParam* pOut,
return TSDB_CODE_APP_ERROR;
}
int32_t rstart = startIndex >= 0 ? startIndex : 0;
int32_t rstart = (startIndex >= 0 && startIndex < pIn->numOfRows) ? startIndex : 0;
int32_t rend = numOfRows > 0 ? rstart + numOfRows - 1 : rstart + pIn->numOfRows - 1;
SSclVectorConvCtx cCtx = {pIn, pOut, rstart, rend, pInputCol->info.type, pOutputCol->info.type};

13
source/libs/sync/src/syncElection.c
View File

@ -96,6 +96,19 @@ int32_t syncNodeElect(SSyncNode* pSyncNode) {
syncNodeCandidate2Leader(pSyncNode);
pSyncNode->pVotesGranted->toLeader = true;
return ret;
}
if (pSyncNode->replicaNum == 1) {
// only myself, to leader
voteGrantedUpdate(pSyncNode->pVotesGranted, pSyncNode);
votesRespondUpdate(pSyncNode->pVotesRespond, pSyncNode);
pSyncNode->quorum = syncUtilQuorum(pSyncNode->pRaftCfg->cfg.replicaNum);
syncNodeCandidate2Leader(pSyncNode);
pSyncNode->pVotesGranted->toLeader = true;
return ret;
}
switch (pSyncNode->pRaftCfg->snapshotStrategy) {

20
source/libs/sync/src/syncMain.c
View File

@ -1891,8 +1891,26 @@ inline bool syncNodeInConfig(SSyncNode* pSyncNode, const SSyncCfg* config) {
return b1;
}
static bool syncIsConfigChanged(const SSyncCfg* pOldCfg, const SSyncCfg* pNewCfg) {
if (pOldCfg->replicaNum != pNewCfg->replicaNum) return true;
if (pOldCfg->myIndex != pNewCfg->myIndex) return true;
for (int32_t i = 0; i < pOldCfg->replicaNum; ++i) {
const SNodeInfo* pOldInfo = &pOldCfg->nodeInfo[i];
const SNodeInfo* pNewInfo = &pNewCfg->nodeInfo[i];
if (strcmp(pOldInfo->nodeFqdn, pNewInfo->nodeFqdn) != 0) return true;
if (pOldInfo->nodePort != pNewInfo->nodePort) return true;
}
return false;
}
void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncIndex lastConfigChangeIndex) {
SSyncCfg oldConfig = pSyncNode->pRaftCfg->cfg;
if (!syncIsConfigChanged(&oldConfig, pNewConfig)) {
sInfo("vgId:1, sync not reconfig since not changed");
return;
}
pSyncNode->pRaftCfg->cfg = *pNewConfig;
pSyncNode->pRaftCfg->lastConfigIndex = lastConfigChangeIndex;
@ -2264,7 +2282,7 @@ void syncNodeBecomeLeader(SSyncNode* pSyncNode, const char* debugStr) {
void syncNodeCandidate2Leader(SSyncNode* pSyncNode) {
ASSERT(pSyncNode->state == TAOS_SYNC_STATE_CANDIDATE);
ASSERT(voteGrantedMajority(pSyncNode->pVotesGranted));
//ASSERT(voteGrantedMajority(pSyncNode->pVotesGranted));
syncNodeBecomeLeader(pSyncNode, "candidate to leader");
syncNodeLog2("==state change syncNodeCandidate2Leader==", pSyncNode);

1
source/libs/tdb/src/db/tdbBtree.c
View File

@ -1086,6 +1086,7 @@ static int tdbBtreeEncodePayload(SPage *pPage, SCell *pCell, int nHeader, const
// fetch next ofp, a new ofp and make it dirty
ret = tdbFetchOvflPage(&pgno, &nextOfp, pTxn, pBt);
if (ret < 0) {
tdbFree(pBuf);
return -1;
}
}

6
source/libs/tdb/src/db/tdbDb.c
View File

@ -106,7 +106,7 @@ int32_t tdbBegin(TDB *pDb, TXN *pTxn) {
for (pPager = pDb->pgrList; pPager; pPager = pPager->pNext) {
ret = tdbPagerBegin(pPager, pTxn);
if (ret < 0) {
tdbError("failed to begin pager since %s. dbName:%s, txnId:%d", tstrerror(terrno), pDb->dbName, pTxn->txnId);
tdbError("failed to begin pager since %s. dbName:%s, txnId:%ld", tstrerror(terrno), pDb->dbName, pTxn->txnId);
return -1;
}
}
@ -121,7 +121,7 @@ int32_t tdbCommit(TDB *pDb, TXN *pTxn) {
for (pPager = pDb->pgrList; pPager; pPager = pPager->pNext) {
ret = tdbPagerCommit(pPager, pTxn);
if (ret < 0) {
tdbError("failed to commit pager since %s. dbName:%s, txnId:%d", tstrerror(terrno), pDb->dbName, pTxn->txnId);
tdbError("failed to commit pager since %s. dbName:%s, txnId:%ld", tstrerror(terrno), pDb->dbName, pTxn->txnId);
return -1;
}
}
@ -136,7 +136,7 @@ int32_t tdbAbort(TDB *pDb, TXN *pTxn) {
for (pPager = pDb->pgrList; pPager; pPager = pPager->pNext) {
ret = tdbPagerAbort(pPager, pTxn);
if (ret < 0) {
tdbError("failed to abort pager since %s. dbName:%s, txnId:%d", tstrerror(terrno), pDb->dbName, pTxn->txnId);
tdbError("failed to abort pager since %s. dbName:%s, txnId:%ld", tstrerror(terrno), pDb->dbName, pTxn->txnId);
return -1;
}
}

2
source/libs/tdb/src/db/tdbPCache.c
View File

@ -268,7 +268,7 @@ static SPage *tdbPCacheFetchImpl(SPCache *pCache, const SPgid *pPgid, TXN *pTxn)
// 4. Try a create new page
if (!pPage) {
ret = tdbPageCreate(pCache->szPage, &pPage, pTxn->xMalloc, pTxn->xArg);
if (ret < 0 && pPage != NULL) {
if (ret < 0 || pPage == NULL) {
// TODO
ASSERT(0);
return NULL;

4
source/libs/tdb/src/db/tdbPager.c
View File

@ -541,7 +541,7 @@ static int tdbPagerWritePageToJournal(SPager *pPager, SPage *pPage) {
ret = tdbOsWrite(pPager->jfd, pPage->pData, pPage->pageSize);
if (ret < 0) {
tdbError("failed to write page data due to %s. file:%s, pageSize:%ld", strerror(errno), pPager->jFileName,
tdbError("failed to write page data due to %s. file:%s, pageSize:%d", strerror(errno), pPager->jFileName,
pPage->pageSize);
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
@ -570,7 +570,7 @@ static int tdbPagerWritePageToDB(SPager *pPager, SPage *pPage) {
ret = tdbOsWrite(pPager->fd, pPage->pData, pPage->pageSize);
if (ret < 0) {
tdbError("failed to write page data due to %s. file:%s, pageSize:%ld", strerror(errno), pPager->dbFileName,
tdbError("failed to write page data due to %s. file:%s, pageSize:%d", strerror(errno), pPager->dbFileName,
pPage->pageSize);
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;

2
source/libs/transport/src/thttp.c
View File

@ -146,7 +146,7 @@ static FORCE_INLINE void clientRecvCb(uv_stream_t* handle, ssize_t nread, const
if (nread < 0) {
uError("http-report recv error:%s", uv_err_name(nread));
} else {
uTrace("http-report succ to recv %d bytes", nread);
uTrace("http-report succ to recv %d bytes", (int32_t)nread);
}
uv_close((uv_handle_t*)&cli->tcp, clientCloseCb);
}

8
source/libs/transport/src/trans.c
View File

@ -54,11 +54,7 @@ void* rpcOpen(const SRpcInit* pInit) {
pRpc->retry = pInit->rfp;
pRpc->startTimer = pInit->tfp;
if (pInit->connType == TAOS_CONN_SERVER) {
pRpc->numOfThreads = pInit->numOfThreads > TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS : pInit->numOfThreads;
} else {
pRpc->numOfThreads = pInit->numOfThreads > TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS : pInit->numOfThreads;
}
pRpc->numOfThreads = pInit->numOfThreads > TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS : pInit->numOfThreads;
uint32_t ip = 0;
if (pInit->connType == TAOS_CONN_SERVER) {
@ -79,7 +75,7 @@ void* rpcOpen(const SRpcInit* pInit) {
}
pRpc->parent = pInit->parent;
if (pInit->user) {
memcpy(pRpc->user, pInit->user, strlen(pInit->user));
memcpy(pRpc->user, pInit->user, TSDB_UNI_LEN);
}
int64_t refId = transAddExHandle(transGetInstMgt(), pRpc);

18
source/libs/transport/src/transCli.c
View File

@ -267,11 +267,12 @@ static void cliReleaseUnfinishedMsg(SCliConn* conn) {
#define EPSET_GET_SIZE(epSet) (epSet)->numOfEps
#define EPSET_GET_INUSE_IP(epSet) ((epSet)->eps[(epSet)->inUse].fqdn)
#define EPSET_GET_INUSE_PORT(epSet) ((epSet)->eps[(epSet)->inUse].port)
#define EPSET_FORWARD_INUSE(epSet) \
do { \
if ((epSet)->numOfEps != 0) { \
(epSet)->inUse = (++((epSet)->inUse)) % ((epSet)->numOfEps); \
} \
#define EPSET_FORWARD_INUSE(epSet) \
do { \
if ((epSet)->numOfEps != 0) { \
++((epSet)->inUse); \
(epSet)->inUse = ((epSet)->inUse) % ((epSet)->numOfEps); \
} \
} while (0)
#define EPSET_DEBUG_STR(epSet, tbuf) \
@ -503,6 +504,7 @@ static SCliConn* getConnFromPool(void* pool, char* ip, uint32_t port) {
SConnList list = {0};
taosHashPut((SHashObj*)pool, key, strlen(key), (void*)&list, sizeof(list));
plist = taosHashGet((SHashObj*)pool, key, strlen(key));
if (plist == NULL) return NULL;
QUEUE_INIT(&plist->conns);
}
@ -1157,7 +1159,7 @@ void* transInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads,
SCliObj* cli = taosMemoryCalloc(1, sizeof(SCliObj));
STrans* pTransInst = shandle;
memcpy(cli->label, label, strlen(label));
memcpy(cli->label, label, TSDB_LABEL_LEN);
cli->numOfThreads = numOfThreads;
cli->pThreadObj = (SCliThrd**)taosMemoryCalloc(cli->numOfThreads, sizeof(SCliThrd*));
@ -1611,8 +1613,8 @@ int transSetDefaultAddr(void* shandle, const char* ip, const char* fqdn) {
SCvtAddr cvtAddr = {0};
if (ip != NULL && fqdn != NULL) {
memcpy(cvtAddr.ip, ip, strlen(ip));
memcpy(cvtAddr.fqdn, fqdn, strlen(fqdn));
if (strlen(ip) <= sizeof(cvtAddr.ip)) memcpy(cvtAddr.ip, ip, strlen(ip));
if (strlen(fqdn) <= sizeof(cvtAddr.fqdn)) memcpy(cvtAddr.fqdn, fqdn, strlen(fqdn));
cvtAddr.cvt = true;
}
for (int i = 0; i < pTransInst->numOfThreads; i++) {

14
source/libs/wal/src/walMeta.c
View File

@ -123,8 +123,8 @@ static FORCE_INLINE int64_t walScanLogGetLastVer(SWal* pWal, int32_t fileIdx) {
}
SWalCkHead* logContent = (SWalCkHead*)candidate;
if (walValidHeadCksum(logContent) != 0) {
wError("vgId:%d, failed to validate checksum of wal entry header. offset:% %" PRId64 ", file:%s",
((char*)(logContent)-buf), fnameStr);
wWarn("vgId:%d, failed to validate checksum of wal entry header. offset:%" PRId64 ", file:%s", pWal->cfg.vgId,
offset + ((char*)(logContent)-buf), fnameStr);
haystack = candidate + 1;
if (firstTrial) {
break;
@ -162,8 +162,8 @@ static FORCE_INLINE int64_t walScanLogGetLastVer(SWal* pWal, int32_t fileIdx) {
}
if (walValidBodyCksum(logContent) != 0) {
terrno = TSDB_CODE_WAL_CHKSUM_MISMATCH;
wError("vgId:%d, failed to validate checksum of wal entry body. offset:% %" PRId64 ", file:%s",
((char*)(logContent)-buf), fnameStr);
wWarn("vgId:%d, failed to validate checksum of wal entry body. offset:%" PRId64 ", file:%s", pWal->cfg.vgId,
offset + ((char*)(logContent)-buf), fnameStr);
haystack = candidate + 1;
if (firstTrial) {
break;
@ -481,6 +481,10 @@ int walCheckAndRepairIdxFile(SWal* pWal, int32_t fileIdx) {
continue;
}
if (offset != (idxEntry.ver - pFileInfo->firstVer) * sizeof(SWalIdxEntry)) {
continue;
}
if (walReadLogHead(pLogFile, idxEntry.offset, &ckHead) < 0) {
wWarn("vgId:%d, failed to read log file since %s. file:%s, offset:%" PRId64 ", idx entry ver:%" PRId64 "",
pWal->cfg.vgId, terrstr(), fLogNameStr, idxEntry.offset, idxEntry.ver);
@ -493,6 +497,8 @@ int walCheckAndRepairIdxFile(SWal* pWal, int32_t fileIdx) {
}
offset += sizeof(SWalIdxEntry);
ASSERT(offset == (idxEntry.ver - pFileInfo->firstVer + 1) * sizeof(SWalIdxEntry));
// ftruncate idx file
if (offset < fileSize) {
if (taosFtruncateFile(pIdxFile, offset) < 0) {

53
source/libs/wal/src/walWrite.c
View File

@ -410,25 +410,35 @@ END:
static int32_t walWriteIndex(SWal *pWal, int64_t ver, int64_t offset) {
SWalIdxEntry entry = {.ver = ver, .offset = offset};
int64_t idxOffset = taosLSeekFile(pWal->pIdxFile, 0, SEEK_END);
SWalFileInfo *pFileInfo = walGetCurFileInfo(pWal);
ASSERT(pFileInfo != NULL);
ASSERT(pFileInfo->firstVer >= 0);
int64_t idxOffset = (entry.ver - pFileInfo->firstVer) * sizeof(SWalIdxEntry);
wDebug("vgId:%d, write index, index:%" PRId64 ", offset:%" PRId64 ", at %" PRId64, pWal->cfg.vgId, ver, offset,
idxOffset);
int64_t size = taosWriteFile(pWal->pIdxFile, &entry, sizeof(SWalIdxEntry));
if (size != sizeof(SWalIdxEntry)) {
wError("vgId:%d, failed to write idx entry due to %s. ver:%lld", pWal->cfg.vgId, strerror(errno), ver);
terrno = TAOS_SYSTEM_ERROR(errno);
// TODO truncate
return -1;
}
ASSERT(taosLSeekFile(pWal->pIdxFile, 0, SEEK_END) == idxOffset + sizeof(SWalIdxEntry) && "Offset of idx entries misaligned");
return 0;
}
// TODO gurantee atomicity by truncate failed writing
static FORCE_INLINE int32_t walWriteImpl(SWal *pWal, int64_t index, tmsg_t msgType, SWalSyncInfo syncMeta,
const void *body, int32_t bodyLen) {
int64_t code = 0;
int64_t offset = walGetCurFileOffset(pWal);
SWalFileInfo *pFileInfo = walGetCurFileInfo(pWal);
ASSERT(pFileInfo != NULL);
if (pFileInfo->firstVer == -1) {
pFileInfo->firstVer = index;
}
pWal->writeHead.head.version = index;
pWal->writeHead.head.bodyLen = bodyLen;
pWal->writeHead.head.msgType = msgType;
@ -439,11 +449,14 @@ static FORCE_INLINE int32_t walWriteImpl(SWal *pWal, int64_t index, tmsg_t msgTy
pWal->writeHead.cksumHead = walCalcHeadCksum(&pWal->writeHead);
pWal->writeHead.cksumBody = walCalcBodyCksum(body, bodyLen);
wDebug("vgId:%d, wal write log %ld, msgType: %s", pWal->cfg.vgId, index, TMSG_INFO(msgType));
code = walWriteIndex(pWal, index, offset);
if (code < 0) {
goto END;
}
if (taosWriteFile(pWal->pLogFile, &pWal->writeHead, sizeof(SWalCkHead)) != sizeof(SWalCkHead)) {
// TODO ftruncate
terrno = TAOS_SYSTEM_ERROR(errno);
wError("vgId:%d, file:%" PRId64 ".log, failed to write since %s", pWal->cfg.vgId, walGetLastFileFirstVer(pWal),
strerror(errno));
@ -452,7 +465,6 @@ static FORCE_INLINE int32_t walWriteImpl(SWal *pWal, int64_t index, tmsg_t msgTy
}
if (taosWriteFile(pWal->pLogFile, (char *)body, bodyLen) != bodyLen) {
// TODO ftruncate
terrno = TAOS_SYSTEM_ERROR(errno);
wError("vgId:%d, file:%" PRId64 ".log, failed to write since %s", pWal->cfg.vgId, walGetLastFileFirstVer(pWal),
strerror(errno));
@ -460,24 +472,31 @@ static FORCE_INLINE int32_t walWriteImpl(SWal *pWal, int64_t index, tmsg_t msgTy
goto END;
}
code = walWriteIndex(pWal, index, offset);
if (code < 0) {
// TODO ftruncate
goto END;
}
// set status
if (pWal->vers.firstVer == -1) pWal->vers.firstVer = index;
pWal->vers.lastVer = index;
pWal->totSize += sizeof(SWalCkHead) + bodyLen;
if (walGetCurFileInfo(pWal)->firstVer == -1) {
walGetCurFileInfo(pWal)->firstVer = index;
}
walGetCurFileInfo(pWal)->lastVer = index;
walGetCurFileInfo(pWal)->fileSize += sizeof(SWalCkHead) + bodyLen;
pFileInfo->lastVer = index;
pFileInfo->fileSize += sizeof(SWalCkHead) + bodyLen;
return 0;
END:
// recover in a reverse order
if (taosFtruncateFile(pWal->pLogFile, offset) < 0) {
wFatal("vgId:%d, failed to ftruncate logfile to offset:%lld during recovery due to %s", pWal->cfg.vgId, offset,
strerror(errno));
terrno = TAOS_SYSTEM_ERROR(errno);
ASSERT(0 && "failed to recover from error");
}
int64_t idxOffset = (index - pFileInfo->firstVer) * sizeof(SWalIdxEntry);
if (taosFtruncateFile(pWal->pIdxFile, idxOffset) < 0) {
wFatal("vgId:%d, failed to ftruncate idxfile to offset:%lld during recovery due to %s", pWal->cfg.vgId, idxOffset,
strerror(errno));
terrno = TAOS_SYSTEM_ERROR(errno);
ASSERT(0 && "failed to recover from error");
}
return -1;
}

8
source/os/src/osSocket.c
View File

@ -590,7 +590,9 @@ TdSocketPtr taosOpenTcpClientSocket(uint32_t destIp, uint16_t destPort, uint32_t
taosCloseSocket(&pSocket);
return NULL;
} else {
taosKeepTcpAlive(pSocket);
if (taosKeepTcpAlive(pSocket) == -1) {
return NULL;
}
}
return pSocket;
@ -1059,18 +1061,22 @@ int32_t taosCreateSocketWithTimeout(uint32_t timeout) {
}
#if defined(WINDOWS)
if (0 != setsockopt(fd, IPPROTO_TCP, TCP_MAXRT, (char *)&timeout, sizeof(timeout))) {
taosCloseSocketNoCheck1(fd);
return -1;
}
#elif defined(_TD_DARWIN_64)
uint32_t conn_timeout_ms = timeout * 1000;
if (0 != setsockopt(fd, IPPROTO_TCP, TCP_CONNECTIONTIMEOUT, (char *)&conn_timeout_ms, sizeof(conn_timeout_ms))) {
taosCloseSocketNoCheck1(fd);
return -1;
}
#else // Linux like systems
uint32_t conn_timeout_ms = timeout * 1000;
if (0 != setsockopt(fd, IPPROTO_TCP, TCP_USER_TIMEOUT, (char *)&conn_timeout_ms, sizeof(conn_timeout_ms))) {
taosCloseSocketNoCheck1(fd);
return -1;
}
#endif
return (int)fd;
}

5
source/util/src/tlog.c
View File

@ -443,10 +443,13 @@ static inline int32_t taosBuildLogHead(char *buffer, const char *flags) {
static inline void taosPrintLogImp(ELogLevel level, int32_t dflag, const char *buffer, int32_t len) {
if ((dflag & DEBUG_FILE) && tsLogObj.logHandle && tsLogObj.logHandle->pFile != NULL && osLogSpaceAvailable()) {
taosUpdateLogNums(level);
if (tsAsyncLog) {
if (tsAsyncLog && level != DEBUG_FATAL) {
taosPushLogBuffer(tsLogObj.logHandle, buffer, len);
} else {
taosWriteFile(tsLogObj.logHandle->pFile, buffer, len);
if (level == DEBUG_FATAL) {
taosFsyncFile(tsLogObj.logHandle->pFile);
}
}
if (tsLogObj.maxLines > 0) {

1
source/util/src/tsched.c
View File

@ -38,6 +38,7 @@ void *taosInitScheduler(int32_t queueSize, int32_t numOfThreads, const char *lab
if (pSched->queue == NULL) {
uError("%s: no enough memory for queue", label);
taosCleanUpScheduler(pSched);
taosMemoryFree(pSched);
return NULL;
}

7
tools/shell/src/shellCommand.c
View File

@ -101,11 +101,8 @@ void shellInsertChar(SShellCmd *cmd, char *c, int32_t size) {
/* update the values */
cmd->commandSize += size;
cmd->cursorOffset += size;
for (int i = 0; i < size; i++) {
taosMbToWchar(&wc, c + i, size);
cmd->screenOffset += taosWcharWidth(wc);
cmd->endOffset += taosWcharWidth(wc);
}
cmd->screenOffset += taosWcharWidth(wc);
cmd->endOffset += taosWcharWidth(wc);
#ifdef WINDOWS
#else
shellShowOnScreen(cmd);

58
utils/test/c/sml_test.c
View File

@ -63,6 +63,7 @@ int smlProcess_influx_Test() {
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
int code = taos_errno(pRes);
taos_free_result(pRes);
taos_close(taos);
return code;
}
@ -86,6 +87,8 @@ int smlProcess_telnet_Test() {
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
int code = taos_errno(pRes);
taos_free_result(pRes);
taos_close(taos);
return code;
}
@ -125,6 +128,8 @@ int smlProcess_json1_Test() {
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
int code = taos_errno(pRes);
taos_free_result(pRes);
taos_close(taos);
return code;
}
@ -165,6 +170,8 @@ int smlProcess_json2_Test() {
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
int code = taos_errno(pRes);
taos_free_result(pRes);
taos_close(taos);
return code;
}
@ -233,6 +240,8 @@ int smlProcess_json3_Test() {
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
int code = taos_errno(pRes);
taos_free_result(pRes);
taos_close(taos);
return code;
}
@ -292,6 +301,8 @@ int smlProcess_json4_Test() {
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
int code = taos_errno(pRes);
taos_free_result(pRes);
taos_close(taos);
return code;
}
@ -313,6 +324,8 @@ int sml_TD15662_Test() {
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
int code = taos_errno(pRes);
taos_free_result(pRes);
taos_close(taos);
return code;
}
@ -333,6 +346,8 @@ int sml_TD15742_Test() {
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
int code = taos_errno(pRes);
taos_free_result(pRes);
taos_close(taos);
return code;
}
@ -362,6 +377,8 @@ int sml_16384_Test() {
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
code = taos_errno(pRes);
taos_free_result(pRes);
taos_close(taos);
return code;
}
@ -781,6 +798,8 @@ int sml_oom_Test() {
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
int code = taos_errno(pRes);
taos_free_result(pRes);
taos_close(taos);
return code;
}
@ -825,6 +844,8 @@ int sml_16368_Test() {
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
int code = taos_errno(pRes);
taos_free_result(pRes);
taos_close(taos);
return code;
}
@ -862,6 +883,8 @@ int sml_dup_time_Test() {
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
int code = taos_errno(pRes);
taos_free_result(pRes);
taos_close(taos);
return code;
}
@ -1068,6 +1091,8 @@ int sml_16960_Test() {
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
int code = taos_errno(pRes);
taos_free_result(pRes);
taos_close(taos);
return code;
}
@ -1097,6 +1122,7 @@ int sml_add_tag_col_Test() {
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
code = taos_errno(pRes);
taos_free_result(pRes);
taos_close(taos);
return code;
}
@ -1151,6 +1177,36 @@ int smlProcess_18784_Test() {
rowIndex++;
}
taos_free_result(pRes);
taos_close(taos);
return code;
}
int sml_19221_Test() {
TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);
TAOS_RES *pRes = taos_query(taos, "create database if not exists sml_db schemaless 1");
taos_free_result(pRes);
const char *sql[] = {
"qelhxo,id=pnnqhsa,t0=t,t1=127i8 c11=L\"ncharColValue\",c0=t,c1=127i8 1626006833639000000\nqelhxo,id=pnnhsa,t0=t,t1=127i8 c11=L\"ncharColValue\",c0=t,c1=127i8 1626006833639000000\n#comment\nqelhxo,id=pnqhsa,t0=t,t1=127i8 c11=L\"ncharColValue\",c0=t,c1=127i8 1626006833639000000",
};
pRes = taos_query(taos, "use sml_db");
taos_free_result(pRes);
char* tmp = (char*)taosMemoryCalloc(1024, 1);
memcpy(tmp, sql[0], strlen(sql[0]));
*(char*)(tmp+44) = 0;
int32_t totalRows = 0;
pRes = taos_schemaless_insert_raw(taos, tmp, strlen(sql[0]), &totalRows, TSDB_SML_LINE_PROTOCOL, TSDB_SML_TIMESTAMP_NANO_SECONDS);
ASSERT(totalRows == 3);
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
int code = taos_errno(pRes);
taos_free_result(pRes);
taos_close(taos);
taosMemoryFree(tmp);
return code;
}
@ -1187,5 +1243,7 @@ int main(int argc, char *argv[]) {
ASSERT(!ret);
ret = smlProcess_18784_Test();
ASSERT(!ret);
ret = sml_19221_Test();
ASSERT(!ret);
return ret;
}