Skip to content

T
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
提交 39243cbb 编写于 作者: C Cary Xu
浏览文件
操作

Merge branch '3.0' into fix/TD-19254-D

上级 a6e3da80 510a7458
展开全部 隐藏空白更改
内联 并排
Showing with 1888 addition and 1536 deletion
cmake/taostools_CMakeLists.txt.in
浏览文件 @ 39243cbb
......@@ -2,7 +2,7 @@
# taos-tools
ExternalProject_Add(taos-tools
GIT_REPOSITORY https://github.com/taosdata/taos-tools.git
GIT_TAG c64858f
GIT_TAG 9284147
SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools"
BINARY_DIR ""
#BUILD_IN_SOURCE TRUE
......
docs/en/12-taos-sql/10-function.md
浏览文件 @ 39243cbb
......@@ -945,7 +945,7 @@ MIN(expr)
MODE(expr)
```
**Description**:The value which has the highest frequency of occurrence. NULL is returned if there are multiple values which have highest frequency of occurrence.
**Description**:The value which has the highest frequency of occurrence. One random value is returned if there are multiple values which have highest frequency of occurrence.
**Return value type**: Same as the input data
......
docs/en/14-reference/03-connector/05-go.mdx
浏览文件 @ 39243cbb
......@@ -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)
docs/en/14-reference/03-connector/_macos_install.mdx
浏览文件 @ 39243cbb
......@@ -2,7 +2,7 @@ import PkgListV3 from "/components/PkgListV3";
1. Download the client installation package
<PkgListV3 type={8} sys="MacOS" />
<PkgListV3 type={8} sys="macOS" />
[All Downloads](../../releases/tdengine)
......
docs/zh/08-connector/20-go.mdx
浏览文件 @ 39243cbb
......@@ -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)
docs/zh/08-connector/_macos_install.mdx
浏览文件 @ 39243cbb
......@@ -2,7 +2,7 @@ import PkgListV3 from "/components/PkgListV3";
1. 下载客户端安装包
<PkgListV3 type={8} sys="MacOS" />
<PkgListV3 type={8} sys="macOS" />
[所有下载](../../releases/tdengine)
......
docs/zh/12-taos-sql/10-function.md
浏览文件 @ 39243cbb
......@@ -946,7 +946,7 @@ MIN(expr)
MODE(expr)
```
**功能说明**：返回出现频率最高的值，若存在多个频率相同的最高值，输出NULL。
**功能说明**：返回出现频率最高的值，若存在多个频率相同的最高值，则随机输出其中某个值。
**返回数据类型**：与输入数据类型一致。
......
include/client/taos.h
浏览文件 @ 39243cbb
......@@ -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------------------------------- */
......
include/common/tcommon.h
浏览文件 @ 39243cbb
......@@ -44,12 +44,17 @@ enum {
)
// clang-format on
typedef struct {
typedef struct SWinKey {
uint64_t groupId;
TSKEY ts;
} SWinKey;
static inline int sWinKeyCmprImpl(const void* pKey1, const void* pKey2) {
typedef struct SSessionKey {
STimeWindow win;
uint64_t groupId;
} SSessionKey;
static inline int winKeyCmprImpl(const void* pKey1, const void* pKey2) {
SWinKey* pWin1 = (SWinKey*)pKey1;
SWinKey* pWin2 = (SWinKey*)pKey2;
......@@ -69,7 +74,7 @@ static inline int sWinKeyCmprImpl(const void* pKey1, const void* pKey2) {
}
static inline int winKeyCmpr(const void* pKey1, int kLen1, const void* pKey2, int kLen2) {
return sWinKeyCmprImpl(pKey1, pKey2);
return winKeyCmprImpl(pKey1, pKey2);
}
typedef struct {
......
include/libs/stream/streamState.h
浏览文件 @ 39243cbb
......@@ -25,6 +25,8 @@ extern "C" {
typedef struct SStreamTask SStreamTask;
typedef bool (*state_key_cmpr_fn)(void* pKey1, void* pKey2);
// incremental state storage
typedef struct {
SStreamTask* pOwner;
......@@ -32,6 +34,7 @@ typedef struct {
TTB* pStateDb;
TTB* pFuncStateDb;
TTB* pFillStateDb; // todo refactor
TTB* pSessionStateDb;
TXN txn;
int32_t number;
} SStreamState;
......@@ -57,6 +60,19 @@ int32_t streamStateDel(SStreamState* pState, const SWinKey* key);
int32_t streamStateClear(SStreamState* pState);
void streamStateSetNumber(SStreamState* pState, int32_t number);
int32_t streamStateSessionAddIfNotExist(SStreamState* pState, SSessionKey* key, void** pVal, int32_t* pVLen);
int32_t streamStateSessionPut(SStreamState* pState, const SSessionKey* key, const void* value, int32_t vLen);
int32_t streamStateSessionGet(SStreamState* pState, SSessionKey* key, void** pVal, int32_t* pVLen);
int32_t streamStateSessionDel(SStreamState* pState, const SSessionKey* key);
int32_t streamStateSessionClear(SStreamState* pState);
int32_t streamStateSessionGetKVByCur(SStreamStateCur* pCur, SSessionKey* pKey, const void** pVal, int32_t* pVLen);
int32_t streamStateStateAddIfNotExist(SStreamState* pState, SSessionKey* key, char* pKeyData, int32_t keyDataLen,
state_key_cmpr_fn fn, void** pVal, int32_t* pVLen);
SStreamStateCur* streamStateSessionSeekKeyNext(SStreamState* pState, const SSessionKey* key);
SStreamStateCur* streamStateSessionSeekKeyPrev(SStreamState* pState, const SSessionKey* key);
SStreamStateCur* streamStateSessionGetCur(SStreamState* pState, const SSessionKey* key);
int32_t streamStateFillPut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen);
int32_t streamStateFillGet(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen);
int32_t streamStateFillDel(SStreamState* pState, const SWinKey* key);
......
include/util/tdef.h
浏览文件 @ 39243cbb
......@@ -498,6 +498,7 @@ enum {
#define MAX_NUM_STR_SIZE 40
#define MAX_META_MSG_IN_BATCH 1048576
#define MAX_META_BATCH_RSP_SIZE (1 * 1048576 * 1024)
#ifdef __cplusplus
}
......
source/client/src/clientImpl.c
浏览文件 @ 39243cbb
......@@ -199,7 +199,7 @@ int32_t buildRequest(uint64_t connId, const char* sql, int sqlLen, void* param,
if (tsQueryUseNodeAllocator && !qIsInsertValuesSql((*pRequest)->sqlstr, (*pRequest)->sqlLen)) {
if (TSDB_CODE_SUCCESS !=
nodesCreateAllocator((*pRequest)->requestId, tsQueryNodeChunkSize, &((*pRequest)->allocatorRefId))) {
tscError("%d failed to create node allocator, reqId:0x%" PRIx64 ", conn:%d, %s", (*pRequest)->self,
tscError("%d failed to create node allocator, reqId:0x%" PRIx64 ", conn:%" PRId64 ", %s", (*pRequest)->self,
(*pRequest)->requestId, pTscObj->id, sql);
destroyRequest(*pRequest);
......@@ -955,7 +955,12 @@ SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, bool keepQue
switch (pQuery->execMode) {
case QUERY_EXEC_MODE_LOCAL:
if (!pRequest->validateOnly) {
code = execLocalCmd(pRequest, pQuery);
if (NULL == pQuery->pRoot) {
terrno = TSDB_CODE_INVALID_PARA;
code = terrno;
} else {
code = execLocalCmd(pRequest, pQuery);
}
}
break;
case QUERY_EXEC_MODE_RPC:
......@@ -997,7 +1002,7 @@ SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, bool keepQue
handleQueryExecRsp(pRequest);
if (NULL != pRequest && TSDB_CODE_SUCCESS != code) {
if (TSDB_CODE_SUCCESS != code) {
pRequest->code = terrno;
}
......@@ -2254,7 +2259,10 @@ void syncQueryFn(void* param, void* res, int32_t code) {
void taosAsyncQueryImpl(uint64_t connId, const char* sql, __taos_async_fn_t fp, void* param, bool validateOnly) {
if (sql == NULL || NULL == fp) {
terrno = TSDB_CODE_INVALID_PARA;
fp(param, NULL, terrno);
if (fp) {
fp(param, NULL, terrno);
}
return;
}
......
source/client/src/clientMain.c
浏览文件 @ 39243cbb
......@@ -944,7 +944,6 @@ void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {
if (pResultInfo->completed) {
// it is a local executed query, no need to do async fetch
if (QUERY_EXEC_MODE_LOCAL == pRequest->body.execMode) {
ASSERT(pResultInfo->numOfRows >= 0);
if (pResultInfo->localResultFetched) {
pResultInfo->numOfRows = 0;
pResultInfo->current = 0;
......
source/client/src/clientMsgHandler.c
浏览文件 @ 39243cbb
......@@ -292,8 +292,10 @@ int32_t processDropDbRsp(void* param, SDataBuf* pMsg, int32_t code) {
tDeserializeSDropDbRsp(pMsg->pData, pMsg->len, &dropdbRsp);
struct SCatalog* pCatalog = NULL;
catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog);
catalogRemoveDB(pCatalog, dropdbRsp.db, dropdbRsp.uid);
int32_t code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog);
if (TSDB_CODE_SUCCESS == code) {
catalogRemoveDB(pCatalog, dropdbRsp.db, dropdbRsp.uid);
}
}
taosMemoryFree(pMsg->pData);
......@@ -397,6 +399,7 @@ static int32_t buildShowVariablesRsp(SArray* pVars, SRetrieveTableRsp** pRsp) {
size_t rspSize = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pBlock);
*pRsp = taosMemoryCalloc(1, rspSize);
if (NULL == *pRsp) {
blockDataDestroy(pBlock);
return TSDB_CODE_OUT_OF_MEMORY;
}
......
source/client/src/clientSml.c
浏览文件 @ 39243cbb
......@@ -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;
......@@ -1371,8 +1372,14 @@ static int32_t smlKvTimeArrayCompare(const void *key1, const void *key2) {
static int32_t smlKvTimeHashCompare(const void *key1, const void *key2) {
SHashObj *s1 = *(SHashObj **)key1;
SHashObj *s2 = *(SHashObj **)key2;
SSmlKv *kv1 = *(SSmlKv **)taosHashGet(s1, TS, TS_LEN);
SSmlKv *kv2 = *(SSmlKv **)taosHashGet(s2, TS, TS_LEN);
SSmlKv **kv1pp = (SSmlKv **)taosHashGet(s1, TS, TS_LEN);
SSmlKv **kv2pp = (SSmlKv **)taosHashGet(s2, TS, TS_LEN);
if(!kv1pp || !kv2pp){
uError("smlKvTimeHashCompare kv is null");
return -1;
}
SSmlKv *kv1 = *kv1pp;
SSmlKv *kv2 = *kv2pp;
if(!kv1 || kv1->type != TSDB_DATA_TYPE_TIMESTAMP){
uError("smlKvTimeHashCompare kv1");
return -1;
......@@ -2073,11 +2080,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 +2191,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 +2205,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 +2296,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 +2385,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 +2401,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 +2533,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 +2558,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 +2608,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 +2637,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);
}
source/client/src/clientStmt.c
浏览文件 @ 39243cbb
......@@ -152,7 +152,7 @@ int32_t stmtUpdateBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags,
pStmt->bInfo.tbType = pTableMeta->tableType;
pStmt->bInfo.boundTags = tags;
pStmt->bInfo.tagsCached = false;
strcpy(pStmt->bInfo.stbFName, sTableName);
tstrncpy(pStmt->bInfo.stbFName, sTableName, sizeof(pStmt->bInfo.stbFName));
return TSDB_CODE_SUCCESS;
}
......
source/client/src/clientTmq.c
浏览文件 @ 39243cbb
......@@ -210,6 +210,8 @@ typedef struct {
typedef struct {
SMqCommitCbParamSet* params;
STqOffset* pOffset;
/*char topicName[TSDB_TOPIC_FNAME_LEN];*/
/*int32_t vgId;*/
} SMqCommitCbParam;
tmq_conf_t* tmq_conf_new() {
......@@ -407,6 +409,14 @@ int32_t tmqCommitDone(SMqCommitCbParamSet* pParamSet) {
return 0;
}
static void tmqCommitRspCountDown(SMqCommitCbParamSet* pParamSet) {
int32_t waitingRspNum = atomic_sub_fetch_32(&pParamSet->waitingRspNum, 1);
ASSERT(waitingRspNum >= 0);
if (waitingRspNum == 0) {
tmqCommitDone(pParamSet);
}
}
int32_t tmqCommitCb(void* param, SDataBuf* pBuf, int32_t code) {
SMqCommitCbParam* pParam = (SMqCommitCbParam*)param;
SMqCommitCbParamSet* pParamSet = (SMqCommitCbParamSet*)pParam->params;
......@@ -420,18 +430,13 @@ int32_t tmqCommitCb(void* param, SDataBuf* pBuf, int32_t code) {
#endif
taosMemoryFree(pParam->pOffset);
if (pBuf->pData) taosMemoryFree(pBuf->pData);
taosMemoryFree(pBuf->pData);
/*tscDebug("receive offset commit cb of %s on vgId:%d, offset is %" PRId64, pParam->pOffset->subKey, pParam->->vgId,
* pOffset->version);*/
// count down waiting rsp
int32_t waitingRspNum = atomic_sub_fetch_32(&pParamSet->waitingRspNum, 1);
ASSERT(waitingRspNum >= 0);
tmqCommitRspCountDown(pParamSet);
if (waitingRspNum == 0) {
tmqCommitDone(pParamSet);
}
return 0;
}
......@@ -591,14 +596,10 @@ FAIL:
return 0;
}
int32_t tmqCommitInner(tmq_t* tmq, const TAOS_RES* msg, int8_t automatic, int8_t async, tmq_commit_cb* userCb,
void* userParam) {
static int32_t tmqCommitConsumerImpl(tmq_t* tmq, int8_t automatic, int8_t async, tmq_commit_cb* userCb,
void* userParam) {
int32_t code = -1;
if (msg != NULL) {
return tmqCommitMsgImpl(tmq, msg, async, userCb, userParam);
}
SMqCommitCbParamSet* pParamSet = taosMemoryCalloc(1, sizeof(SMqCommitCbParamSet));
if (pParamSet == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
......@@ -646,33 +647,37 @@ int32_t tmqCommitInner(tmq_t* tmq, const TAOS_RES* msg, int8_t automatic, int8_t
}
}
// no request is sent
if (pParamSet->totalRspNum == 0) {
tsem_destroy(&pParamSet->rspSem);
taosMemoryFree(pParamSet);
return 0;
}
int32_t waitingRspNum = atomic_sub_fetch_32(&pParamSet->waitingRspNum, 1);
ASSERT(waitingRspNum >= 0);
if (waitingRspNum == 0) {
tmqCommitDone(pParamSet);
}
// count down since waiting rsp num init as 1
tmqCommitRspCountDown(pParamSet);
if (!async) {
tsem_wait(&pParamSet->rspSem);
code = pParamSet->rspErr;
tsem_destroy(&pParamSet->rspSem);
taosMemoryFree(pParamSet);
}
#if 0
if (!async) {
taosArrayDestroyP(pParamSet->successfulOffsets, taosMemoryFree);
taosArrayDestroyP(pParamSet->failedOffsets, taosMemoryFree);
}
#endif
}
return 0;
return code;
}
int32_t tmqCommitInner(tmq_t* tmq, const TAOS_RES* msg, int8_t automatic, int8_t async, tmq_commit_cb* userCb,
void* userParam) {
if (msg) {
return tmqCommitMsgImpl(tmq, msg, async, userCb, userParam);
} else {
return tmqCommitConsumerImpl(tmq, automatic, async, userCb, userParam);
}
}
void tmqAssignAskEpTask(void* param, void* tmrId) {
......
source/client/test/smlTest.cpp
浏览文件 @ 39243cbb
......@@ -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);
......
source/common/src/tdatablock.c
浏览文件 @ 39243cbb
......@@ -1892,12 +1892,13 @@ char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf)
for (int32_t k = 0; k < colNum; k++) {
SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, k);
void* var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);
if (colDataIsNull(pColInfoData, rows, j, NULL) || !pColInfoData->pData) {
len += snprintf(dumpBuf + len, size - len, " %15s |", "NULL");
if (len >= size - 1) return dumpBuf;
continue;
}
void* var = colDataGetData(pColInfoData, j);
switch (pColInfoData->info.type) {
case TSDB_DATA_TYPE_TIMESTAMP:
memset(pBuf, 0, sizeof(pBuf));
......@@ -1926,8 +1927,8 @@ char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf)
if (len >= size - 1) return dumpBuf;
break;
case TSDB_DATA_TYPE_DOUBLE:
len += snprintf(dumpBuf + len, size - len, " %15lf |", *(double*)var);
if (len >= size - 1) return dumpBuf;
// len += snprintf(dumpBuf + len, size - len, " %15lf |", *(double*)var);
// if (len >= size - 1) return dumpBuf;
break;
case TSDB_DATA_TYPE_BOOL:
len += snprintf(dumpBuf + len, size - len, " %15d |", *(bool*)var);
......
source/dnode/mgmt/node_mgmt/src/dmEnv.c
浏览文件 @ 39243cbb
......@@ -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() {
......
source/dnode/mgmt/test/sut/src/sut.cpp
浏览文件 @ 39243cbb
......@@ -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);
......
source/dnode/mnode/impl/src/mndMain.c
浏览文件 @ 39243cbb
......@@ -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) {
......
source/dnode/mnode/impl/src/mndMnode.c
浏览文件 @ 39243cbb
......@@ -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,25 @@ static void mndReloadSyncConfig(SMnode *pMnode) {
sdbReleaseLock(pSdb, pObj, false);
}
if (cfg.myIndex == -1) {
mInfo("vgId:1, mnode not reload since selfIndex is -1");
if (readyMnodes <= 0 || updatingMnodes <= 0) {
mInfo("vgId:1, mnode sync not reconfig since readyMnodes:%d updatingMnodes:%d", readyMnodes, updatingMnodes);
return;
}
// ASSERT(0);
if (!mndGetRestored(pMnode)) {
mInfo("vgId:1, mnode not reload since restore not finished");
if (cfg.myIndex == -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 (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);
......
source/dnode/mnode/impl/src/mndQuery.c
浏览文件 @ 39243cbb
......@@ -90,14 +90,39 @@ int32_t mndProcessBatchMetaMsg(SRpcMsg *pMsg) {
}
for (int32_t i = 0; i < msgNum; ++i) {
if (offset >= pMsg->contLen) {
mError("offset %d is bigger than contLen %d", offset, pMsg->contLen);
terrno = TSDB_CODE_MSG_NOT_PROCESSED;
taosArrayDestroy(batchRsp);
return -1;
}
req.msgIdx = ntohl(*(int32_t *)((char *)pMsg->pCont + offset));
offset += sizeof(req.msgIdx);
if (offset >= pMsg->contLen) {
mError("offset %d is bigger than contLen %d", offset, pMsg->contLen);
terrno = TSDB_CODE_MSG_NOT_PROCESSED;
taosArrayDestroy(batchRsp);
return -1;
}
req.msgType = ntohl(*(int32_t *)((char *)pMsg->pCont + offset));
offset += sizeof(req.msgType);
if (offset >= pMsg->contLen) {
mError("offset %d is bigger than contLen %d", offset, pMsg->contLen);
terrno = TSDB_CODE_MSG_NOT_PROCESSED;
taosArrayDestroy(batchRsp);
return -1;
}
req.msgLen = ntohl(*(int32_t *)((char *)pMsg->pCont + offset));
offset += sizeof(req.msgLen);
if (offset >= pMsg->contLen) {
mError("offset %d is bigger than contLen %d", offset, pMsg->contLen);
terrno = TSDB_CODE_MSG_NOT_PROCESSED;
taosArrayDestroy(batchRsp);
return -1;
}
req.msg = (char *)pMsg->pCont + offset;
offset += req.msgLen;
......
source/dnode/mnode/impl/src/mndStb.c
浏览文件 @ 39243cbb
......@@ -2553,12 +2553,17 @@ static int32_t mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBloc
char rollup[160 + VARSTR_HEADER_SIZE] = {0};
int32_t rollupNum = (int32_t)taosArrayGetSize(pStb->pFuncs);
char *sep = ", ";
int32_t sepLen = strlen(sep);
int32_t rollupLen = sizeof(rollup) - 2;
for (int32_t i = 0; i < rollupNum; ++i) {
char *funcName = taosArrayGet(pStb->pFuncs, i);
if (i) {
strcat(varDataVal(rollup), ", ");
strncat(varDataVal(rollup), sep, rollupLen);
rollupLen -= sepLen;
}
strcat(varDataVal(rollup), funcName);
strncat(varDataVal(rollup), funcName, rollupLen);
rollupLen -= strlen(funcName);
}
varDataSetLen(rollup, strlen(varDataVal(rollup)));
......
source/dnode/mnode/impl/src/mndSync.c
浏览文件 @ 39243cbb
......@@ -293,6 +293,14 @@ int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw, int32_t transId) {
if (code == 0) {
tsem_wait(&pMgmt->syncSem);
} else if (code > 0) {
mInfo("trans:%d, confirm at once since replica is 1, continue execute", transId);
taosWLockLatch(&pMgmt->lock);
pMgmt->transId = 0;
taosWUnLockLatch(&pMgmt->lock);
sdbWriteWithoutFree(pMnode->pSdb, pRaw);
sdbSetApplyInfo(pMnode->pSdb, req.info.conn.applyIndex, req.info.conn.applyTerm, SYNC_INDEX_INVALID);
code = 0;
} else if (code == -1 && terrno == TSDB_CODE_SYN_NOT_LEADER) {
terrno = TSDB_CODE_APP_NOT_READY;
} else if (code == -1 && terrno == TSDB_CODE_SYN_INTERNAL_ERROR) {
......
source/dnode/mnode/sdb/inc/sdb.h
浏览文件 @ 39243cbb
......@@ -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
......
source/dnode/mnode/sdb/src/sdb.c
浏览文件 @ 39243cbb
......@@ -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);
}
source/dnode/mnode/sdb/src/sdbFile.c
浏览文件 @ 39243cbb
......@@ -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) {
......
source/dnode/mnode/sdb/src/sdbHash.c
浏览文件 @ 39243cbb
......@@ -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;
}
......
source/dnode/vnode/src/inc/tq.h
浏览文件 @ 39243cbb
......@@ -149,7 +149,7 @@ int32_t tEncodeSTqHandle(SEncoder* pEncoder, const STqHandle* pHandle);
int32_t tDecodeSTqHandle(SDecoder* pDecoder, STqHandle* pHandle);
// tqRead
int32_t tqScan(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMetaRsp* pMetaRsp, STqOffsetVal* offset);
int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMetaRsp* pMetaRsp, STqOffsetVal* offset);
int32_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffsetVal* pOffset);
int64_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalCkHead** pHeadWithCkSum);
......@@ -181,8 +181,8 @@ int32_t tqOffsetDelete(STqOffsetStore* pStore, const char* subscribeKey)
int32_t tqOffsetCommitFile(STqOffsetStore* pStore);
// tqSink
void tqTableSink(SStreamTask* pTask, void* vnode, int64_t ver, void* data);
void tqTableSink1(SStreamTask* pTask, void* vnode, int64_t ver, void* data);
void tqSinkToTableMerge(SStreamTask* pTask, void* vnode, int64_t ver, void* data);
void tqSinkToTablePipeline(SStreamTask* pTask, void* vnode, int64_t ver, void* data);
// tqOffset
char* tqOffsetBuildFName(const char* path, int32_t ver);
......
source/dnode/vnode/src/meta/metaTable.c
浏览文件 @ 39243cbb
......@@ -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;
......
source/dnode/vnode/src/tq/tq.c
浏览文件 @ 39243cbb
......@@ -595,7 +595,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
tqInitTaosxRsp(&taosxRsp, pReq);
if (fetchOffsetNew.type != TMQ_OFFSET__LOG) {
tqScan(pTq, pHandle, &taosxRsp, &metaRsp, &fetchOffsetNew);
tqScanTaosx(pTq, pHandle, &taosxRsp, &metaRsp, &fetchOffsetNew);
if (metaRsp.metaRspLen > 0) {
if (tqSendMetaPollRsp(pTq, pMsg, pReq, &metaRsp) < 0) {
......@@ -924,7 +924,7 @@ int32_t tqExpandTask(STQ* pTq, SStreamTask* pTask) {
pTask->smaSink.smaSink = smaHandleRes;
} else if (pTask->outputType == TASK_OUTPUT__TABLE) {
pTask->tbSink.vnode = pTq->pVnode;
pTask->tbSink.tbSinkFunc = tqTableSink1;
pTask->tbSink.tbSinkFunc = tqSinkToTablePipeline;
ASSERT(pTask->tbSink.pSchemaWrapper);
ASSERT(pTask->tbSink.pSchemaWrapper->pSchema);
......
source/dnode/vnode/src/tq/tqExec.c
浏览文件 @ 39243cbb
......@@ -123,7 +123,7 @@ int32_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffs
return 0;
}
int32_t tqScan(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMetaRsp* pMetaRsp, STqOffsetVal* pOffset) {
int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMetaRsp* pMetaRsp, STqOffsetVal* pOffset) {
const STqExecHandle* pExec = &pHandle->execHandle;
qTaskInfo_t task = pExec->task;
......
source/dnode/vnode/src/tq/tqSink.c
浏览文件 @ 39243cbb
......@@ -284,7 +284,7 @@ SSubmitReq* tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchem
return ret;
}
void tqTableSink1(SStreamTask* pTask, void* vnode, int64_t ver, void* data) {
void tqSinkToTablePipeline(SStreamTask* pTask, void* vnode, int64_t ver, void* data) {
const SArray* pBlocks = (const SArray*)data;
SVnode* pVnode = (SVnode*)vnode;
int64_t suid = pTask->tbSink.stbUid;
......@@ -528,7 +528,7 @@ void tqTableSink1(SStreamTask* pTask, void* vnode, int64_t ver, void* data) {
taosArrayDestroy(tagArray);
}
void tqTableSink(SStreamTask* pTask, void* vnode, int64_t ver, void* data) {
void tqSinkToTableMerge(SStreamTask* pTask, void* vnode, int64_t ver, void* data) {
const SArray* pRes = (const SArray*)data;
SVnode* pVnode = (SVnode*)vnode;
SBatchDeleteReq deleteReq = {0};
......
source/dnode/vnode/src/vnd/vnodeQuery.c
浏览文件 @ 39243cbb
......@@ -330,6 +330,11 @@ int32_t vnodeGetBatchMeta(SVnode *pVnode, SRpcMsg *pMsg) {
rspSize += sizeof(int32_t);
offset = 0;
if (rspSize > MAX_META_BATCH_RSP_SIZE) {
code = TSDB_CODE_INVALID_MSG_LEN;
goto _exit;
}
pRsp = rpcMallocCont(rspSize);
if (pRsp == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
......
source/libs/catalog/src/catalog.c
浏览文件 @ 39243cbb
......@@ -302,9 +302,11 @@ int32_t ctgUpdateTbMeta(SCatalog* pCtg, STableMetaRsp* rspMsg, bool syncOp) {
_return:
taosMemoryFreeClear(output->tbMeta);
taosMemoryFreeClear(output);
if (output) {
taosMemoryFreeClear(output->tbMeta);
taosMemoryFreeClear(output);
}
CTG_RET(code);
}
......
source/libs/catalog/src/ctgAsync.c
浏览文件 @ 39243cbb
......@@ -252,7 +252,7 @@ int32_t ctgInitGetIndexTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
SCtgIndexCtx* ctx = task.taskCtx;
strcpy(ctx->indexFName, name);
tstrncpy(ctx->indexFName, name, sizeof(ctx->indexFName));
taosArrayPush(pJob->pTasks, &task);
......@@ -277,7 +277,7 @@ int32_t ctgInitGetUdfTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
SCtgUdfCtx* ctx = task.taskCtx;
strcpy(ctx->udfName, name);
tstrncpy(ctx->udfName, name, sizeof(ctx->udfName));
taosArrayPush(pJob->pTasks, &task);
......
source/libs/catalog/src/ctgCache.c
浏览文件 @ 39243cbb
......@@ -660,7 +660,7 @@ int32_t ctgDropDbCacheEnqueue(SCatalog *pCtg, const char *dbFName, int64_t dbId)
}
msg->pCtg = pCtg;
strncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
tstrncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
msg->dbId = dbId;
op->data = msg;
......@@ -693,7 +693,7 @@ int32_t ctgDropDbVgroupEnqueue(SCatalog *pCtg, const char *dbFName, bool syncOp)
}
msg->pCtg = pCtg;
strncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
tstrncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
op->data = msg;
......@@ -721,8 +721,8 @@ int32_t ctgDropStbMetaEnqueue(SCatalog *pCtg, const char *dbFName, int64_t dbId,
}
msg->pCtg = pCtg;
strncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
strncpy(msg->stbName, stbName, sizeof(msg->stbName));
tstrncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
tstrncpy(msg->stbName, stbName, sizeof(msg->stbName));
msg->dbId = dbId;
msg->suid = suid;
......@@ -751,8 +751,8 @@ int32_t ctgDropTbMetaEnqueue(SCatalog *pCtg, const char *dbFName, int64_t dbId,
}
msg->pCtg = pCtg;
strncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
strncpy(msg->tbName, tbName, sizeof(msg->tbName));
tstrncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
tstrncpy(msg->tbName, tbName, sizeof(msg->tbName));
msg->dbId = dbId;
op->data = msg;
......@@ -785,7 +785,7 @@ int32_t ctgUpdateVgroupEnqueue(SCatalog *pCtg, const char *dbFName, int64_t dbId
dbFName = p + 1;
}
strncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
tstrncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
msg->pCtg = pCtg;
msg->dbId = dbId;
msg->dbInfo = dbInfo;
......@@ -817,7 +817,8 @@ int32_t ctgUpdateTbMetaEnqueue(SCatalog *pCtg, STableMetaOutput *output, bool sy
char *p = strchr(output->dbFName, '.');
if (p && IS_SYS_DBNAME(p + 1)) {
memmove(output->dbFName, p + 1, strlen(p + 1));
int32_t len = strlen(p + 1);
memmove(output->dbFName, p + 1, len >= TSDB_DB_FNAME_LEN ? TSDB_DB_FNAME_LEN - 1 : len);
}
msg->pCtg = pCtg;
......@@ -852,7 +853,7 @@ int32_t ctgUpdateVgEpsetEnqueue(SCatalog *pCtg, char *dbFName, int32_t vgId, SEp
}
msg->pCtg = pCtg;
strcpy(msg->dbFName, dbFName);
tstrncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
msg->vgId = vgId;
msg->epSet = *pEpSet;
......@@ -1215,7 +1216,7 @@ int32_t ctgAddNewDBCache(SCatalog *pCtg, const char *dbFName, uint64_t dbId) {
CTG_CACHE_STAT_INC(numOfDb, 1);
SDbVgVersion vgVersion = {.dbId = newDBCache.dbId, .vgVersion = -1};
strncpy(vgVersion.dbFName, dbFName, sizeof(vgVersion.dbFName));
tstrncpy(vgVersion.dbFName, dbFName, sizeof(vgVersion.dbFName));
ctgDebug("db added to cache, dbFName:%s, dbId:0x%" PRIx64, dbFName, dbId);
......@@ -1331,8 +1332,8 @@ int32_t ctgUpdateRentStbVersion(SCatalog *pCtg, char *dbFName, char *tbName, uin
metaRent.smaVer = pCache->pIndex->version;
}
strcpy(metaRent.dbFName, dbFName);
strcpy(metaRent.stbName, tbName);
tstrncpy(metaRent.dbFName, dbFName, sizeof(metaRent.dbFName));
tstrncpy(metaRent.stbName, tbName, sizeof(metaRent.stbName));
CTG_ERR_RET(ctgMetaRentUpdate(&pCtg->stbRent, &metaRent, metaRent.suid, sizeof(SSTableVersion),
ctgStbVersionSortCompare, ctgStbVersionSearchCompare));
......@@ -1418,8 +1419,10 @@ int32_t ctgWriteTbMetaToCache(SCatalog *pCtg, SCtgDBCache *dbCache, char *dbFNam
ctgDebug("stb 0x%" PRIx64 " updated to cache, dbFName:%s, tbName:%s, tbType:%d", meta->suid, dbFName, tbName,
meta->tableType);
CTG_ERR_RET(ctgUpdateRentStbVersion(pCtg, dbFName, tbName, dbId, meta->suid, pCache));
if (pCache) {
CTG_ERR_RET(ctgUpdateRentStbVersion(pCtg, dbFName, tbName, dbId, meta->suid, pCache));
}
return TSDB_CODE_SUCCESS;
}
......@@ -1590,7 +1593,7 @@ int32_t ctgOpUpdateVgroup(SCtgCacheOperation *operation) {
dbCache = NULL;
strncpy(vgVersion.dbFName, dbFName, sizeof(vgVersion.dbFName));
tstrncpy(vgVersion.dbFName, dbFName, sizeof(vgVersion.dbFName));
CTG_ERR_JRET(ctgMetaRentUpdate(&msg->pCtg->dbRent, &vgVersion, vgVersion.dbId, sizeof(SDbVgVersion),
ctgDbVgVersionSortCompare, ctgDbVgVersionSearchCompare));
......@@ -1680,9 +1683,9 @@ int32_t ctgOpUpdateTbMeta(SCtgCacheOperation *operation) {
if (CTG_IS_META_TABLE(pMeta->metaType) || CTG_IS_META_BOTH(pMeta->metaType)) {
int32_t metaSize = CTG_META_SIZE(pMeta->tbMeta);
CTG_ERR_JRET(
ctgWriteTbMetaToCache(pCtg, dbCache, pMeta->dbFName, pMeta->dbId, pMeta->tbName, pMeta->tbMeta, metaSize));
code = ctgWriteTbMetaToCache(pCtg, dbCache, pMeta->dbFName, pMeta->dbId, pMeta->tbName, pMeta->tbMeta, metaSize);
pMeta->tbMeta = NULL;
CTG_ERR_JRET(code);
}
if (CTG_IS_META_CTABLE(pMeta->metaType) || CTG_IS_META_BOTH(pMeta->metaType)) {
......@@ -1697,10 +1700,8 @@ int32_t ctgOpUpdateTbMeta(SCtgCacheOperation *operation) {
_return:
if (pMeta) {
taosMemoryFreeClear(pMeta->tbMeta);
taosMemoryFreeClear(pMeta);
}
taosMemoryFreeClear(pMeta->tbMeta);
taosMemoryFreeClear(pMeta);
taosMemoryFreeClear(msg);
......
source/libs/command/src/command.c
浏览文件 @ 39243cbb
......@@ -361,7 +361,12 @@ int32_t appendTagValues(char* buf, int32_t* len, STableCfg* pCfg) {
SArray* pTagVals = NULL;
STag* pTag = (STag*)pCfg->pTags;
if (pCfg->pTags && tTagIsJson(pTag)) {
if (NULL == pCfg->pTags || pCfg->numOfTags <= 0) {
qError("tag missed in table cfg, pointer:%p, numOfTags:%d", pCfg->pTags, pCfg->numOfTags);
return TSDB_CODE_APP_ERROR;
}
if (tTagIsJson(pTag)) {
char* pJson = parseTagDatatoJson(pTag);
if (pJson) {
*len += sprintf(buf + VARSTR_HEADER_SIZE + *len, "%s", pJson);
......
source/libs/executor/inc/executorimpl.h
浏览文件 @ 39243cbb
......@@ -53,6 +53,11 @@ typedef int32_t (*__block_search_fn_t)(char* data, int32_t num, int64_t key, int
#define NEEDTO_COMPRESS_QUERY(size) ((size) > tsCompressColData ? 1 : 0)
#define IS_VALID_SESSION_WIN(winInfo) ((winInfo).sessionWin.win.skey > 0)
#define SET_SESSION_WIN_INVALID(winInfo) ((winInfo).sessionWin.win.skey = INT64_MIN)
#define IS_INVALID_SESSION_WIN_KEY(winKey) ((winKey).win.skey <= 0)
#define SET_SESSION_WIN_KEY_INVALID(pWinKey) ((pWinKey)->win.skey = INT64_MIN)
enum {
// when this task starts to execute, this status will set
TASK_NOT_COMPLETED = 0x1u,
......@@ -434,15 +439,15 @@ typedef struct SCatchSupporter {
} SCatchSupporter;
typedef struct SStreamAggSupporter {
SHashObj* pResultRows;
SArray* pCurWins;
int32_t valueSize;
int32_t keySize;
char* pKeyBuf; // window key buffer
SDiskbasedBuf* pResultBuf; // query result buffer based on blocked-wised disk file
int32_t resultRowSize; // the result buffer size for each result row, with the meta data size for each row
int32_t currentPageId; // buffer page that is active
SSDataBlock* pScanBlock;
int32_t resultRowSize; // the result buffer size for each result row, with the meta data size for each row
SSDataBlock* pScanBlock;
SStreamState* pState;
int64_t gap; // stream session window gap
SqlFunctionCtx* pDummyCtx; // for combine
SSHashObj* pResultRows;
int32_t stateKeySize;
int16_t stateKeyType;
SDiskbasedBuf* pResultBuf;
} SStreamAggSupporter;
typedef struct SWindowSupporter {
......@@ -736,42 +741,54 @@ typedef struct SSessionAggOperatorInfo {
} SSessionAggOperatorInfo;
typedef struct SResultWindowInfo {
SResultRowPosition pos;
STimeWindow win;
uint64_t groupId;
void* pOutputBuf;
SSessionKey sessionWin;
bool isOutput;
bool isClosed;
} SResultWindowInfo;
typedef struct SStateWindowInfo {
SResultWindowInfo winInfo;
SStateKeys stateKey;
SStateKeys* pStateKey;
} SStateWindowInfo;
typedef struct SStreamSessionAggOperatorInfo {
SOptrBasicInfo binfo;
SStreamAggSupporter streamAggSup;
SExprSupp scalarSupp; // supporter for perform scalar function
SExprSupp scalarSupp; // supporter for perform scalar function
SGroupResInfo groupResInfo;
int64_t gap; // session window gap
int32_t primaryTsIndex; // primary timestamp slot id
int32_t endTsIndex; // window end timestamp slot id
int32_t order; // current SSDataBlock scan order
STimeWindowAggSupp twAggSup;
SSDataBlock* pWinBlock; // window result
SqlFunctionCtx* pDummyCtx; // for combine
SSDataBlock* pDelRes; // delete result
SSDataBlock* pUpdateRes; // update window
SSDataBlock* pWinBlock; // window result
SSDataBlock* pDelRes; // delete result
SSDataBlock* pUpdateRes; // update window
bool returnUpdate;
SHashObj* pStDeleted;
SSHashObj* pStDeleted;
void* pDelIterator;
SArray* pChildren; // cache for children's result; final stream operator
SPhysiNode* pPhyNode; // create new child
SArray* pChildren; // cache for children's result; final stream operator
SPhysiNode* pPhyNode; // create new child
bool isFinal;
bool ignoreExpiredData;
SHashObj* pGroupIdTbNameMap;
} SStreamSessionAggOperatorInfo;
typedef struct SStreamStateAggOperatorInfo {
SOptrBasicInfo binfo;
SStreamAggSupporter streamAggSup;
SExprSupp scalarSupp; // supporter for perform scalar function
SGroupResInfo groupResInfo;
int32_t primaryTsIndex; // primary timestamp slot id
STimeWindowAggSupp twAggSup;
SColumn stateCol;
SSDataBlock* pDelRes;
SSHashObj* pSeDeleted;
void* pDelIterator;
SArray* pChildren; // cache for children's result;
bool ignoreExpiredData;
SHashObj* pGroupIdTbNameMap;
} SStreamStateAggOperatorInfo;
typedef struct SStreamPartitionOperatorInfo {
SOptrBasicInfo binfo;
SPartitionBySupporter partitionSup;
......@@ -834,24 +851,6 @@ typedef struct SStateWindowOperatorInfo {
const SNode* pCondition;
} SStateWindowOperatorInfo;
typedef struct SStreamStateAggOperatorInfo {
SOptrBasicInfo binfo;
SStreamAggSupporter streamAggSup;
SExprSupp scalarSupp; // supporter for perform scalar function
SGroupResInfo groupResInfo;
int32_t primaryTsIndex; // primary timestamp slot id
int32_t order; // current SSDataBlock scan order
STimeWindowAggSupp twAggSup;
SColumn stateCol;
SqlFunctionCtx* pDummyCtx; // for combine
SSDataBlock* pDelRes;
SHashObj* pSeDeleted;
void* pDelIterator;
SArray* pChildren; // cache for children's result;
bool ignoreExpiredData;
SHashObj* pGroupIdTbNameMap;
} SStreamStateAggOperatorInfo;
typedef struct SSortOperatorInfo {
SOptrBasicInfo binfo;
uint32_t sortBufSize; // max buffer size for in-memory sort
......@@ -1064,13 +1063,8 @@ STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowI
int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimaryColumn, int32_t startPos, TSKEY ekey,
__block_search_fn_t searchFn, STableQueryInfo* item, int32_t order);
int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order);
int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, const char* pKey, SqlFunctionCtx* pCtx, int32_t numOfOutput,
int32_t size);
SResultRow* getNewResultRow(SDiskbasedBuf* pResultBuf, int32_t* currentPageId, int32_t interBufSize);
SResultWindowInfo* getSessionTimeWindow(SStreamAggSupporter* pAggSup, TSKEY startTs, TSKEY endTs, uint64_t groupId,
int64_t gap, int32_t* pIndex);
SResultWindowInfo* getCurSessionWindow(SStreamAggSupporter* pAggSup, TSKEY startTs, TSKEY endTs, uint64_t groupId,
int64_t gap, int32_t* pIndex);
void getCurSessionWindow(SStreamAggSupporter* pAggSup, TSKEY startTs, TSKEY endTs, uint64_t groupId, SSessionKey* pKey);
bool isInTimeWindow(STimeWindow* pWin, TSKEY ts, int64_t gap);
bool functionNeedToExecute(SqlFunctionCtx* pCtx);
bool isOverdue(TSKEY ts, STimeWindowAggSupp* pSup);
......@@ -1100,6 +1094,9 @@ int32_t generateGroupIdMap(STableListInfo* pTableListInfo, SReadHandle* pHandle,
void* destroySqlFunctionCtx(SqlFunctionCtx* pCtx, int32_t numOfOutput);
int32_t buildDataBlockFromGroupRes(SOperatorInfo* pOperator, SStreamState* pState, SSDataBlock* pBlock, SExprSupp* pSup,
SGroupResInfo* pGroupResInfo);
int32_t saveSessionDiscBuf(SStreamState* pState, SSessionKey* key, void* buf, int32_t size);
int32_t buildSessionResultDataBlock(SExecTaskInfo* pTaskInfo, SStreamState* pState, SSDataBlock* pBlock,
SExprSupp* pSup, SGroupResInfo* pGroupResInfo);
int32_t setOutputBuf(SStreamState* pState, STimeWindow* win, SResultRow** pResult, int64_t tableGroupId, SqlFunctionCtx* pCtx,
int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup);
int32_t releaseOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult);
......
source/libs/executor/src/dataDispatcher.c
浏览文件 @ 39243cbb
......@@ -143,9 +143,15 @@ static int32_t getStatus(SDataDispatchHandle* pDispatcher) {
static int32_t putDataBlock(SDataSinkHandle* pHandle, const SInputData* pInput, bool* pContinue) {
SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;
SDataDispatchBuf* pBuf = taosAllocateQitem(sizeof(SDataDispatchBuf), DEF_QITEM);
if (NULL == pBuf || !allocBuf(pDispatcher, pInput, pBuf)) {
if (NULL == pBuf) {
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
if (!allocBuf(pDispatcher, pInput, pBuf)) {
taosFreeQitem(pBuf);
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
toDataCacheEntry(pDispatcher, pInput, pBuf);
taosWriteQitem(pDispatcher->pDataBlocks, pBuf);
*pContinue = (DS_BUF_LOW == updateStatus(pDispatcher) ? true : false);
......
source/libs/executor/src/dataInserter.c
浏览文件 @ 39243cbb
......@@ -323,7 +323,7 @@ int32_t createDataInserter(SDataSinkManager* pManager, const SDataSinkNode* pDat
int32_t code =
tsdbGetTableSchema(inserter->pParam->readHandle->vnode, pInserterNode->tableId, &inserter->pSchema, &suid);
if (code) {
destroyDataSinker((SDataSinkHandle*)pInserterNode);
destroyDataSinker((SDataSinkHandle*)inserter);
return code;
}
......
source/libs/executor/src/executorimpl.c
浏览文件 @ 39243cbb
......@@ -4192,42 +4192,6 @@ int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SArray* pExecInf
return TSDB_CODE_SUCCESS;
}
int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, const char* pKey, SqlFunctionCtx* pCtx, int32_t numOfOutput,
int32_t size) {
pSup->currentPageId = -1;
pSup->resultRowSize = getResultRowSize(pCtx, numOfOutput);
pSup->keySize = sizeof(int64_t) + sizeof(TSKEY);
pSup->pKeyBuf = taosMemoryCalloc(1, pSup->keySize);
_hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
pSup->pResultRows = taosHashInit(1024, hashFn, false, HASH_NO_LOCK);
if (pSup->pKeyBuf == NULL || pSup->pResultRows == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pSup->valueSize = size;
pSup->pScanBlock = createSpecialDataBlock(STREAM_CLEAR);
int32_t pageSize = 4096;
while (pageSize < pSup->resultRowSize * 4) {
pageSize <<= 1u;
}
// at least four pages need to be in buffer
int32_t bufSize = 4096 * 256;
if (bufSize <= pageSize) {
bufSize = pageSize * 4;
}
if (!osTempSpaceAvailable()) {
terrno = TSDB_CODE_NO_AVAIL_DISK;
qError("Init stream agg supporter failed since %s", terrstr(terrno));
return terrno;
}
int32_t code = createDiskbasedBuf(&pSup->pResultBuf, pageSize, bufSize, pKey, tsTempDir);
for (int32_t i = 0; i < numOfOutput; ++i) {
pCtx[i].saveHandle.pBuf = pSup->pResultBuf;
}
return code;
}
int32_t setOutputBuf(SStreamState* pState, STimeWindow* win, SResultRow** pResult, int64_t tableGroupId,
SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup) {
SWinKey key = {
......@@ -4237,7 +4201,6 @@ int32_t setOutputBuf(SStreamState* pState, STimeWindow* win, SResultRow** pResul
char* value = NULL;
int32_t size = pAggSup->resultRowSize;
tSimpleHashPut(pAggSup->pResultRowHashTable, &key, sizeof(SWinKey), NULL, 0);
if (streamStateAddIfNotExist(pState, &key, (void**)&value, &size) < 0) {
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
......@@ -4342,3 +4305,82 @@ int32_t buildDataBlockFromGroupRes(SOperatorInfo* pOperator, SStreamState* pStat
blockDataUpdateTsWindow(pBlock, 0);
return TSDB_CODE_SUCCESS;
}
int32_t saveSessionDiscBuf(SStreamState* pState, SSessionKey* key, void* buf, int32_t size) {
streamStateSessionPut(pState, key, (const void*)buf, size);
releaseOutputBuf(pState, NULL, (SResultRow*)buf);
return TSDB_CODE_SUCCESS;
}
int32_t buildSessionResultDataBlock(SExecTaskInfo* pTaskInfo, SStreamState* pState, SSDataBlock* pBlock,
SExprSupp* pSup, SGroupResInfo* pGroupResInfo) {
SExprInfo* pExprInfo = pSup->pExprInfo;
int32_t numOfExprs = pSup->numOfExprs;
int32_t* rowEntryOffset = pSup->rowEntryInfoOffset;
SqlFunctionCtx* pCtx = pSup->pCtx;
int32_t numOfRows = getNumOfTotalRes(pGroupResInfo);
for (int32_t i = pGroupResInfo->index; i < numOfRows; i += 1) {
SSessionKey* pKey = taosArrayGet(pGroupResInfo->pRows, i);
int32_t size = 0;
void* pVal = NULL;
int32_t code = streamStateSessionGet(pState, pKey, &pVal, &size);
ASSERT(code == 0);
SResultRow* pRow = (SResultRow*)pVal;
doUpdateNumOfRows(pCtx, pRow, numOfExprs, rowEntryOffset);
// no results, continue to check the next one
if (pRow->numOfRows == 0) {
pGroupResInfo->index += 1;
releaseOutputBuf(pState, NULL, pRow);
continue;
}
if (pBlock->info.groupId == 0) {
pBlock->info.groupId = pKey->groupId;
} else {
// current value belongs to different group, it can't be packed into one datablock
if (pBlock->info.groupId != pKey->groupId) {
releaseOutputBuf(pState, NULL, pRow);
break;
}
}
if (pBlock->info.rows + pRow->numOfRows > pBlock->info.capacity) {
ASSERT(pBlock->info.rows > 0);
releaseOutputBuf(pState, NULL, pRow);
break;
}
pGroupResInfo->index += 1;
for (int32_t j = 0; j < numOfExprs; ++j) {
int32_t slotId = pExprInfo[j].base.resSchema.slotId;
pCtx[j].resultInfo = getResultEntryInfo(pRow, j, rowEntryOffset);
if (pCtx[j].fpSet.finalize) {
int32_t code1 = pCtx[j].fpSet.finalize(&pCtx[j], pBlock);
if (TAOS_FAILED(code1)) {
qError("%s build result data block error, code %s", GET_TASKID(pTaskInfo), tstrerror(code1));
T_LONG_JMP(pTaskInfo->env, code1);
}
} else if (strcmp(pCtx[j].pExpr->pExpr->_function.functionName, "_select_value") == 0) {
// do nothing, todo refactor
} else {
// expand the result into multiple rows. E.g., _wstart, top(k, 20)
// the _wstart needs to copy to 20 following rows, since the results of top-k expands to 20 different rows.
SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, slotId);
char* in = GET_ROWCELL_INTERBUF(pCtx[j].resultInfo);
for (int32_t k = 0; k < pRow->numOfRows; ++k) {
colDataAppend(pColInfoData, pBlock->info.rows + k, in, pCtx[j].resultInfo->isNullRes);
}
}
}
pBlock->info.rows += pRow->numOfRows;
// saveSessionDiscBuf(pState, pKey, pVal, size);
releaseOutputBuf(pState, NULL, pRow);
}
blockDataUpdateTsWindow(pBlock, 0);
return TSDB_CODE_SUCCESS;
}
\ No newline at end of file
source/libs/executor/src/scanoperator.c
浏览文件 @ 39243cbb
......@@ -1190,23 +1190,22 @@ static int32_t generateSessionScanRange(SStreamScanInfo* pInfo, SSDataBlock* pSr
SColumnInfoData* pDestGpCol = taosArrayGet(pDestBlock->pDataBlock, GROUPID_COLUMN_INDEX);
SColumnInfoData* pDestCalStartTsCol = taosArrayGet(pDestBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);
SColumnInfoData* pDestCalEndTsCol = taosArrayGet(pDestBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);
int32_t dummy = 0;
int64_t version = pSrcBlock->info.version - 1;
for (int32_t i = 0; i < pSrcBlock->info.rows; i++) {
uint64_t groupId = getGroupIdByData(pInfo, uidCol[i], startData[i], version);
// gap must be 0.
SResultWindowInfo* pStartWin =
getCurSessionWindow(pInfo->windowSup.pStreamAggSup, startData[i], endData[i], groupId, 0, &dummy);
if (!pStartWin) {
SSessionKey startWin = {0};
getCurSessionWindow(pInfo->windowSup.pStreamAggSup, startData[i], endData[i], groupId, &startWin);
if (IS_INVALID_SESSION_WIN_KEY(startWin)) {
// window has been closed.
continue;
}
SResultWindowInfo* pEndWin =
getCurSessionWindow(pInfo->windowSup.pStreamAggSup, endData[i], endData[i], groupId, 0, &dummy);
ASSERT(pEndWin);
TSKEY ts = INT64_MIN;
colDataAppend(pDestStartCol, i, (const char*)&pStartWin->win.skey, false);
colDataAppend(pDestEndCol, i, (const char*)&pEndWin->win.ekey, false);
SSessionKey endWin = {0};
getCurSessionWindow(pInfo->windowSup.pStreamAggSup, endData[i], endData[i], groupId, &endWin);
ASSERT(!IS_INVALID_SESSION_WIN_KEY(endWin));
colDataAppend(pDestStartCol, i, (const char*)&startWin.win.skey, false);
colDataAppend(pDestEndCol, i, (const char*)&endWin.win.ekey, false);
colDataAppendNULL(pDestUidCol, i);
colDataAppend(pDestGpCol, i, (const char*)&groupId, false);
colDataAppendNULL(pDestCalStartTsCol, i);
......
source/libs/executor/src/tfill.c
浏览文件 @ 39243cbb
......@@ -693,7 +693,7 @@ void* destroyStreamFillSupporter(SStreamFillSupporter* pFillSup) {
pFillSup->pAllColInfo = destroyFillColumnInfo(pFillSup->pAllColInfo, pFillSup->numOfFillCols, pFillSup->numOfAllCols);
tSimpleHashCleanup(pFillSup->pResMap);
pFillSup->pResMap = NULL;
streamStateReleaseBuf(NULL, NULL, pFillSup->cur.pRowVal);
releaseOutputBuf(NULL, NULL, (SResultRow*)pFillSup->cur.pRowVal);
pFillSup->cur.pRowVal = NULL;
taosMemoryFree(pFillSup);
......@@ -736,7 +736,7 @@ static void resetFillWindow(SResultRowData* pRowData) {
void resetPrevAndNextWindow(SStreamFillSupporter* pFillSup, SStreamState* pState) {
resetFillWindow(&pFillSup->prev);
streamStateReleaseBuf(NULL, NULL, pFillSup->cur.pRowVal);
releaseOutputBuf(NULL, NULL, (SResultRow*)pFillSup->cur.pRowVal);
resetFillWindow(&pFillSup->cur);
resetFillWindow(&pFillSup->next);
resetFillWindow(&pFillSup->nextNext);
......
source/libs/function/src/builtinsimpl.c
浏览文件 @ 39243cbb
......@@ -2527,6 +2527,8 @@ int32_t apercentileFunction(SqlFunctionCtx* pCtx) {
int32_t start = pInput->startRowIndex;
if (pInfo->algo == APERCT_ALGO_TDIGEST) {
buildTDigestInfo(pInfo);
tdigestAutoFill(pInfo->pTDigest, COMPRESSION);
for (int32_t i = start; i < pInput->numOfRows + start; ++i) {
if (colDataIsNull_f(pCol->nullbitmap, i)) {
continue;
......@@ -2540,12 +2542,11 @@ int32_t apercentileFunction(SqlFunctionCtx* pCtx) {
tdigestAdd(pInfo->pTDigest, v, w);
}
} else {
qDebug("%s before add %d elements into histogram, total:%d, numOfEntry:%d, pHisto:%p, elems: %p", __FUNCTION__,
numOfElems, pInfo->pHisto->numOfElems, pInfo->pHisto->numOfEntries, pInfo->pHisto, pInfo->pHisto->elems);
// might be a race condition here that pHisto can be overwritten or setup function
// has not been called, need to relink the buffer pHisto points to.
buildHistogramInfo(pInfo);
qDebug("%s before add %d elements into histogram, total:%d, numOfEntry:%d, pHisto:%p, elems: %p", __FUNCTION__,
numOfElems, pInfo->pHisto->numOfElems, pInfo->pHisto->numOfEntries, pInfo->pHisto, pInfo->pHisto->elems);
for (int32_t i = start; i < pInput->numOfRows + start; ++i) {
if (colDataIsNull_f(pCol->nullbitmap, i)) {
continue;
......@@ -2579,8 +2580,9 @@ static void apercentileTransferInfo(SAPercentileInfo* pInput, SAPercentileInfo*
buildTDigestInfo(pOutput);
TDigest* pTDigest = pOutput->pTDigest;
tdigestAutoFill(pTDigest, COMPRESSION);
if (pTDigest->num_centroids <= 0) {
if (pTDigest->num_centroids <= 0 && pTDigest->num_buffered_pts == 0) {
memcpy(pTDigest, pInput->pTDigest, (size_t)TDIGEST_SIZE(COMPRESSION));
tdigestAutoFill(pTDigest, COMPRESSION);
} else {
......@@ -2652,6 +2654,7 @@ int32_t apercentileFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
if (pInfo->algo == APERCT_ALGO_TDIGEST) {
buildTDigestInfo(pInfo);
tdigestAutoFill(pInfo->pTDigest, COMPRESSION);
if (pInfo->pTDigest->size > 0) {
pInfo->result = tdigestQuantile(pInfo->pTDigest, pInfo->percent / 100);
} else { // no need to free
......@@ -5356,16 +5359,14 @@ int32_t modeFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
int32_t maxCount = 0;
for (int32_t i = 0; i < pInfo->numOfPoints; ++i) {
SModeItem* pItem = (SModeItem*)(pInfo->pItems + i * (sizeof(SModeItem) + pInfo->colBytes));
if (pItem->count > maxCount) {
if (pItem->count >= maxCount) {
maxCount = pItem->count;
resIndex = i;
} else if (pItem->count == maxCount) {
resIndex = -1;
}
}
SModeItem* pResItem = (SModeItem*)(pInfo->pItems + resIndex * (sizeof(SModeItem) + pInfo->colBytes));
colDataAppend(pCol, currentRow, pResItem->data, (resIndex == -1) ? true : false);
colDataAppend(pCol, currentRow, pResItem->data, (maxCount == 0) ? true : false);
return pResInfo->numOfRes;
}
......
source/libs/planner/src/planOptimizer.c
浏览文件 @ 39243cbb
......@@ -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;
}
......
source/libs/planner/test/planPartByTest.cpp
浏览文件 @ 39243cbb
......@@ -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) {
......
source/libs/qcom/src/queryUtil.c
浏览文件 @ 39243cbb
......@@ -357,8 +357,7 @@ char* parseTagDatatoJson(void* p) {
for (int j = 0; j < nCols; ++j) {
STagVal* pTagVal = (STagVal*)taosArrayGet(pTagVals, j);
// json key encode by binary
memset(tagJsonKey, 0, sizeof(tagJsonKey));
memcpy(tagJsonKey, pTagVal->pKey, strlen(pTagVal->pKey));
tstrncpy(tagJsonKey, pTagVal->pKey, sizeof(tagJsonKey));
// json value
char type = pTagVal->type;
if (type == TSDB_DATA_TYPE_NULL) {
......
source/libs/qcom/src/querymsg.c
浏览文件 @ 39243cbb
......@@ -173,7 +173,7 @@ int32_t queryBuildGetDBCfgMsg(void *input, char **msg, int32_t msgSize, int32_t
}
SDbCfgReq dbCfgReq = {0};
strcpy(dbCfgReq.db, input);
strncpy(dbCfgReq.db, input, sizeof(dbCfgReq.db) - 1);
int32_t bufLen = tSerializeSDbCfgReq(NULL, 0, &dbCfgReq);
void *pBuf = (*mallcFp)(bufLen);
......@@ -191,7 +191,7 @@ int32_t queryBuildGetIndexMsg(void *input, char **msg, int32_t msgSize, int32_t
}
SUserIndexReq indexReq = {0};
strcpy(indexReq.indexFName, input);
strncpy(indexReq.indexFName, input, sizeof(indexReq.indexFName) - 1);
int32_t bufLen = tSerializeSUserIndexReq(NULL, 0, &indexReq);
void *pBuf = (*mallcFp)(bufLen);
......@@ -233,7 +233,7 @@ int32_t queryBuildGetUserAuthMsg(void *input, char **msg, int32_t msgSize, int32
}
SGetUserAuthReq req = {0};
strncpy(req.user, input, sizeof(req.user));
strncpy(req.user, input, sizeof(req.user) - 1);
int32_t bufLen = tSerializeSGetUserAuthReq(NULL, 0, &req);
void *pBuf = (*mallcFp)(bufLen);
......@@ -251,7 +251,7 @@ int32_t queryBuildGetTbIndexMsg(void *input, char **msg, int32_t msgSize, int32_
}
STableIndexReq indexReq = {0};
strcpy(indexReq.tbFName, input);
strncpy(indexReq.tbFName, input, sizeof(indexReq.tbFName) - 1);
int32_t bufLen = tSerializeSTableIndexReq(NULL, 0, &indexReq);
void *pBuf = (*mallcFp)(bufLen);
......@@ -271,8 +271,8 @@ int32_t queryBuildGetTbCfgMsg(void *input, char **msg, int32_t msgSize, int32_t
SBuildTableInput *pInput = input;
STableCfgReq cfgReq = {0};
cfgReq.header.vgId = pInput->vgId;
strncpy(cfgReq.dbFName, pInput->dbFName, sizeof(cfgReq.dbFName));
strncpy(cfgReq.tbName, pInput->tbName, sizeof(cfgReq.tbName));
strncpy(cfgReq.dbFName, pInput->dbFName, sizeof(cfgReq.dbFName) - 1);
strncpy(cfgReq.tbName, pInput->tbName, sizeof(cfgReq.tbName) - 1);
int32_t bufLen = tSerializeSTableCfgReq(NULL, 0, &cfgReq);
void *pBuf = (*mallcFp)(bufLen);
......
source/libs/qworker/src/qwUtil.c
浏览文件 @ 39243cbb
......@@ -412,7 +412,7 @@ void qwSetHbParam(int64_t refId, SQWHbParam **pParam) {
while (true) {
paramIdx = atomic_load_32(&gQwMgmt.paramIdx);
if (paramIdx == tListLen(gQwMgmt.param)) {
newParamIdx = 0;
newParamIdx = 1;
} else {
newParamIdx = paramIdx + 1;
}
......@@ -422,6 +422,10 @@ void qwSetHbParam(int64_t refId, SQWHbParam **pParam) {
}
}
if (paramIdx == tListLen(gQwMgmt.param)) {
paramIdx = 0;
}
gQwMgmt.param[paramIdx].qwrId = gQwMgmt.qwRef;
gQwMgmt.param[paramIdx].refId = refId;
......
source/libs/qworker/src/qworker.c
浏览文件 @ 39243cbb
......@@ -398,7 +398,6 @@ int32_t qwHandlePrePhaseEvents(QW_FPARAMS_DEF, int8_t phase, SQWPhaseInput *inpu
if (QW_EVENT_PROCESSED(ctx, QW_EVENT_DROP)) {
QW_TASK_ELOG("task already dropped at wrong phase %s", qwPhaseStr(phase));
QW_ERR_JRET(TSDB_CODE_QRY_TASK_STATUS_ERROR);
break;
}
if (QW_EVENT_RECEIVED(ctx, QW_EVENT_DROP)) {
......
source/libs/scalar/src/scalar.c
浏览文件 @ 39243cbb
......@@ -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);
}
......
source/libs/scalar/src/sclvector.c
浏览文件 @ 39243cbb
......@@ -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};
......
source/libs/scheduler/src/schTask.c
浏览文件 @ 39243cbb
......@@ -430,7 +430,8 @@ int32_t schHandleRedirect(SSchJob *pJob, SSchTask *pTask, SDataBuf *pData, int32
if (SCH_IS_DATA_BIND_TASK(pTask)) {
if (NULL == pData->pEpSet) {
SCH_TASK_ELOG("no epset updated while got error %s", tstrerror(rspCode));
SCH_ERR_JRET(rspCode);
code = rspCode;
goto _return;
}
}
......
source/libs/stream/src/streamState.c
浏览文件 @ 39243cbb
......@@ -24,6 +24,40 @@ typedef struct SStateKey {
int64_t opNum;
} SStateKey;
typedef struct SStateSessionKey {
SSessionKey key;
int64_t opNum;
} SStateSessionKey;
static inline int sessionKeyCmpr(const SSessionKey* pWin1, const SSessionKey* pWin2) {
if (pWin1->groupId > pWin2->groupId) {
return 1;
} else if (pWin1->groupId < pWin2->groupId) {
return -1;
}
if (pWin1->win.skey > pWin2->win.ekey) {
return 1;
} else if (pWin1->win.ekey < pWin2->win.skey) {
return -1;
}
return 0;
}
static inline int stateSessionKeyCmpr(const void* pKey1, int kLen1, const void* pKey2, int kLen2) {
SStateSessionKey* pWin1 = (SStateSessionKey*)pKey1;
SStateSessionKey* pWin2 = (SStateSessionKey*)pKey2;
if (pWin1->opNum > pWin2->opNum) {
return 1;
} else if (pWin1->opNum < pWin2->opNum) {
return -1;
}
return sessionKeyCmpr(&pWin1->key, &pWin2->key);
}
static inline int stateKeyCmpr(const void* pKey1, int kLen1, const void* pKey2, int kLen2) {
SStateKey* pWin1 = (SStateKey*)pKey1;
SStateKey* pWin2 = (SStateKey*)pKey2;
......@@ -79,6 +113,11 @@ SStreamState* streamStateOpen(char* path, SStreamTask* pTask, bool specPath, int
goto _err;
}
if (tdbTbOpen("session.state.db", sizeof(SStateSessionKey), -1, stateSessionKeyCmpr, pState->db,
&pState->pSessionStateDb) < 0) {
goto _err;
}
if (tdbTbOpen("func.state.db", sizeof(STupleKey), -1, STupleKeyCmpr, pState->db, &pState->pFuncStateDb) < 0) {
goto _err;
}
......@@ -95,6 +134,7 @@ _err:
tdbTbClose(pState->pStateDb);
tdbTbClose(pState->pFuncStateDb);
tdbTbClose(pState->pFillStateDb);
tdbTbClose(pState->pSessionStateDb);
tdbClose(pState->db);
taosMemoryFree(pState);
return NULL;
......@@ -105,6 +145,7 @@ void streamStateClose(SStreamState* pState) {
tdbTbClose(pState->pStateDb);
tdbTbClose(pState->pFuncStateDb);
tdbTbClose(pState->pFillStateDb);
tdbTbClose(pState->pSessionStateDb);
tdbClose(pState->db);
taosMemoryFree(pState);
......@@ -241,11 +282,11 @@ SStreamStateCur* streamStateGetCur(SStreamState* pState, const SWinKey* key) {
if (pCur == NULL) return NULL;
tdbTbcOpen(pState->pStateDb, &pCur->pCur, NULL);
int32_t c;
int32_t c = 0;
SStateKey sKey = {.key = *key, .opNum = pState->number};
tdbTbcMoveTo(pCur->pCur, &sKey, sizeof(SStateKey), &c);
if (c != 0) {
taosMemoryFree(pCur);
streamStateFreeCur(pCur);
return NULL;
}
pCur->number = pState->number;
......@@ -257,7 +298,7 @@ SStreamStateCur* streamStateFillGetCur(SStreamState* pState, const SWinKey* key)
if (pCur == NULL) return NULL;
tdbTbcOpen(pState->pFillStateDb, &pCur->pCur, NULL);
int32_t c;
int32_t c = 0;
tdbTbcMoveTo(pCur->pCur, key, sizeof(SWinKey), &c);
if (c != 0) {
streamStateFreeCur(pCur);
......@@ -348,21 +389,21 @@ SStreamStateCur* streamStateSeekKeyNext(SStreamState* pState, const SWinKey* key
}
pCur->number = pState->number;
if (tdbTbcOpen(pState->pStateDb, &pCur->pCur, NULL) < 0) {
taosMemoryFree(pCur);
streamStateFreeCur(pCur);
return NULL;
}
SStateKey sKey = {.key = *key, .opNum = pState->number};
int32_t c;
int32_t c = 0;
if (tdbTbcMoveTo(pCur->pCur, &sKey, sizeof(SStateKey), &c) < 0) {
tdbTbcClose(pCur->pCur);
taosMemoryFree(pCur);
streamStateFreeCur(pCur);
return NULL;
}
if (c > 0) return pCur;
if (tdbTbcMoveToNext(pCur->pCur) < 0) {
taosMemoryFree(pCur);
streamStateFreeCur(pCur);
return NULL;
}
......@@ -375,20 +416,20 @@ SStreamStateCur* streamStateFillSeekKeyNext(SStreamState* pState, const SWinKey*
return NULL;
}
if (tdbTbcOpen(pState->pFillStateDb, &pCur->pCur, NULL) < 0) {
taosMemoryFree(pCur);
streamStateFreeCur(pCur);
return NULL;
}
int32_t c;
int32_t c = 0;
if (tdbTbcMoveTo(pCur->pCur, key, sizeof(SWinKey), &c) < 0) {
tdbTbcClose(pCur->pCur);
taosMemoryFree(pCur);
streamStateFreeCur(pCur);
return NULL;
}
if (c > 0) return pCur;
if (tdbTbcMoveToNext(pCur->pCur) < 0) {
taosMemoryFree(pCur);
streamStateFreeCur(pCur);
return NULL;
}
......@@ -401,20 +442,20 @@ SStreamStateCur* streamStateFillSeekKeyPrev(SStreamState* pState, const SWinKey*
return NULL;
}
if (tdbTbcOpen(pState->pFillStateDb, &pCur->pCur, NULL) < 0) {
taosMemoryFree(pCur);
streamStateFreeCur(pCur);
return NULL;
}
int32_t c;
int32_t c = 0;
if (tdbTbcMoveTo(pCur->pCur, key, sizeof(SWinKey), &c) < 0) {
tdbTbcClose(pCur->pCur);
taosMemoryFree(pCur);
streamStateFreeCur(pCur);
return NULL;
}
if (c < 0) return pCur;
if (tdbTbcMoveToPrev(pCur->pCur) < 0) {
taosMemoryFree(pCur);
streamStateFreeCur(pCur);
return NULL;
}
......@@ -445,3 +486,229 @@ void streamStateFreeCur(SStreamStateCur* pCur) {
}
void streamFreeVal(void* val) { tdbFree(val); }
int32_t streamStateSessionPut(SStreamState* pState, const SSessionKey* key, const void* value, int32_t vLen) {
SStateSessionKey sKey = {.key = *key, .opNum = pState->number};
return tdbTbUpsert(pState->pSessionStateDb, &sKey, sizeof(SStateSessionKey), value, vLen, &pState->txn);
}
SStreamStateCur* streamStateSessionGetRanomCur(SStreamState* pState, const SSessionKey* key) {
SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
if (pCur == NULL) return NULL;
tdbTbcOpen(pState->pSessionStateDb, &pCur->pCur, NULL);
int32_t c = 0;
SStateSessionKey sKey = {.key = *key, .opNum = pState->number};
tdbTbcMoveTo(pCur->pCur, &sKey, sizeof(SStateSessionKey), &c);
if (c != 0) {
streamStateFreeCur(pCur);
return NULL;
}
pCur->number = pState->number;
return pCur;
}
int32_t streamStateSessionGet(SStreamState* pState, SSessionKey* key, void** pVal, int32_t* pVLen) {
SStreamStateCur* pCur = streamStateSessionGetRanomCur(pState, key);
void* tmp = NULL;
if (streamStateSessionGetKVByCur(pCur, key, (const void**)&tmp, pVLen) == 0) {
*pVal = tdbRealloc(NULL, *pVLen);
memcpy(*pVal, tmp, *pVLen);
streamStateFreeCur(pCur);
return 0;
}
streamStateFreeCur(pCur);
return -1;
}
int32_t streamStateSessionDel(SStreamState* pState, const SSessionKey* key) {
SStateSessionKey sKey = {.key = *key, .opNum = pState->number};
return tdbTbDelete(pState->pSessionStateDb, &sKey, sizeof(SStateSessionKey), &pState->txn);
}
SStreamStateCur* streamStateSessionSeekKeyPrev(SStreamState* pState, const SSessionKey* key) {
SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
if (pCur == NULL) {
return NULL;
}
pCur->number = pState->number;
if (tdbTbcOpen(pState->pSessionStateDb, &pCur->pCur, NULL) < 0) {
streamStateFreeCur(pCur);
return NULL;
}
SStateSessionKey sKey = {.key = *key, .opNum = pState->number};
int32_t c = 0;
if (tdbTbcMoveTo(pCur->pCur, &sKey, sizeof(SStateSessionKey), &c) < 0) {
tdbTbcClose(pCur->pCur);
streamStateFreeCur(pCur);
return NULL;
}
if (c > 0) return pCur;
if (tdbTbcMoveToPrev(pCur->pCur) < 0) {
streamStateFreeCur(pCur);
return NULL;
}
return pCur;
}
SStreamStateCur* streamStateSessionSeekKeyNext(SStreamState* pState, const SSessionKey* key) {
SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
if (pCur == NULL) {
return NULL;
}
pCur->number = pState->number;
if (tdbTbcOpen(pState->pSessionStateDb, &pCur->pCur, NULL) < 0) {
streamStateFreeCur(pCur);
return NULL;
}
SStateSessionKey sKey = {.key = *key, .opNum = pState->number};
int32_t c = 0;
if (tdbTbcMoveTo(pCur->pCur, &sKey, sizeof(SStateSessionKey), &c) < 0) {
tdbTbcClose(pCur->pCur);
streamStateFreeCur(pCur);
return NULL;
}
if (c > 0) return pCur;
if (tdbTbcMoveToNext(pCur->pCur) < 0) {
streamStateFreeCur(pCur);
return NULL;
}
return pCur;
}
int32_t streamStateSessionGetKVByCur(SStreamStateCur* pCur, SSessionKey* pKey, const void** pVal, int32_t* pVLen) {
if (!pCur) {
return -1;
}
const SStateSessionKey* pKTmp = NULL;
int32_t kLen;
if (tdbTbcGet(pCur->pCur, (const void**)&pKTmp, &kLen, pVal, pVLen) < 0) {
return -1;
}
if (pKTmp->opNum != pCur->number) {
return -1;
}
if (pKey->groupId != 0 && pKey->groupId != pKTmp->key.groupId) {
return -1;
}
*pKey = pKTmp->key;
return 0;
}
int32_t streamStateSessionClear(SStreamState* pState) {
SSessionKey key = {.win.skey = 0, .win.ekey = 0, .groupId = 0};
streamStateSessionPut(pState, &key, NULL, 0);
SStreamStateCur* pCur = streamStateSessionSeekKeyNext(pState, &key);
while (1) {
SSessionKey delKey = {0};
void* buf = NULL;
int32_t size = 0;
int32_t code = streamStateSessionGetKVByCur(pCur, &delKey, buf, &size);
if (code == 0) {
memset(buf, 0, size);
streamStateSessionPut(pState, &delKey, buf, size);
} else {
break;
}
streamStateCurNext(pState, pCur);
}
streamStateFreeCur(pCur);
streamStateSessionDel(pState, &key);
return 0;
}
SStreamStateCur* streamStateSessionGetCur(SStreamState* pState, const SSessionKey* key) {
SStreamStateCur* pCur = streamStateSessionGetRanomCur(pState, key);
SSessionKey resKey = *key;
while (1) {
streamStateCurPrev(pState, pCur);
SSessionKey tmpKey = *key;
int32_t code = streamStateSessionGetKVByCur(pCur, &tmpKey, NULL, 0);
if (code == TSDB_CODE_SUCCESS && sessionKeyCmpr(key, &tmpKey) == 0) {
resKey = tmpKey;
} else {
break;
}
}
streamStateFreeCur(pCur);
return streamStateSessionGetRanomCur(pState, &resKey);
}
int32_t streamStateSessionAddIfNotExist(SStreamState* pState, SSessionKey* key, void** pVal, int32_t* pVLen) {
// todo refactor
SStreamStateCur* pCur = streamStateSessionGetCur(pState, key);
int32_t size = *pVLen;
void* tmp = NULL;
*pVal = tdbRealloc(NULL, size);
memset(*pVal, 0, size);
if (streamStateSessionGetKVByCur(pCur, key, (const void**)&tmp, pVLen) == 0) {
memcpy(*pVal, tmp, *pVLen);
streamStateFreeCur(pCur);
return 0;
}
streamStateFreeCur(pCur);
return 1;
}
int32_t streamStateStateAddIfNotExist(SStreamState* pState, SSessionKey* key, char* pKeyData, int32_t keyDataLen,
state_key_cmpr_fn fn, void** pVal, int32_t* pVLen) {
// todo refactor
int32_t res = TSDB_CODE_SUCCESS;
SSessionKey tmpKey = *key;
int32_t valSize = *pVLen;
void* tmp = tdbRealloc(NULL, valSize);
if (!tmp) {
return -1;
}
SStreamStateCur* pCur = streamStateSessionGetRanomCur(pState, key);
int32_t code = streamStateSessionGetKVByCur(pCur, key, (const void**)pVal, pVLen);
if (code == TSDB_CODE_SUCCESS) {
memcpy(tmp, *pVal, valSize);
*pVal = tmp;
streamStateFreeCur(pCur);
return res;
}
streamStateFreeCur(pCur);
streamStateSessionPut(pState, key, NULL, 0);
pCur = streamStateSessionGetRanomCur(pState, key);
streamStateCurPrev(pState, pCur);
code = streamStateSessionGetKVByCur(pCur, key, (const void**)pVal, pVLen);
if (code == TSDB_CODE_SUCCESS) {
void* stateKey = (char*)(*pVal) + (valSize - keyDataLen);
if (fn(pKeyData, stateKey) == true) {
memcpy(tmp, *pVal, valSize);
goto _end;
}
}
streamStateFreeCur(pCur);
*key = tmpKey;
pCur = streamStateSessionSeekKeyNext(pState, key);
code = streamStateSessionGetKVByCur(pCur, key, (const void**)pVal, pVLen);
if (code == TSDB_CODE_SUCCESS) {
void* stateKey = (char*)(*pVal) + (valSize - keyDataLen);
if (fn(pKeyData, stateKey) == true) {
memcpy(tmp, *pVal, valSize);
goto _end;
}
}
*key = tmpKey;
res = 1;
memset(tmp, 0, valSize);
_end:
*pVal = tmp;
streamStateSessionDel(pState, &tmpKey);
streamStateFreeCur(pCur);
return res;
}
source/libs/sync/src/syncElection.c
浏览文件 @ 39243cbb
......@@ -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) {
......
source/libs/sync/src/syncMain.c
浏览文件 @ 39243cbb
......@@ -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);
......@@ -3014,7 +3032,8 @@ static int32_t syncNodeProposeConfigChangeFinish(SSyncNode* ths, SyncReconfigFin
}
bool syncNodeIsOptimizedOneReplica(SSyncNode* ths, SRpcMsg* pMsg) {
return (ths->replicaNum == 1 && syncUtilUserCommit(pMsg->msgType) && ths->vgId != 1);
return (ths->replicaNum == 1 && syncUtilUserCommit(pMsg->msgType));
// return (ths->replicaNum == 1 && syncUtilUserCommit(pMsg->msgType) && ths->vgId != 1);
}
int32_t syncNodeCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex, uint64_t flag) {
......@@ -3066,7 +3085,8 @@ int32_t syncNodeCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex,
// user commit
if ((ths->pFsm->FpCommitCb != NULL) && syncUtilUserCommit(pEntry->originalRpcType)) {
bool internalExecute = true;
if ((ths->replicaNum == 1) && ths->restoreFinish && ths->vgId != 1) {
if ((ths->replicaNum == 1) && ths->restoreFinish) {
// if ((ths->replicaNum == 1) && ths->restoreFinish && ths->vgId != 1) {
internalExecute = false;
}
......
source/libs/tdb/src/db/tdbBtree.c
浏览文件 @ 39243cbb
......@@ -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;
}
}
......
source/libs/tdb/src/db/tdbDb.c
浏览文件 @ 39243cbb
此差异已折叠。
source/libs/tdb/src/db/tdbPCache.c
浏览文件 @ 39243cbb
此差异已折叠。
source/libs/tdb/src/db/tdbPager.c
浏览文件 @ 39243cbb
此差异已折叠。
source/libs/wal/src/walMeta.c
浏览文件 @ 39243cbb
此差异已折叠。
source/libs/wal/src/walWrite.c
浏览文件 @ 39243cbb
此差异已折叠。
source/util/src/tlog.c
浏览文件 @ 39243cbb
此差异已折叠。
tests/parallel_test/run_case.sh
浏览文件 @ 39243cbb
此差异已折叠。
tests/script/jenkins/basic.txt
浏览文件 @ 39243cbb
此差异已折叠。
tests/script/tsim/parser/like.sim
浏览文件 @ 39243cbb
此差异已折叠。
tests/script/tsim/stream/state0.sim
浏览文件 @ 39243cbb
此差异已折叠。
tools/shell/src/shellCommand.c
浏览文件 @ 39243cbb
此差异已折叠。
utils/test/c/sml_test.c
浏览文件 @ 39243cbb
此差异已折叠。
utils/test/c/tmqDemo.c
浏览文件 @ 39243cbb
此差异已折叠。
utils/test/c/tmqSim.c
浏览文件 @ 39243cbb
此差异已折叠。
utils/tsim/src/simExe.c
浏览文件 @ 39243cbb
此差异已折叠。
utils/tsim/src/simParse.c
浏览文件 @ 39243cbb
此差异已折叠。
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或