Skip to content

T
TDengine

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

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

Merge branch '3.0' into feature/TD-14481-3.0

上级 de891966 9dcf7854
展开全部 隐藏空白更改
内联 并排
Showing with 1446 addition and 418 deletion
docs/en/12-taos-sql/12-keywords.md → docs/en/12-taos-sql/20-keywords.md
浏览文件 @ cdb8cae0
...@@ -4,50 +4,275 @@ title: Keywords ...@@ -4,50 +4,275 @@ title: Keywords
There are about 200 keywords reserved by TDengine, they can't be used as the name of database, STable or table with either upper case, lower case or mixed case. There are about 200 keywords reserved by TDengine, they can't be used as the name of database, STable or table with either upper case, lower case or mixed case.
**Keywords List** ## Keywords List
| | | | | | ### A
| ----------- | ---------- | --------- | ---------- | ------------ |
| ABORT | CREATE | IGNORE | NULL | STAR | - ABORT
| ACCOUNT | CTIME | IMMEDIATE | OF | STATE | - ACCOUNT
| ACCOUNTS | DATABASE | IMPORT | OFFSET | STATEMENT | - ACCOUNTS
| ADD | DATABASES | IN | OR | STATE_WINDOW | - ADD
| AFTER | DAYS | INITIALLY | ORDER | STORAGE | - AFTER
| ALL | DBS | INSERT | PARTITIONS | STREAM | - ALL
| ALTER | DEFERRED | INSTEAD | PASS | STREAMS | - ALTER
| AND | DELIMITERS | INT | PLUS | STRING | - AND
| AS | DESC | INTEGER | PPS | SYNCDB | - AS
| ASC | DESCRIBE | INTERVAL | PRECISION | TABLE | - ASC
| ATTACH | DETACH | INTO | PREV | TABLES | - ATTACH
| BEFORE | DISTINCT | IS | PRIVILEGE | TAG |
| BEGIN | DIVIDE | ISNULL | QTIME | TAGS | ### B
| BETWEEN | DNODE | JOIN | QUERIES | TBNAME |
| BIGINT | DNODES | KEEP | QUERY | TIMES | - BEFORE
| BINARY | DOT | KEY | QUORUM | TIMESTAMP | - BEGIN
| BITAND | DOUBLE | KILL | RAISE | TINYINT | - BETWEEN
| BITNOT | DROP | LE | REM | TOPIC | - BIGINT
| BITOR | EACH | LIKE | REPLACE | TOPICS | - BINARY
| BLOCKS | END | LIMIT | REPLICA | TRIGGER | - BITAND
| BOOL | EQ | LINEAR | RESET | TSERIES | - BITNOT
| BY | EXISTS | LOCAL | RESTRICT | UMINUS | - BITOR
| CACHE | EXPLAIN | LP | ROW | UNION | - BLOCKS
| CACHELAST | FAIL | LSHIFT | RP | UNSIGNED | - BOOL
| CASCADE | FILE | LT | RSHIFT | UPDATE | - BY
| CHANGE | FILL | MATCH | SCORES | UPLUS |
| CLUSTER | FLOAT | MAXROWS | SELECT | USE | ### C
| COLON | FOR | MINROWS | SEMI | USER |
| COLUMN | FROM | MINUS | SESSION | USERS | - CACHE
| COMMA | FSYNC | MNODES | SET | USING | - CACHELAST
| COMP | GE | MODIFY | SHOW | VALUES | - CASCADE
| COMPACT | GLOB | MODULES | SLASH | VARIABLE | - CHANGE
| CONCAT | GRANTS | NCHAR | SLIDING | VARIABLES | - CLUSTER
| CONFLICT | GROUP | NE | SLIMIT | VGROUPS | - COLON
| CONNECTION | GT | NONE | SMALLINT | VIEW | - COLUMN
| CONNECTIONS | HAVING | NOT | SOFFSET | VNODES | - COMMA
| CONNS | ID | NOTNULL | STable | WAL | - COMP
| COPY | IF | NOW | STableS | WHERE | - COMPACT
| _C0 | _QSTART | _QSTOP | _QDURATION | _WSTART | - CONCAT
| _WSTOP | _WDURATION | _ROWTS | - CONFLICT
- CONNECTION
- CONNECTIONS
- CONNS
- COPY
- CREATE
- CTIME
### D
- DATABASE
- DATABASES
- DAYS
- DBS
- DEFERRED
- DELETE
- DELIMITERS
- DESC
- DESCRIBE
- DETACH
- DISTINCT
- DIVIDE
- DNODE
- DNODES
- DOT
- DOUBLE
- DROP
### E
- END
- EQ
- EXISTS
- EXPLAIN
### F
- FAIL
- FILE
- FILL
- FLOAT
- FOR
- FROM
- FSYNC
### G
- GE
- GLOB
- GRANTS
- GROUP
- GT
### H
- HAVING
### I
- ID
- IF
- IGNORE
- IMMEDIA
- IMPORT
- IN
- INITIAL
- INSERT
- INSTEAD
- INT
- INTEGER
- INTERVA
- INTO
- IS
- ISNULL
### J
- JOIN
### K
- KEEP
- KEY
- KILL
### L
- LE
- LIKE
- LIMIT
- LINEAR
- LOCAL
- LP
- LSHIFT
- LT
### M
- MATCH
- MAXROWS
- MINROWS
- MINUS
- MNODES
- MODIFY
- MODULES
### N
- NE
- NONE
- NOT
- NOTNULL
- NOW
- NULL
### O
- OF
- OFFSET
- OR
- ORDER
### P
- PARTITION
- PASS
- PLUS
- PPS
- PRECISION
- PREV
- PRIVILEGE
### Q
- QTIME
- QUERIE
- QUERY
- QUORUM
### R
- RAISE
- REM
- REPLACE
- REPLICA
- RESET
- RESTRIC
- ROW
- RP
- RSHIFT
### S
- SCORES
- SELECT
- SEMI
- SESSION
- SET
- SHOW
- SLASH
- SLIDING
- SLIMIT
- SMALLIN
- SOFFSET
- STable
- STableS
- STAR
- STATE
- STATEMEN
- STATE_WI
- STORAGE
- STREAM
- STREAMS
- STRING
- SYNCDB
### T
- TABLE
- TABLES
- TAG
- TAGS
- TBNAME
- TIMES
- TIMESTAMP
- TINYINT
- TOPIC
- TOPICS
- TRIGGER
- TSERIES
### U
- UMINUS
- UNION
- UNSIGNED
- UPDATE
- UPLUS
- USE
- USER
- USERS
- USING
### V
- VALUES
- VARIABLE
- VARIABLES
- VGROUPS
- VIEW
- VNODES
### W
- WAL
- WHERE
### _
- _C0
- _QSTART
- _QSTOP
- _QDURATION
- _WSTART
- _WSTOP
- _WDURATION
## Explanations ## Explanations
### TBNAME ### TBNAME
......
docs/zh/12-taos-sql/12-keywords.md → docs/zh/12-taos-sql/20-keywords.md
浏览文件 @ cdb8cae0
...@@ -45,48 +45,274 @@ title: TDengine 参数限制与保留关键字 ...@@ -45,48 +45,274 @@ title: TDengine 参数限制与保留关键字
目前 TDengine 有将近 200 个内部保留关键字，这些关键字无论大小写均不可以用作库名、表名、STable 名、数据列名及标签列名等。这些关键字列表如下： 目前 TDengine 有将近 200 个内部保留关键字，这些关键字无论大小写均不可以用作库名、表名、STable 名、数据列名及标签列名等。这些关键字列表如下：
| 关键字列表 | | | | | ### A
| ----------- | ---------- | --------- | ---------- | ------------ |
| ABORT | CREATE | IGNORE | NULL | STAR | - ABORT
| ACCOUNT | CTIME | IMMEDIATE | OF | STATE | - ACCOUNT
| ACCOUNTS | DATABASE | IMPORT | OFFSET | STATEMENT | - ACCOUNTS
| ADD | DATABASES | IN | OR | STATE_WINDOW | - ADD
| AFTER | DAYS | INITIALLY | ORDER | STORAGE | - AFTER
| ALL | DBS | INSERT | PARTITIONS | STREAM | - ALL
| ALTER | DEFERRED | INSTEAD | PASS | STREAMS | - ALTER
| AND | DELIMITERS | INT | PLUS | STRING | - AND
| AS | DESC | INTEGER | PPS | SYNCDB | - AS
| ASC | DESCRIBE | INTERVAL | PRECISION | TABLE | - ASC
| ATTACH | DETACH | INTO | PREV | TABLES | - ATTACH
| BEFORE | DISTINCT | IS | PRIVILEGE | TAG |
| BEGIN | DIVIDE | ISNULL | QTIME | TAGS | ### B
| BETWEEN | DNODE | JOIN | QUERIES | TBNAME |
| BIGINT | DNODES | KEEP | QUERY | TIMES | - BEFORE
| BINARY | DOT | KEY | QUORUM | TIMESTAMP | - BEGIN
| BITAND | DOUBLE | KILL | RAISE | TINYINT | - BETWEEN
| BITNOT | DROP | LE | REM | TOPIC | - BIGINT
| BITOR | EACH | LIKE | REPLACE | TOPICS | - BINARY
| BLOCKS | END | LIMIT | REPLICA | TRIGGER | - BITAND
| BOOL | EQ | LINEAR | RESET | TSERIES | - BITNOT
| BY | EXISTS | LOCAL | RESTRICT | UMINUS | - BITOR
| CACHE | EXPLAIN | LP | ROW | UNION | - BLOCKS
| CACHELAST | FAIL | LSHIFT | RP | UNSIGNED | - BOOL
| CASCADE | FILE | LT | RSHIFT | UPDATE | - BY
| CHANGE | FILL | MATCH | SCORES | UPLUS |
| CLUSTER | FLOAT | MAXROWS | SELECT | USE | ### C
| COLON | FOR | MINROWS | SEMI | USER |
| COLUMN | FROM | MINUS | SESSION | USERS | - CACHE
| COMMA | FSYNC | MNODES | SET | USING | - CACHELAST
| COMP | GE | MODIFY | SHOW | VALUES | - CASCADE
| COMPACT | GLOB | MODULES | SLASH | VARIABLE | - CHANGE
| CONCAT | GRANTS | NCHAR | SLIDING | VARIABLES | - CLUSTER
| CONFLICT | GROUP | NE | SLIMIT | VGROUPS | - COLON
| CONNECTION | GT | NONE | SMALLINT | VIEW | - COLUMN
| CONNECTIONS | HAVING | NOT | SOFFSET | VNODES | - COMMA
| CONNS | ID | NOTNULL | STABLE | WAL | - COMP
| COPY | IF | NOW | STABLES | WHERE | - COMPACT
| _C0 | _QSTART | _QSTOP | _QDURATION | _WSTART | - CONCAT
| _WSTOP | _WDURATION | _ROWTS | - CONFLICT
- CONNECTION
- CONNECTIONS
- CONNS
- COPY
- CREATE
- CTIME
### D
- DATABASE
- DATABASES
- DAYS
- DBS
- DEFERRED
- DELETE
- DELIMITERS
- DESC
- DESCRIBE
- DETACH
- DISTINCT
- DIVIDE
- DNODE
- DNODES
- DOT
- DOUBLE
- DROP
### E
- END
- EQ
- EXISTS
- EXPLAIN
### F
- FAIL
- FILE
- FILL
- FLOAT
- FOR
- FROM
- FSYNC
### G
- GE
- GLOB
- GRANTS
- GROUP
- GT
### H
- HAVING
### I
- ID
- IF
- IGNORE
- IMMEDIA
- IMPORT
- IN
- INITIAL
- INSERT
- INSTEAD
- INT
- INTEGER
- INTERVA
- INTO
- IS
- ISNULL
### J
- JOIN
### K
- KEEP
- KEY
- KILL
### L
- LE
- LIKE
- LIMIT
- LINEAR
- LOCAL
- LP
- LSHIFT
- LT
### M
- MATCH
- MAXROWS
- MINROWS
- MINUS
- MNODES
- MODIFY
- MODULES
### N
- NE
- NONE
- NOT
- NOTNULL
- NOW
- NULL
### O
- OF
- OFFSET
- OR
- ORDER
### P
- PARTITION
- PASS
- PLUS
- PPS
- PRECISION
- PREV
- PRIVILEGE
### Q
- QTIME
- QUERIE
- QUERY
- QUORUM
### R
- RAISE
- REM
- REPLACE
- REPLICA
- RESET
- RESTRIC
- ROW
- RP
- RSHIFT
### S
- SCORES
- SELECT
- SEMI
- SESSION
- SET
- SHOW
- SLASH
- SLIDING
- SLIMIT
- SMALLIN
- SOFFSET
- STable
- STableS
- STAR
- STATE
- STATEMEN
- STATE_WI
- STORAGE
- STREAM
- STREAMS
- STRING
- SYNCDB
### T
- TABLE
- TABLES
- TAG
- TAGS
- TBNAME
- TIMES
- TIMESTAMP
- TINYINT
- TOPIC
- TOPICS
- TRIGGER
- TSERIES
### U
- UMINUS
- UNION
- UNSIGNED
- UPDATE
- UPLUS
- USE
- USER
- USERS
- USING
### V
- VALUES
- VARIABLE
- VARIABLES
- VGROUPS
- VIEW
- VNODES
### W
- WAL
- WHERE
### _
- _C0
- _QSTART
- _QSTOP
- _QDURATION
- _WSTART
- _WSTOP
- _WDURATION
## 特殊说明 ## 特殊说明
### TBNAME ### TBNAME
......
include/dnode/mnode/mnode.h
浏览文件 @ cdb8cae0
...@@ -32,9 +32,7 @@ typedef struct { ...@@ -32,9 +32,7 @@ typedef struct {
int32_t dnodeId; int32_t dnodeId;
bool standby; bool standby;
bool deploy; bool deploy;
int8_t replica; SReplica replica;
int8_t selfIndex;
SReplica replicas[TSDB_MAX_REPLICA];
SMsgCb msgCb; SMsgCb msgCb;
} SMnodeOpt; } SMnodeOpt;
...@@ -83,6 +81,7 @@ int32_t mndGetLoad(SMnode *pMnode, SMnodeLoad *pLoad); ...@@ -83,6 +81,7 @@ int32_t mndGetLoad(SMnode *pMnode, SMnodeLoad *pLoad);
*/ */
int32_t mndProcessRpcMsg(SRpcMsg *pMsg); int32_t mndProcessRpcMsg(SRpcMsg *pMsg);
int32_t mndProcessSyncMsg(SRpcMsg *pMsg); int32_t mndProcessSyncMsg(SRpcMsg *pMsg);
int32_t mndPreprocessQueryMsg(SMnode * pMnode, SRpcMsg * pMsg);
/** /**
* @brief Generate machine code * @brief Generate machine code
......
include/libs/function/functionMgt.h
浏览文件 @ cdb8cae0
...@@ -138,6 +138,10 @@ typedef enum EFunctionType { ...@@ -138,6 +138,10 @@ typedef enum EFunctionType {
FUNCTION_TYPE_TOP_MERGE, FUNCTION_TYPE_TOP_MERGE,
FUNCTION_TYPE_BOTTOM_PARTIAL, FUNCTION_TYPE_BOTTOM_PARTIAL,
FUNCTION_TYPE_BOTTOM_MERGE, FUNCTION_TYPE_BOTTOM_MERGE,
FUNCTION_TYPE_FIRST_PARTIAL,
FUNCTION_TYPE_FIRST_MERGE,
FUNCTION_TYPE_LAST_PARTIAL,
FUNCTION_TYPE_LAST_MERGE,
// user defined funcion // user defined funcion
FUNCTION_TYPE_UDF = 10000 FUNCTION_TYPE_UDF = 10000
......
source/client/src/clientSml.c
浏览文件 @ cdb8cae0
...@@ -1327,8 +1327,8 @@ static int32_t smlKvTimeArrayCompare(const void *key1, const void *key2) { ...@@ -1327,8 +1327,8 @@ static int32_t smlKvTimeArrayCompare(const void *key1, const void *key2) {
static int32_t smlKvTimeHashCompare(const void *key1, const void *key2) { static int32_t smlKvTimeHashCompare(const void *key1, const void *key2) {
SHashObj *s1 = *(SHashObj **)key1; SHashObj *s1 = *(SHashObj **)key1;
SHashObj *s2 = *(SHashObj **)key2; SHashObj *s2 = *(SHashObj **)key2;
SSmlKv *kv1 = (SSmlKv *)taosHashGet(s1, TS, TS_LEN); SSmlKv *kv1 = *(SSmlKv **)taosHashGet(s1, TS, TS_LEN);
SSmlKv *kv2 = (SSmlKv *)taosHashGet(s2, TS, TS_LEN); SSmlKv *kv2 = *(SSmlKv **)taosHashGet(s2, TS, TS_LEN);
ASSERT(kv1->type == TSDB_DATA_TYPE_TIMESTAMP); ASSERT(kv1->type == TSDB_DATA_TYPE_TIMESTAMP);
ASSERT(kv2->type == TSDB_DATA_TYPE_TIMESTAMP); ASSERT(kv2->type == TSDB_DATA_TYPE_TIMESTAMP);
if (kv1->i < kv2->i) { if (kv1->i < kv2->i) {
...@@ -1340,29 +1340,13 @@ static int32_t smlKvTimeHashCompare(const void *key1, const void *key2) { ...@@ -1340,29 +1340,13 @@ static int32_t smlKvTimeHashCompare(const void *key1, const void *key2) {
} }
} }
static int32_t smlDealCols(SSmlTableInfo *oneTable, bool dataFormat, SArray *cols) { static int32_t smlDealCols(SSmlTableInfo* oneTable, bool dataFormat, SArray *cols){
if (dataFormat) { if(dataFormat){
void *p = taosArraySearch(oneTable->cols, &cols, smlKvTimeArrayCompare, TD_GE); void *p = taosArraySearch(oneTable->cols, &cols, smlKvTimeArrayCompare, TD_GT);
if (p == NULL) { if(p == NULL){
taosArrayPush(oneTable->cols, &cols); taosArrayPush(oneTable->cols, &cols);
} else { // to make the sort stable for update data }else{
SArray *sa = (SArray *)p; taosArrayInsert(oneTable->cols, TARRAY_ELEM_IDX(oneTable->cols, p), &cols);
SSmlKv *cur = (SSmlKv *)taosArrayGet(sa, 0);
SSmlKv *dCur = (SSmlKv *)taosArrayGet(cols, 0);
if (cur->i > dCur->i) {
taosArrayInsert(oneTable->cols, TARRAY_ELEM_IDX(oneTable->cols, p), &cols);
} else {
ASSERT(cur->i == dCur->i);
int32_t index = TARRAY_ELEM_IDX(oneTable->cols, p) + 1;
for (; index < taosArrayGetSize(oneTable->cols); index++) {
SArray *tmp = (SArray *)taosArrayGet(oneTable->cols, index);
SSmlKv *curTs = (SSmlKv *)taosArrayGet(tmp, 0);
if (curTs->i > dCur->i) {
break;
}
}
taosArrayInsert(oneTable->cols, index, &cols);
}
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
...@@ -1377,27 +1361,11 @@ static int32_t smlDealCols(SSmlTableInfo *oneTable, bool dataFormat, SArray *col ...@@ -1377,27 +1361,11 @@ static int32_t smlDealCols(SSmlTableInfo *oneTable, bool dataFormat, SArray *col
taosHashPut(kvHash, kv->key, kv->keyLen, &kv, POINTER_BYTES); taosHashPut(kvHash, kv->key, kv->keyLen, &kv, POINTER_BYTES);
} }
void *p = taosArraySearch(oneTable->cols, &kvHash, smlKvTimeHashCompare, TD_GE); void *p = taosArraySearch(oneTable->cols, &kvHash, smlKvTimeHashCompare, TD_GT);
if (p == NULL) { if(p == NULL){
taosArrayPush(oneTable->cols, &kvHash); taosArrayPush(oneTable->cols, &kvHash);
} else { // to make the sort stable for update data }else{
SHashObj *sa = (SHashObj *)p; taosArrayInsert(oneTable->cols, TARRAY_ELEM_IDX(oneTable->cols, p), &kvHash);
SSmlKv *cur = (SSmlKv *)taosHashGet(sa, TS, TS_LEN);
SSmlKv *dCur = (SSmlKv *)taosArrayGet(cols, 0);
if (cur->i > dCur->i) {
taosArrayInsert(oneTable->cols, TARRAY_ELEM_IDX(oneTable->cols, p), &cols);
} else {
ASSERT(cur->i == dCur->i);
int32_t index = TARRAY_ELEM_IDX(oneTable->cols, p) + 1;
for (; index < taosArrayGetSize(oneTable->cols); index++) {
SHashObj *tmp = (SHashObj *)taosArrayGet(oneTable->cols, index);
SSmlKv *curTs = (SSmlKv *)taosHashGet(tmp, TS, TS_LEN);
if (curTs->i > dCur->i) {
break;
}
}
taosArrayInsert(oneTable->cols, index, &cols);
}
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
......
source/client/test/smlTest.cpp
浏览文件 @ cdb8cae0
...@@ -1260,4 +1260,28 @@ TEST(testCase, sml_16368_Test) { ...@@ -1260,4 +1260,28 @@ TEST(testCase, sml_16368_Test) {
pRes = taos_schemaless_insert(taos, (char**)sql, sizeof(sql)/sizeof(sql[0]), TSDB_SML_JSON_PROTOCOL, TSDB_SML_TIMESTAMP_MICRO_SECONDS); pRes = taos_schemaless_insert(taos, (char**)sql, sizeof(sql)/sizeof(sql[0]), TSDB_SML_JSON_PROTOCOL, TSDB_SML_TIMESTAMP_MICRO_SECONDS);
ASSERT_EQ(taos_errno(pRes), 0); ASSERT_EQ(taos_errno(pRes), 0);
taos_free_result(pRes); taos_free_result(pRes);
}*/ }
TEST(testCase, sml_dup_time_Test) {
TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);
ASSERT_NE(taos, nullptr);
TAOS_RES* pRes = taos_query(taos, "create database if not exists dup_time schemaless 1");
taos_free_result(pRes);
const char *sql[] = {
//"test_ms,t0=t c0=f 1626006833641",
"ubzlsr,id=qmtcvgd,t0=t,t1=127i8,t2=32767i16,t3=2147483647i32,t4=9223372036854775807i64,t5=11.12345f32,t6=22.123456789f64,t7=\"binaryTagValue\",t8=L\"ncharTagValue\" c0=false,c1=1i8,c2=32767i16,c3=2147483647i32,c4=9223372036854775807i64,c5=11.12345f32,c6=22.123456789f64,c7=\"xcxvwjvf\",c8=L\"ncharColValue\",c9=7u64 1626006833639000000",
"ubzlsr,id=qmtcvgd,t0=t,t1=127i8,t2=32767i16,t3=2147483647i32,t4=9223372036854775807i64,t5=11.12345f32,t6=22.123456789f64,t7=\"binaryTagValue\",t8=L\"ncharTagValue\" c0=T,c1=2i8,c2=32767i16,c3=2147483647i32,c4=9223372036854775807i64,c5=11.12345f32,c6=22.123456789f64,c7=\"fixrzcuq\",c8=L\"ncharColValue\",c9=7u64 1626006833639000000",
"ubzlsr,id=qmtcvgd,t0=t,t1=127i8,t2=32767i16,t3=2147483647i32,t4=9223372036854775807i64,t5=11.12345f32,t6=22.123456789f64,t7=\"binaryTagValue\",t8=L\"ncharTagValue\" c0=t,c1=3i8,c2=32767i16,c3=2147483647i32,c4=9223372036854775807i64,c5=11.12345f32,c6=22.123456789f64,c7=\"iupzdqub\",c8=L\"ncharColValue\",c9=7u64 1626006833639000000",
"ubzlsr,id=qmtcvgd,t0=t,t1=127i8,t2=32767i16,t3=2147483647i32,t4=9223372036854775807i64,t5=11.12345f32,t6=22.123456789f64,t7=\"binaryTagValue\",t8=L\"ncharTagValue\" c0=t,c1=4i8,c2=32767i16,c3=2147483647i32,c4=9223372036854775807i64,c5=11.12345f32,c6=22.123456789f64,c7=\"yvvtzzof\",c8=L\"ncharColValue\",c9=7u64 1626006833639000000",
"ubzlsr,id=qmtcvgd,t0=t,t1=127i8,t2=32767i16,t3=2147483647i32,t4=9223372036854775807i64,t5=11.12345f32,t6=22.123456789f64,t7=\"binaryTagValue\",t8=L\"ncharTagValue\" c0=t,c1=5i8,c2=32767i16,c3=2147483647i32,c4=9223372036854775807i64,c5=11.12345f32,c6=22.123456789f64,c7=\"vbxpilkj\",c8=L\"ncharColValue\",c9=7u64 1626006833639000000"
};
pRes = taos_query(taos, "use dup_time");
taos_free_result(pRes);
pRes = taos_schemaless_insert(taos, (char**)sql, sizeof(sql)/sizeof(sql[0]), TSDB_SML_LINE_PROTOCOL, 0);
ASSERT_EQ(taos_errno(pRes), 0);
taos_free_result(pRes);
}
*/
source/dnode/mgmt/mgmt_mnode/inc/mmInt.h
浏览文件 @ cdb8cae0
...@@ -34,39 +34,31 @@ typedef struct SMnodeMgmt { ...@@ -34,39 +34,31 @@ typedef struct SMnodeMgmt {
SSingleWorker writeWorker; SSingleWorker writeWorker;
SSingleWorker syncWorker; SSingleWorker syncWorker;
SSingleWorker monitorWorker; SSingleWorker monitorWorker;
SReplica replicas[TSDB_MAX_REPLICA];
int8_t replica;
bool stopped; bool stopped;
int32_t refCount; int32_t refCount;
TdThreadRwlock lock; TdThreadRwlock lock;
} SMnodeMgmt; } SMnodeMgmt;
// mmFile.c // mmFile.c
int32_t mmReadFile(SMnodeMgmt *pMgmt, bool *pDeployed); int32_t mmReadFile(SMnodeMgmt *pMgmt, SReplica *pReplica, bool *pDeployed);
int32_t mmWriteFile(SMnodeMgmt *pMgmt, SDCreateMnodeReq *pMsg, bool deployed); int32_t mmWriteFile(SMnodeMgmt *pMgmt, const SReplica *pReplica, bool deployed);
// mmInt.c
int32_t mmAcquire(SMnodeMgmt *pMgmt);
void mmRelease(SMnodeMgmt *pMgmt);
// mmHandle.c // mmHandle.c
SArray *mmGetMsgHandles(); SArray *mmGetMsgHandles();
int32_t mmProcessCreateReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg); int32_t mmProcessCreateReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg);
int32_t mmProcessDropReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg); int32_t mmProcessDropReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg);
int32_t mmProcessAlterReq(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t mmProcessGetMonitorInfoReq(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t mmProcessGetMonitorInfoReq(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t mmProcessGetLoadsReq(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t mmProcessGetLoadsReq(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t mndPreprocessQueryMsg(SMnode * pMnode, SRpcMsg * pMsg);
// mmWorker.c // mmWorker.c
int32_t mmStartWorker(SMnodeMgmt *pMgmt); int32_t mmStartWorker(SMnodeMgmt *pMgmt);
void mmStopWorker(SMnodeMgmt *pMgmt); void mmStopWorker(SMnodeMgmt *pMgmt);
int32_t mmPutNodeMsgToWriteQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t mmPutMsgToWriteQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t mmPutNodeMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t mmPutMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t mmPutNodeMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t mmPutMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t mmPutNodeMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t mmPutMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t mmPutNodeMsgToMonitorQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg); int32_t mmPutMsgToMonitorQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t mmPutRpcMsgToQueue(SMnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc); int32_t mmPutMsgToQueue(SMnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc);
#ifdef __cplusplus #ifdef __cplusplus
} }
......
source/dnode/mgmt/mgmt_mnode/src/mmFile.c
浏览文件 @ cdb8cae0
...@@ -16,7 +16,7 @@ ...@@ -16,7 +16,7 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "mmInt.h" #include "mmInt.h"
int32_t mmReadFile(SMnodeMgmt *pMgmt, bool *pDeployed) { int32_t mmReadFile(SMnodeMgmt *pMgmt, SReplica *pReplica, bool *pDeployed) {
int32_t code = TSDB_CODE_INVALID_JSON_FORMAT; int32_t code = TSDB_CODE_INVALID_JSON_FORMAT;
int32_t len = 0; int32_t len = 0;
int32_t maxLen = 4096; int32_t maxLen = 4096;
...@@ -52,61 +52,54 @@ int32_t mmReadFile(SMnodeMgmt *pMgmt, bool *pDeployed) { ...@@ -52,61 +52,54 @@ int32_t mmReadFile(SMnodeMgmt *pMgmt, bool *pDeployed) {
} }
*pDeployed = deployed->valueint; *pDeployed = deployed->valueint;
cJSON *mnodes = cJSON_GetObjectItem(root, "mnodes"); cJSON *id = cJSON_GetObjectItem(root, "id");
if (mnodes != NULL) { if (id) {
if (!mnodes || mnodes->type != cJSON_Array) { if (id->type != cJSON_Number) {
dError("failed to read %s since nodes not found", file); dError("failed to read %s since id not found", file);
goto _OVER; goto _OVER;
} }
if (pReplica) {
pReplica->id = id->valueint;
}
}
pMgmt->replica = cJSON_GetArraySize(mnodes); cJSON *fqdn = cJSON_GetObjectItem(root, "fqdn");
if (pMgmt->replica <= 0 || pMgmt->replica > TSDB_MAX_REPLICA) { if (fqdn) {
dError("failed to read %s since mnodes size %d invalid", file, pMgmt->replica); if (fqdn->type != cJSON_String || fqdn->valuestring == NULL) {
dError("failed to read %s since fqdn not found", file);
goto _OVER; goto _OVER;
} }
if (pReplica) {
for (int32_t i = 0; i < pMgmt->replica; ++i) {
cJSON *node = cJSON_GetArrayItem(mnodes, i);
if (node == NULL) break;
SReplica *pReplica = &pMgmt->replicas[i];
cJSON *id = cJSON_GetObjectItem(node, "id");
if (!id || id->type != cJSON_Number) {
dError("failed to read %s since id not found", file);
goto _OVER;
}
pReplica->id = id->valueint;
cJSON *fqdn = cJSON_GetObjectItem(node, "fqdn");
if (!fqdn || fqdn->type != cJSON_String || fqdn->valuestring == NULL) {
dError("failed to read %s since fqdn not found", file);
goto _OVER;
}
tstrncpy(pReplica->fqdn, fqdn->valuestring, TSDB_FQDN_LEN); tstrncpy(pReplica->fqdn, fqdn->valuestring, TSDB_FQDN_LEN);
}
}
cJSON *port = cJSON_GetObjectItem(node, "port"); cJSON *port = cJSON_GetObjectItem(root, "port");
if (!port || port->type != cJSON_Number) { if (port) {
dError("failed to read %s since port not found", file); if (port->type != cJSON_Number) {
goto _OVER; dError("failed to read %s since port not found", file);
} goto _OVER;
pReplica->port = port->valueint; }
if (pReplica) {
pReplica->port = (uint16_t)port->valueint;
} }
} }
code = 0; code = 0;
dDebug("succcessed to read file %s, deployed:%d", file, *pDeployed);
_OVER: _OVER:
if (content != NULL) taosMemoryFree(content); if (content != NULL) taosMemoryFree(content);
if (root != NULL) cJSON_Delete(root); if (root != NULL) cJSON_Delete(root);
if (pFile != NULL) taosCloseFile(&pFile); if (pFile != NULL) taosCloseFile(&pFile);
if (code == 0) {
dDebug("succcessed to read file %s, deployed:%d", file, *pDeployed);
}
terrno = code; terrno = code;
return code; return code;
} }
int32_t mmWriteFile(SMnodeMgmt *pMgmt, SDCreateMnodeReq *pMsg, bool deployed) { int32_t mmWriteFile(SMnodeMgmt *pMgmt, const SReplica *pReplica, bool deployed) {
char file[PATH_MAX] = {0}; char file[PATH_MAX] = {0};
char realfile[PATH_MAX] = {0}; char realfile[PATH_MAX] = {0};
snprintf(file, sizeof(file), "%s%smnode.json.bak", pMgmt->path, TD_DIRSEP); snprintf(file, sizeof(file), "%s%smnode.json.bak", pMgmt->path, TD_DIRSEP);
...@@ -124,26 +117,11 @@ int32_t mmWriteFile(SMnodeMgmt *pMgmt, SDCreateMnodeReq *pMsg, bool deployed) { ...@@ -124,26 +117,11 @@ int32_t mmWriteFile(SMnodeMgmt *pMgmt, SDCreateMnodeReq *pMsg, bool deployed) {
char *content = taosMemoryCalloc(1, maxLen + 1); char *content = taosMemoryCalloc(1, maxLen + 1);
len += snprintf(content + len, maxLen - len, "{\n"); len += snprintf(content + len, maxLen - len, "{\n");
if (pReplica != NULL && pReplica->id > 0) {
int8_t replica = (pMsg != NULL ? pMsg->replica : pMgmt->replica); len += snprintf(content + len, maxLen - len, " \"id\": %d,\n", pReplica->id);
if (replica > 0) { len += snprintf(content + len, maxLen - len, " \"fqdn\": \"%s\",\n", pReplica->fqdn);
len += snprintf(content + len, maxLen - len, " \"mnodes\": [{\n"); len += snprintf(content + len, maxLen - len, " \"port\": %u\n,", pReplica->port);
for (int32_t i = 0; i < replica; ++i) {
SReplica *pReplica = &pMgmt->replicas[i];
if (pMsg != NULL) {
pReplica = &pMsg->replicas[i];
}
len += snprintf(content + len, maxLen - len, " \"id\": %d,\n", pReplica->id);
len += snprintf(content + len, maxLen - len, " \"fqdn\": \"%s\",\n", pReplica->fqdn);
len += snprintf(content + len, maxLen - len, " \"port\": %u\n", pReplica->port);
if (i < replica - 1) {
len += snprintf(content + len, maxLen - len, " },{\n");
} else {
len += snprintf(content + len, maxLen - len, " }],\n");
}
}
} }
len += snprintf(content + len, maxLen - len, " \"deployed\": %d\n", deployed); len += snprintf(content + len, maxLen - len, " \"deployed\": %d\n", deployed);
len += snprintf(content + len, maxLen - len, "}\n"); len += snprintf(content + len, maxLen - len, "}\n");
......
source/dnode/mgmt/mgmt_mnode/src/mmInt.c
浏览文件 @ cdb8cae0
...@@ -27,7 +27,7 @@ static bool mmDeployRequired(const SMgmtInputOpt *pInput) { ...@@ -27,7 +27,7 @@ static bool mmDeployRequired(const SMgmtInputOpt *pInput) {
static int32_t mmRequire(const SMgmtInputOpt *pInput, bool *required) { static int32_t mmRequire(const SMgmtInputOpt *pInput, bool *required) {
SMnodeMgmt mgmt = {0}; SMnodeMgmt mgmt = {0};
mgmt.path = pInput->path; mgmt.path = pInput->path;
if (mmReadFile(&mgmt, required) != 0) { if (mmReadFile(&mgmt, NULL, required) != 0) {
return -1; return -1;
} }
...@@ -43,33 +43,19 @@ static void mmBuildOptionForDeploy(SMnodeMgmt *pMgmt, const SMgmtInputOpt *pInpu ...@@ -43,33 +43,19 @@ static void mmBuildOptionForDeploy(SMnodeMgmt *pMgmt, const SMgmtInputOpt *pInpu
pOption->deploy = true; pOption->deploy = true;
pOption->msgCb = pMgmt->msgCb; pOption->msgCb = pMgmt->msgCb;
pOption->dnodeId = pMgmt->pData->dnodeId; pOption->dnodeId = pMgmt->pData->dnodeId;
pOption->replica.id = 1;
pOption->replica = 1; pOption->replica.port = tsServerPort;
pOption->selfIndex = 0; tstrncpy(pOption->replica.fqdn, tsLocalFqdn, TSDB_FQDN_LEN);
SReplica *pReplica = &pOption->replicas[0];
pReplica->id = 1;
pReplica->port = tsServerPort;
tstrncpy(pReplica->fqdn, tsLocalFqdn, TSDB_FQDN_LEN);
} }
static void mmBuildOptionForOpen(SMnodeMgmt *pMgmt, SMnodeOpt *pOption) { static void mmBuildOptionForOpen(SMnodeMgmt *pMgmt, const SReplica *pReplica, SMnodeOpt *pOption) {
pOption->deploy = false;
pOption->standby = false; pOption->standby = false;
pOption->deploy = false;
pOption->msgCb = pMgmt->msgCb; pOption->msgCb = pMgmt->msgCb;
pOption->dnodeId = pMgmt->pData->dnodeId; pOption->dnodeId = pMgmt->pData->dnodeId;
if (pReplica->id > 0) {
if (pMgmt->replica > 0) {
pOption->standby = true; pOption->standby = true;
pOption->replica = 1; pOption->replica = *pReplica;
pOption->selfIndex = 0;
SReplica *pReplica = &pOption->replicas[0];
for (int32_t i = 0; i < pMgmt->replica; ++i) {
if (pMgmt->replicas[i].id != pMgmt->pData->dnodeId) continue;
pReplica->id = pMgmt->replicas[i].id;
pReplica->port = pMgmt->replicas[i].port;
memcpy(pReplica->fqdn, pMgmt->replicas[i].fqdn, TSDB_FQDN_LEN);
}
} }
} }
...@@ -105,12 +91,13 @@ static int32_t mmOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) { ...@@ -105,12 +91,13 @@ static int32_t mmOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
pMgmt->path = pInput->path; pMgmt->path = pInput->path;
pMgmt->name = pInput->name; pMgmt->name = pInput->name;
pMgmt->msgCb = pInput->msgCb; pMgmt->msgCb = pInput->msgCb;
pMgmt->msgCb.putToQueueFp = (PutToQueueFp)mmPutRpcMsgToQueue; pMgmt->msgCb.putToQueueFp = (PutToQueueFp)mmPutMsgToQueue;
pMgmt->msgCb.mgmt = pMgmt; pMgmt->msgCb.mgmt = pMgmt;
taosThreadRwlockInit(&pMgmt->lock, NULL); taosThreadRwlockInit(&pMgmt->lock, NULL);
bool deployed = false; bool deployed = false;
if (mmReadFile(pMgmt, &deployed) != 0) { SReplica replica = {0};
if (mmReadFile(pMgmt, &replica, &deployed) != 0) {
dError("failed to read file since %s", terrstr()); dError("failed to read file since %s", terrstr());
mmClose(pMgmt); mmClose(pMgmt);
return -1; return -1;
...@@ -123,7 +110,7 @@ static int32_t mmOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) { ...@@ -123,7 +110,7 @@ static int32_t mmOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
mmBuildOptionForDeploy(pMgmt, pInput, &option); mmBuildOptionForDeploy(pMgmt, pInput, &option);
} else { } else {
dInfo("mnode start to open"); dInfo("mnode start to open");
mmBuildOptionForOpen(pMgmt, &option); mmBuildOptionForOpen(pMgmt, &replica, &option);
} }
pMgmt->pMnode = mndOpen(pMgmt->path, &option); pMgmt->pMnode = mndOpen(pMgmt->path, &option);
...@@ -141,8 +128,7 @@ static int32_t mmOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) { ...@@ -141,8 +128,7 @@ static int32_t mmOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
} }
tmsgReportStartup("mnode-worker", "initialized"); tmsgReportStartup("mnode-worker", "initialized");
if (!deployed || pMgmt->replica > 0) { if (!deployed || replica.id > 0) {
pMgmt->replica = 0;
deployed = true; deployed = true;
if (mmWriteFile(pMgmt, NULL, deployed) != 0) { if (mmWriteFile(pMgmt, NULL, deployed) != 0) {
dError("failed to write mnode file since %s", terrstr()); dError("failed to write mnode file since %s", terrstr());
...@@ -178,22 +164,3 @@ SMgmtFunc mmGetMgmtFunc() { ...@@ -178,22 +164,3 @@ SMgmtFunc mmGetMgmtFunc() {
return mgmtFunc; return mgmtFunc;
} }
int32_t mmAcquire(SMnodeMgmt *pMgmt) {
int32_t code = 0;
taosThreadRwlockRdlock(&pMgmt->lock);
if (pMgmt->stopped) {
code = -1;
} else {
atomic_add_fetch_32(&pMgmt->refCount, 1);
}
taosThreadRwlockUnlock(&pMgmt->lock);
return code;
}
void mmRelease(SMnodeMgmt *pMgmt) {
taosThreadRwlockRdlock(&pMgmt->lock);
atomic_sub_fetch_32(&pMgmt->refCount, 1);
taosThreadRwlockUnlock(&pMgmt->lock);
}
\ No newline at end of file
source/dnode/mgmt/mgmt_mnode/src/mmWorker.c
浏览文件 @ cdb8cae0
...@@ -16,6 +16,25 @@ ...@@ -16,6 +16,25 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "mmInt.h" #include "mmInt.h"
static inline int32_t mmAcquire(SMnodeMgmt *pMgmt) {
int32_t code = 0;
taosThreadRwlockRdlock(&pMgmt->lock);
if (pMgmt->stopped) {
code = -1;
} else {
atomic_add_fetch_32(&pMgmt->refCount, 1);
}
taosThreadRwlockUnlock(&pMgmt->lock);
return code;
}
static inline void mmRelease(SMnodeMgmt *pMgmt) {
taosThreadRwlockRdlock(&pMgmt->lock);
atomic_sub_fetch_32(&pMgmt->refCount, 1);
taosThreadRwlockUnlock(&pMgmt->lock);
}
static inline void mmSendRsp(SRpcMsg *pMsg, int32_t code) { static inline void mmSendRsp(SRpcMsg *pMsg, int32_t code) {
SRpcMsg rsp = { SRpcMsg rsp = {
.code = code, .code = code,
...@@ -26,7 +45,7 @@ static inline void mmSendRsp(SRpcMsg *pMsg, int32_t code) { ...@@ -26,7 +45,7 @@ static inline void mmSendRsp(SRpcMsg *pMsg, int32_t code) {
tmsgSendRsp(&rsp); tmsgSendRsp(&rsp);
} }
static void mmProcessQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) { static void mmProcessRpcMsg(SQueueInfo *pInfo, SRpcMsg *pMsg) {
SMnodeMgmt *pMgmt = pInfo->ahandle; SMnodeMgmt *pMgmt = pInfo->ahandle;
int32_t code = -1; int32_t code = -1;
dTrace("msg:%p, get from mnode queue", pMsg); dTrace("msg:%p, get from mnode queue", pMsg);
...@@ -53,11 +72,10 @@ static void mmProcessQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) { ...@@ -53,11 +72,10 @@ static void mmProcessQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
taosFreeQitem(pMsg); taosFreeQitem(pMsg);
} }
static void mmProcessSyncQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) { static void mmProcessSyncMsg(SQueueInfo *pInfo, SRpcMsg *pMsg) {
SMnodeMgmt *pMgmt = pInfo->ahandle; SMnodeMgmt *pMgmt = pInfo->ahandle;
dTrace("msg:%p, get from mnode-sync queue", pMsg);
pMsg->info.node = pMgmt->pMnode; pMsg->info.node = pMgmt->pMnode;
dTrace("msg:%p, get from mnode-sync queue", pMsg);
SMsgHead *pHead = pMsg->pCont; SMsgHead *pHead = pMsg->pCont;
pHead->contLen = ntohl(pHead->contLen); pHead->contLen = ntohl(pHead->contLen);
...@@ -70,66 +88,69 @@ static void mmProcessSyncQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) { ...@@ -70,66 +88,69 @@ static void mmProcessSyncQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
taosFreeQitem(pMsg); taosFreeQitem(pMsg);
} }
static int32_t mmPutNodeMsgToWorker(SSingleWorker *pWorker, SRpcMsg *pMsg) { static inline int32_t mmPutMsgToWorker(SMnodeMgmt *pMgmt, SSingleWorker *pWorker, SRpcMsg *pMsg) {
dTrace("msg:%p, put into %s queue, type:%s", pMsg, pWorker->name, TMSG_INFO(pMsg->msgType)); if (mmAcquire(pMgmt) == 0) {
taosWriteQitem(pWorker->queue, pMsg); dTrace("msg:%p, put into %s queue, type:%s", pMsg, pWorker->name, TMSG_INFO(pMsg->msgType));
return 0; taosWriteQitem(pWorker->queue, pMsg);
mmRelease(pMgmt);
return 0;
} else {
dTrace("msg:%p, failed to put into %s queue since %s, type:%s", pMsg, pWorker->name, terrstr(),
TMSG_INFO(pMsg->msgType));
return -1;
}
} }
int32_t mmPutNodeMsgToWriteQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) { int32_t mmPutMsgToWriteQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
return mmPutNodeMsgToWorker(&pMgmt->writeWorker, pMsg); return mmPutMsgToWorker(pMgmt, &pMgmt->writeWorker, pMsg);
} }
int32_t mmPutNodeMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) { int32_t mmPutMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
return mmPutNodeMsgToWorker(&pMgmt->syncWorker, pMsg); return mmPutMsgToWorker(pMgmt, &pMgmt->syncWorker, pMsg);
} }
int32_t mmPutNodeMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) { int32_t mmPutMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
return mmPutNodeMsgToWorker(&pMgmt->readWorker, pMsg); return mmPutMsgToWorker(pMgmt, &pMgmt->readWorker, pMsg);
} }
int32_t mmPutNodeMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) { int32_t mmPutMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
mndPreprocessQueryMsg(pMgmt->pMnode, pMsg); if (mndPreprocessQueryMsg(pMgmt->pMnode, pMsg) != 0) {
dError("msg:%p, failed to pre-process in mnode since %s, type:%s", pMsg, terrstr(), TMSG_INFO(pMsg->msgType));
return mmPutNodeMsgToWorker(&pMgmt->queryWorker, pMsg); return -1;
}
return mmPutMsgToWorker(pMgmt, &pMgmt->queryWorker, pMsg);
} }
int32_t mmPutNodeMsgToMonitorQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) { int32_t mmPutMsgToMonitorQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
return mmPutNodeMsgToWorker(&pMgmt->monitorWorker, pMsg); return mmPutMsgToWorker(pMgmt, &pMgmt->monitorWorker, pMsg);
} }
int32_t mmPutRpcMsgToQueue(SMnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) { int32_t mmPutMsgToQueue(SMnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) {
SRpcMsg *pMsg = taosAllocateQitem(sizeof(SRpcMsg), RPC_QITEM); SSingleWorker *pWorker = NULL;
if (pMsg == NULL) return -1;
memcpy(pMsg, pRpc, sizeof(SRpcMsg));
switch (qtype) { switch (qtype) {
case WRITE_QUEUE: case WRITE_QUEUE:
dTrace("msg:%p, is created and will put into vnode-write queue", pMsg); pWorker = &pMgmt->writeWorker;
taosWriteQitem(pMgmt->writeWorker.queue, pMsg); break;
return 0;
case QUERY_QUEUE: case QUERY_QUEUE:
dTrace("msg:%p, is created and will put into vnode-query queue", pMsg); pWorker = &pMgmt->queryWorker;
taosWriteQitem(pMgmt->queryWorker.queue, pMsg); break;
return 0;
case READ_QUEUE: case READ_QUEUE:
dTrace("msg:%p, is created and will put into vnode-read queue", pMsg); pWorker = &pMgmt->readWorker;
taosWriteQitem(pMgmt->readWorker.queue, pMsg); break;
return 0;
case SYNC_QUEUE: case SYNC_QUEUE:
if (mmAcquire(pMgmt) == 0) { pWorker = &pMgmt->syncWorker;
dTrace("msg:%p, is created and will put into vnode-sync queue", pMsg); break;
taosWriteQitem(pMgmt->syncWorker.queue, pMsg);
mmRelease(pMgmt);
return 0;
} else {
return -1;
}
default: default:
terrno = TSDB_CODE_INVALID_PARA; terrno = TSDB_CODE_INVALID_PARA;
return -1;
} }
if (pWorker == NULL) return -1;
SRpcMsg *pMsg = taosAllocateQitem(sizeof(SRpcMsg), RPC_QITEM);
if (pMsg == NULL) return -1;
memcpy(pMsg, pRpc, sizeof(SRpcMsg));
dTrace("msg:%p, is created and will put int %s queue", pMsg, pWorker->name);
return mmPutMsgToWorker(pMgmt, pWorker, pMsg);
} }
int32_t mmStartWorker(SMnodeMgmt *pMgmt) { int32_t mmStartWorker(SMnodeMgmt *pMgmt) {
...@@ -137,7 +158,7 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt) { ...@@ -137,7 +158,7 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt) {
.min = tsNumOfMnodeQueryThreads, .min = tsNumOfMnodeQueryThreads,
.max = tsNumOfMnodeQueryThreads, .max = tsNumOfMnodeQueryThreads,
.name = "mnode-query", .name = "mnode-query",
.fp = (FItem)mmProcessQueue, .fp = (FItem)mmProcessRpcMsg,
.param = pMgmt, .param = pMgmt,
}; };
if (tSingleWorkerInit(&pMgmt->queryWorker, &qCfg) != 0) { if (tSingleWorkerInit(&pMgmt->queryWorker, &qCfg) != 0) {
...@@ -149,7 +170,7 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt) { ...@@ -149,7 +170,7 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt) {
.min = tsNumOfMnodeReadThreads, .min = tsNumOfMnodeReadThreads,
.max = tsNumOfMnodeReadThreads, .max = tsNumOfMnodeReadThreads,
.name = "mnode-read", .name = "mnode-read",
.fp = (FItem)mmProcessQueue, .fp = (FItem)mmProcessRpcMsg,
.param = pMgmt, .param = pMgmt,
}; };
if (tSingleWorkerInit(&pMgmt->readWorker, &rCfg) != 0) { if (tSingleWorkerInit(&pMgmt->readWorker, &rCfg) != 0) {
...@@ -161,7 +182,7 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt) { ...@@ -161,7 +182,7 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt) {
.min = 1, .min = 1,
.max = 1, .max = 1,
.name = "mnode-write", .name = "mnode-write",
.fp = (FItem)mmProcessQueue, .fp = (FItem)mmProcessRpcMsg,
.param = pMgmt, .param = pMgmt,
}; };
if (tSingleWorkerInit(&pMgmt->writeWorker, &wCfg) != 0) { if (tSingleWorkerInit(&pMgmt->writeWorker, &wCfg) != 0) {
...@@ -173,7 +194,7 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt) { ...@@ -173,7 +194,7 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt) {
.min = 1, .min = 1,
.max = 1, .max = 1,
.name = "mnode-sync", .name = "mnode-sync",
.fp = (FItem)mmProcessSyncQueue, .fp = (FItem)mmProcessSyncMsg,
.param = pMgmt, .param = pMgmt,
}; };
if (tSingleWorkerInit(&pMgmt->syncWorker, &sCfg) != 0) { if (tSingleWorkerInit(&pMgmt->syncWorker, &sCfg) != 0) {
...@@ -185,7 +206,7 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt) { ...@@ -185,7 +206,7 @@ int32_t mmStartWorker(SMnodeMgmt *pMgmt) {
.min = 1, .min = 1,
.max = 1, .max = 1,
.name = "mnode-monitor", .name = "mnode-monitor",
.fp = (FItem)mmProcessQueue, .fp = (FItem)mmProcessRpcMsg,
.param = pMgmt, .param = pMgmt,
}; };
if (tSingleWorkerInit(&pMgmt->monitorWorker, &mCfg) != 0) { if (tSingleWorkerInit(&pMgmt->monitorWorker, &mCfg) != 0) {
......
source/dnode/mgmt/node_mgmt/src/dmEnv.c
浏览文件 @ cdb8cae0
...@@ -123,10 +123,12 @@ static int32_t dmProcessCreateNodeReq(EDndNodeType ntype, SRpcMsg *pMsg) { ...@@ -123,10 +123,12 @@ static int32_t dmProcessCreateNodeReq(EDndNodeType ntype, SRpcMsg *pMsg) {
dError("node:%s, failed to create since %s", pWrapper->name, terrstr()); dError("node:%s, failed to create since %s", pWrapper->name, terrstr());
} else { } else {
dInfo("node:%s, has been created", pWrapper->name); dInfo("node:%s, has been created", pWrapper->name);
(void)dmOpenNode(pWrapper); code = dmOpenNode(pWrapper);
(void)dmStartNode(pWrapper); if (code != 0) {
pWrapper->required = true; code = dmStartNode(pWrapper);
}
pWrapper->deployed = true; pWrapper->deployed = true;
pWrapper->required = true;
pWrapper->proc.ptype = pDnode->ptype; pWrapper->proc.ptype = pDnode->ptype;
} }
......
source/dnode/mnode/impl/inc/mndInt.h
浏览文件 @ cdb8cae0
...@@ -76,11 +76,12 @@ typedef struct { ...@@ -76,11 +76,12 @@ typedef struct {
} STelemMgmt; } STelemMgmt;
typedef struct { typedef struct {
sem_t syncSem; sem_t syncSem;
int64_t sync; int64_t sync;
bool standby; bool standby;
int32_t errCode; SReplica replica;
int32_t transId; int32_t errCode;
int32_t transId;
} SSyncMgmt; } SSyncMgmt;
typedef struct { typedef struct {
...@@ -98,9 +99,6 @@ typedef struct SMnode { ...@@ -98,9 +99,6 @@ typedef struct SMnode {
bool stopped; bool stopped;
bool restored; bool restored;
bool deploy; bool deploy;
int8_t replica;
int8_t selfIndex;
SReplica replicas[TSDB_MAX_REPLICA];
char *path; char *path;
int64_t checkTime; int64_t checkTime;
SSdb *pSdb; SSdb *pSdb;
......
source/dnode/mnode/impl/src/mndDnode.c
浏览文件 @ cdb8cae0
...@@ -95,8 +95,8 @@ static int32_t mndCreateDefaultDnode(SMnode *pMnode) { ...@@ -95,8 +95,8 @@ static int32_t mndCreateDefaultDnode(SMnode *pMnode) {
dnodeObj.id = 1; dnodeObj.id = 1;
dnodeObj.createdTime = taosGetTimestampMs(); dnodeObj.createdTime = taosGetTimestampMs();
dnodeObj.updateTime = dnodeObj.createdTime; dnodeObj.updateTime = dnodeObj.createdTime;
dnodeObj.port = pMnode->replicas[0].port; dnodeObj.port = tsServerPort;
memcpy(&dnodeObj.fqdn, pMnode->replicas[0].fqdn, TSDB_FQDN_LEN); memcpy(&dnodeObj.fqdn, tsLocalFqdn, TSDB_FQDN_LEN);
snprintf(dnodeObj.ep, TSDB_EP_LEN, "%s:%u", dnodeObj.fqdn, dnodeObj.port); snprintf(dnodeObj.ep, TSDB_EP_LEN, "%s:%u", dnodeObj.fqdn, dnodeObj.port);
pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, NULL); pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, NULL);
......
source/dnode/mnode/impl/src/mndMain.c
浏览文件 @ cdb8cae0
...@@ -289,11 +289,9 @@ static int32_t mndExecSteps(SMnode *pMnode) { ...@@ -289,11 +289,9 @@ static int32_t mndExecSteps(SMnode *pMnode) {
} }
static void mndSetOptions(SMnode *pMnode, const SMnodeOpt *pOption) { static void mndSetOptions(SMnode *pMnode, const SMnodeOpt *pOption) {
pMnode->replica = pOption->replica;
pMnode->selfIndex = pOption->selfIndex;
memcpy(&pMnode->replicas, pOption->replicas, sizeof(SReplica) * TSDB_MAX_REPLICA);
pMnode->msgCb = pOption->msgCb; pMnode->msgCb = pOption->msgCb;
pMnode->selfDnodeId = pOption->dnodeId; pMnode->selfDnodeId = pOption->dnodeId;
pMnode->syncMgmt.replica = pOption->replica;
pMnode->syncMgmt.standby = pOption->standby; pMnode->syncMgmt.standby = pOption->standby;
} }
......
source/dnode/mnode/impl/src/mndSync.c
浏览文件 @ cdb8cae0
...@@ -188,15 +188,15 @@ int32_t mndInitSync(SMnode *pMnode) { ...@@ -188,15 +188,15 @@ int32_t mndInitSync(SMnode *pMnode) {
syncInfo.isStandBy = pMgmt->standby; syncInfo.isStandBy = pMgmt->standby;
syncInfo.snapshotEnable = true; syncInfo.snapshotEnable = true;
SSyncCfg *pCfg = &syncInfo.syncCfg; mInfo("start to open mnode sync, standby:%d", pMgmt->standby);
pCfg->replicaNum = pMnode->replica; if (pMgmt->standby || pMgmt->replica.id > 0) {
pCfg->myIndex = pMnode->selfIndex; SSyncCfg *pCfg = &syncInfo.syncCfg;
mInfo("start to open mnode sync, replica:%d myindex:%d standby:%d", pCfg->replicaNum, pCfg->myIndex, pMgmt->standby); pCfg->replicaNum = 1;
for (int32_t i = 0; i < pMnode->replica; ++i) { pCfg->myIndex = 0;
SNodeInfo *pNode = &pCfg->nodeInfo[i]; SNodeInfo *pNode = &pCfg->nodeInfo[0];
tstrncpy(pNode->nodeFqdn, pMnode->replicas[i].fqdn, sizeof(pNode->nodeFqdn)); tstrncpy(pNode->nodeFqdn, pMgmt->replica.fqdn, sizeof(pNode->nodeFqdn));
pNode->nodePort = pMnode->replicas[i].port; pNode->nodePort = pMgmt->replica.port;
mInfo("index:%d, fqdn:%s port:%d", i, pNode->nodeFqdn, pNode->nodePort); mInfo("fqdn:%s port:%u", pNode->nodeFqdn, pNode->nodePort);
} }
tsem_init(&pMgmt->syncSem, 0, 0); tsem_init(&pMgmt->syncSem, 0, 0);
......
source/libs/executor/src/groupoperator.c
浏览文件 @ cdb8cae0
...@@ -582,6 +582,15 @@ int32_t* setupColumnOffset(const SSDataBlock* pBlock, int32_t rowCapacity) { ...@@ -582,6 +582,15 @@ int32_t* setupColumnOffset(const SSDataBlock* pBlock, int32_t rowCapacity) {
return offset; return offset;
} }
static void clearPartitionOperator(SPartitionOperatorInfo* pInfo) {
void *ite = NULL;
while( (ite = taosHashIterate(pInfo->pGroupSet, ite)) != NULL ) {
taosArrayDestroy( ((SDataGroupInfo *)ite)->pPageList);
}
taosHashClear(pInfo->pGroupSet);
clearDiskbasedBuf(pInfo->pBuf);
}
static SSDataBlock* buildPartitionResult(SOperatorInfo* pOperator) { static SSDataBlock* buildPartitionResult(SOperatorInfo* pOperator) {
SPartitionOperatorInfo* pInfo = pOperator->info; SPartitionOperatorInfo* pInfo = pOperator->info;
...@@ -591,6 +600,7 @@ static SSDataBlock* buildPartitionResult(SOperatorInfo* pOperator) { ...@@ -591,6 +600,7 @@ static SSDataBlock* buildPartitionResult(SOperatorInfo* pOperator) {
pInfo->pGroupIter = taosHashIterate(pInfo->pGroupSet, pInfo->pGroupIter); pInfo->pGroupIter = taosHashIterate(pInfo->pGroupSet, pInfo->pGroupIter);
if (pInfo->pGroupIter == NULL) { if (pInfo->pGroupIter == NULL) {
doSetOperatorCompleted(pOperator); doSetOperatorCompleted(pOperator);
clearPartitionOperator(pInfo);
return NULL; return NULL;
} }
......
source/libs/function/inc/builtinsimpl.h
浏览文件 @ cdb8cae0
...@@ -93,10 +93,14 @@ int32_t diffFunction(SqlFunctionCtx *pCtx); ...@@ -93,10 +93,14 @@ int32_t diffFunction(SqlFunctionCtx *pCtx);
bool getFirstLastFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv); bool getFirstLastFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
int32_t firstFunction(SqlFunctionCtx *pCtx); int32_t firstFunction(SqlFunctionCtx *pCtx);
int32_t firstFunctionMerge(SqlFunctionCtx *pCtx);
int32_t lastFunction(SqlFunctionCtx *pCtx); int32_t lastFunction(SqlFunctionCtx *pCtx);
int32_t lastFunctionMerge(SqlFunctionCtx *pCtx);
int32_t firstLastFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock); int32_t firstLastFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
int32_t firstLastPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
int32_t firstCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx); int32_t firstCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
int32_t lastCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx); int32_t lastCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
int32_t getFirstLastInfoSize(int32_t resBytes);
bool getTopBotFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv); bool getTopBotFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv);
bool getTopBotMergeFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv); bool getTopBotMergeFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv);
......
source/libs/function/src/builtins.c
浏览文件 @ cdb8cae0
...@@ -954,6 +954,41 @@ static int32_t translateFirstLast(SFunctionNode* pFunc, char* pErrBuf, int32_t l ...@@ -954,6 +954,41 @@ static int32_t translateFirstLast(SFunctionNode* pFunc, char* pErrBuf, int32_t l
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t translateFirstLastImpl(SFunctionNode* pFunc, char* pErrBuf, int32_t len, bool isPartial) {
// first(col_list) will be rewritten as first(col)
if (2 != LIST_LENGTH(pFunc->pParameterList)) { //input has two params c0,ts, is this a bug?
return TSDB_CODE_SUCCESS;
}
SNode* pPara = nodesListGetNode(pFunc->pParameterList, 0);
uint8_t paraType = ((SExprNode*)pPara)->resType.type;
int32_t paraBytes = ((SExprNode*)pPara)->resType.bytes;
if (isPartial) {
if (QUERY_NODE_COLUMN != nodeType(pPara)) {
return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR,
"The parameters of first/last can only be columns");
}
pFunc->node.resType = (SDataType){.bytes = getFirstLastInfoSize(paraBytes) + VARSTR_HEADER_SIZE,
.type = TSDB_DATA_TYPE_BINARY};
} else {
if (TSDB_DATA_TYPE_BINARY != paraType) {
return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
}
pFunc->node.resType = ((SExprNode*)pPara)->resType;
}
return TSDB_CODE_SUCCESS;
}
static int32_t translateFirstLastPartial(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
return translateFirstLastImpl(pFunc, pErrBuf, len, true);
}
static int32_t translateFirstLastMerge(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
return translateFirstLastImpl(pFunc, pErrBuf, len, false);
}
static int32_t translateUnique(SFunctionNode* pFunc, char* pErrBuf, int32_t len) { static int32_t translateUnique(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
if (1 != LIST_LENGTH(pFunc->pParameterList)) { if (1 != LIST_LENGTH(pFunc->pParameterList)) {
return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName); return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
...@@ -1704,6 +1739,30 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { ...@@ -1704,6 +1739,30 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
.initFunc = functionSetup, .initFunc = functionSetup,
.processFunc = firstFunction, .processFunc = firstFunction,
.finalizeFunc = firstLastFinalize, .finalizeFunc = firstLastFinalize,
.pPartialFunc = "_first_partial",
.pMergeFunc = "_first_merge",
.combineFunc = firstCombine,
},
{
.name = "_first_partial",
.type = FUNCTION_TYPE_FIRST_PARTIAL,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC,
.translateFunc = translateFirstLastPartial,
.getEnvFunc = getFirstLastFuncEnv,
.initFunc = functionSetup,
.processFunc = firstFunction,
.finalizeFunc = firstLastPartialFinalize,
.combineFunc = firstCombine,
},
{
.name = "_first_merge",
.type = FUNCTION_TYPE_FIRST_MERGE,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC,
.translateFunc = translateFirstLastMerge,
.getEnvFunc = getFirstLastFuncEnv,
.initFunc = functionSetup,
.processFunc = firstFunctionMerge,
.finalizeFunc = firstLastFinalize,
.combineFunc = firstCombine, .combineFunc = firstCombine,
}, },
{ {
...@@ -1715,6 +1774,30 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { ...@@ -1715,6 +1774,30 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
.initFunc = functionSetup, .initFunc = functionSetup,
.processFunc = lastFunction, .processFunc = lastFunction,
.finalizeFunc = firstLastFinalize, .finalizeFunc = firstLastFinalize,
.pPartialFunc = "_last_partial",
.pMergeFunc = "_last_merge",
.combineFunc = lastCombine,
},
{
.name = "_last_partial",
.type = FUNCTION_TYPE_LAST_PARTIAL,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC,
.translateFunc = translateFirstLastPartial,
.getEnvFunc = getFirstLastFuncEnv,
.initFunc = functionSetup,
.processFunc = lastFunction,
.finalizeFunc = firstLastPartialFinalize,
.combineFunc = lastCombine,
},
{
.name = "_last_merge",
.type = FUNCTION_TYPE_LAST_MERGE,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC,
.translateFunc = translateFirstLastMerge,
.getEnvFunc = getFirstLastFuncEnv,
.initFunc = functionSetup,
.processFunc = lastFunctionMerge,
.finalizeFunc = firstLastFinalize,
.combineFunc = lastCombine, .combineFunc = lastCombine,
}, },
{ {
......
source/libs/function/src/builtinsimpl.c
浏览文件 @ cdb8cae0
...@@ -71,6 +71,12 @@ typedef struct STopBotRes { ...@@ -71,6 +71,12 @@ typedef struct STopBotRes {
STopBotResItem* pItems; STopBotResItem* pItems;
} STopBotRes; } STopBotRes;
typedef struct SFirstLastRes {
bool hasResult;
int32_t bytes;
char buf[];
} SFirstLastRes;
typedef struct SStddevRes { typedef struct SStddevRes {
double result; double result;
int64_t count; int64_t count;
...@@ -2230,9 +2236,13 @@ int32_t apercentileCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) ...@@ -2230,9 +2236,13 @@ int32_t apercentileCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx)
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t getFirstLastInfoSize(int32_t resBytes) {
return sizeof(SFirstLastRes) + resBytes + sizeof(int64_t);
}
bool getFirstLastFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) { bool getFirstLastFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
SColumnNode* pNode = (SColumnNode*)nodesListGetNode(pFunc->pParameterList, 0); SColumnNode* pNode = (SColumnNode *)nodesListGetNode(pFunc->pParameterList, 0);
pEnv->calcMemSize = pNode->node.resType.bytes + sizeof(int64_t); pEnv->calcMemSize = sizeof(SFirstLastRes) + pNode->node.resType.bytes + sizeof(int64_t);
return true; return true;
} }
...@@ -2256,12 +2266,13 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) { ...@@ -2256,12 +2266,13 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
int32_t numOfElems = 0; int32_t numOfElems = 0;
SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx); SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
char* buf = GET_ROWCELL_INTERBUF(pResInfo); SFirstLastRes *pInfo = GET_ROWCELL_INTERBUF(pResInfo);
SInputColumnInfoData* pInput = &pCtx->input; SInputColumnInfoData* pInput = &pCtx->input;
SColumnInfoData* pInputCol = pInput->pData[0]; SColumnInfoData* pInputCol = pInput->pData[0];
int32_t bytes = pInputCol->info.bytes; int32_t bytes = pInputCol->info.bytes;
pInfo->bytes = bytes;
// All null data column, return directly. // All null data column, return directly.
if (pInput->colDataAggIsSet && (pInput->pColumnDataAgg[0]->numOfNull == pInput->totalRows)) { if (pInput->colDataAggIsSet && (pInput->pColumnDataAgg[0]->numOfNull == pInput->totalRows)) {
...@@ -2279,7 +2290,7 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) { ...@@ -2279,7 +2290,7 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
if (blockDataOrder == TSDB_ORDER_ASC) { if (blockDataOrder == TSDB_ORDER_ASC) {
// filter according to current result firstly // filter according to current result firstly
if (pResInfo->numOfRes > 0) { if (pResInfo->numOfRes > 0) {
TSKEY ts = *(TSKEY*)(buf + bytes); TSKEY ts = *(TSKEY*)(pInfo->buf + bytes);
if (ts < startKey) { if (ts < startKey) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
...@@ -2295,9 +2306,10 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) { ...@@ -2295,9 +2306,10 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
char* data = colDataGetData(pInputCol, i); char* data = colDataGetData(pInputCol, i);
TSKEY cts = getRowPTs(pInput->pPTS, i); TSKEY cts = getRowPTs(pInput->pPTS, i);
if (pResInfo->numOfRes == 0 || *(TSKEY*)(buf + bytes) > cts) { if (pResInfo->numOfRes == 0 || *(TSKEY*)(pInfo->buf + bytes) > cts) {
memcpy(buf, data, bytes); memcpy(pInfo->buf, data, bytes);
*(TSKEY*)(buf + bytes) = cts; *(TSKEY*)(pInfo->buf + bytes) = cts;
pInfo->hasResult = true;
// DO_UPDATE_TAG_COLUMNS(pCtx, ts); // DO_UPDATE_TAG_COLUMNS(pCtx, ts);
pResInfo->numOfRes = 1; pResInfo->numOfRes = 1;
...@@ -2308,7 +2320,7 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) { ...@@ -2308,7 +2320,7 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
// in case of descending order time stamp serial, which usually happens as the results of the nest query, // in case of descending order time stamp serial, which usually happens as the results of the nest query,
// all data needs to be check. // all data needs to be check.
if (pResInfo->numOfRes > 0) { if (pResInfo->numOfRes > 0) {
TSKEY ts = *(TSKEY*)(buf + bytes); TSKEY ts = *(TSKEY*)(pInfo->buf + bytes);
if (ts < endKey) { if (ts < endKey) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
...@@ -2324,9 +2336,10 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) { ...@@ -2324,9 +2336,10 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
char* data = colDataGetData(pInputCol, i); char* data = colDataGetData(pInputCol, i);
TSKEY cts = getRowPTs(pInput->pPTS, i); TSKEY cts = getRowPTs(pInput->pPTS, i);
if (pResInfo->numOfRes == 0 || *(TSKEY*)(buf + bytes) > cts) { if (pResInfo->numOfRes == 0 || *(TSKEY*)(pInfo->buf + bytes) > cts) {
memcpy(buf, data, bytes); memcpy(pInfo->buf, data, bytes);
*(TSKEY*)(buf + bytes) = cts; *(TSKEY*)(pInfo->buf + bytes) = cts;
pInfo->hasResult = true;
// DO_UPDATE_TAG_COLUMNS(pCtx, ts); // DO_UPDATE_TAG_COLUMNS(pCtx, ts);
pResInfo->numOfRes = 1; pResInfo->numOfRes = 1;
break; break;
...@@ -2342,12 +2355,13 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) { ...@@ -2342,12 +2355,13 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
int32_t numOfElems = 0; int32_t numOfElems = 0;
SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx); SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
char* buf = GET_ROWCELL_INTERBUF(pResInfo); SFirstLastRes *pInfo = GET_ROWCELL_INTERBUF(pResInfo);
SInputColumnInfoData* pInput = &pCtx->input; SInputColumnInfoData* pInput = &pCtx->input;
SColumnInfoData* pInputCol = pInput->pData[0]; SColumnInfoData* pInputCol = pInput->pData[0];
int32_t bytes = pInputCol->info.bytes; int32_t bytes = pInputCol->info.bytes;
pInfo->bytes = bytes;
// All null data column, return directly. // All null data column, return directly.
if (pInput->colDataAggIsSet && (pInput->pColumnDataAgg[0]->numOfNull == pInput->totalRows)) { if (pInput->colDataAggIsSet && (pInput->pColumnDataAgg[0]->numOfNull == pInput->totalRows)) {
...@@ -2372,10 +2386,11 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) { ...@@ -2372,10 +2386,11 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
char* data = colDataGetData(pInputCol, i); char* data = colDataGetData(pInputCol, i);
TSKEY cts = getRowPTs(pInput->pPTS, i); TSKEY cts = getRowPTs(pInput->pPTS, i);
if (pResInfo->numOfRes == 0 || *(TSKEY*)(buf + bytes) < cts) { if (pResInfo->numOfRes == 0 || *(TSKEY*)(pInfo->buf + bytes) < cts) {
memcpy(buf, data, bytes); memcpy(pInfo->buf, data, bytes);
*(TSKEY*)(buf + bytes) = cts; *(TSKEY*)(pInfo->buf + bytes) = cts;
// DO_UPDATE_TAG_COLUMNS(pCtx, ts); // DO_UPDATE_TAG_COLUMNS(pCtx, ts);
pInfo->hasResult = true;
pResInfo->numOfRes = 1; pResInfo->numOfRes = 1;
} }
break; break;
...@@ -2390,9 +2405,10 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) { ...@@ -2390,9 +2405,10 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
char* data = colDataGetData(pInputCol, i); char* data = colDataGetData(pInputCol, i);
TSKEY cts = getRowPTs(pInput->pPTS, i); TSKEY cts = getRowPTs(pInput->pPTS, i);
if (pResInfo->numOfRes == 0 || *(TSKEY*)(buf + bytes) < cts) { if (pResInfo->numOfRes == 0 || *(TSKEY*)(pInfo->buf + bytes) < cts) {
memcpy(buf, data, bytes); memcpy(pInfo->buf, data, bytes);
*(TSKEY*)(buf + bytes) = cts; *(TSKEY*)(pInfo->buf + bytes) = cts;
pInfo->hasResult = true;
pResInfo->numOfRes = 1; pResInfo->numOfRes = 1;
// DO_UPDATE_TAG_COLUMNS(pCtx, ts); // DO_UPDATE_TAG_COLUMNS(pCtx, ts);
} }
...@@ -2404,6 +2420,56 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) { ...@@ -2404,6 +2420,56 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static void firstLastTransferInfo(SFirstLastRes* pInput, SFirstLastRes* pOutput, bool isFirst) {
if (!pInput->hasResult) {
return;
}
pOutput->bytes = pInput->bytes;
TSKEY *tsIn = (TSKEY*)(pInput->buf + pInput->bytes);
TSKEY *tsOut = (TSKEY*)(pOutput->buf + pInput->bytes);
if (pOutput->hasResult) {
if (isFirst) {
if (*tsIn > *tsOut) {
return;
}
} else {
if (*tsIn < *tsOut) {
return;
}
}
}
*tsOut = *tsIn;
memcpy(pOutput->buf, pInput->buf, pOutput->bytes);
pOutput->hasResult = true;
return;
}
static int32_t firstLastFunctionMergeImpl(SqlFunctionCtx *pCtx, bool isFirstQuery) {
SInputColumnInfoData* pInput = &pCtx->input;
SColumnInfoData* pCol = pInput->pData[0];
ASSERT(pCol->info.type == TSDB_DATA_TYPE_BINARY);
SFirstLastRes* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
int32_t start = pInput->startRowIndex;
char* data = colDataGetData(pCol, start);
SFirstLastRes* pInputInfo = (SFirstLastRes *)varDataVal(data);
firstLastTransferInfo(pInputInfo, pInfo, isFirstQuery);
SET_VAL(GET_RES_INFO(pCtx), 1, 1);
return TSDB_CODE_SUCCESS;
}
int32_t firstFunctionMerge(SqlFunctionCtx *pCtx) {
return firstLastFunctionMergeImpl(pCtx, true);
}
int32_t lastFunctionMerge(SqlFunctionCtx *pCtx) {
return firstLastFunctionMergeImpl(pCtx, false);
}
int32_t firstLastFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) { int32_t firstLastFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
int32_t slotId = pCtx->pExpr->base.resSchema.slotId; int32_t slotId = pCtx->pExpr->base.resSchema.slotId;
SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId); SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
...@@ -2411,12 +2477,30 @@ int32_t firstLastFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) { ...@@ -2411,12 +2477,30 @@ int32_t firstLastFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx); SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
pResInfo->isNullRes = (pResInfo->numOfRes == 0) ? 1 : 0; pResInfo->isNullRes = (pResInfo->numOfRes == 0) ? 1 : 0;
char* in = GET_ROWCELL_INTERBUF(pResInfo); SFirstLastRes* pRes = GET_ROWCELL_INTERBUF(pResInfo);
colDataAppend(pCol, pBlock->info.rows, in, pResInfo->isNullRes); colDataAppend(pCol, pBlock->info.rows, pRes->buf, pResInfo->isNullRes);
return pResInfo->numOfRes; return pResInfo->numOfRes;
} }
int32_t firstLastPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(pCtx);
SFirstLastRes* pRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
int32_t resultBytes = getFirstLastInfoSize(pRes->bytes);
char *res = taosMemoryCalloc(resultBytes + VARSTR_HEADER_SIZE, sizeof(char));
memcpy(varDataVal(res), pRes, resultBytes);
varDataSetLen(res, resultBytes);
int32_t slotId = pCtx->pExpr->base.resSchema.slotId;
SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
colDataAppend(pCol, pBlock->info.rows, res, false);
taosMemoryFree(res);
return 1;
}
int32_t lastCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) { int32_t lastCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) {
SResultRowEntryInfo* pDResInfo = GET_RES_INFO(pDestCtx); SResultRowEntryInfo* pDResInfo = GET_RES_INFO(pDestCtx);
char* pDBuf = GET_ROWCELL_INTERBUF(pDResInfo); char* pDBuf = GET_ROWCELL_INTERBUF(pDResInfo);
......
source/libs/parser/src/parInsertData.c
浏览文件 @ cdb8cae0
...@@ -283,6 +283,8 @@ void sortRemoveDataBlockDupRowsRaw(STableDataBlocks* dataBuf) { ...@@ -283,6 +283,8 @@ void sortRemoveDataBlockDupRowsRaw(STableDataBlocks* dataBuf) {
if (!dataBuf->ordered) { if (!dataBuf->ordered) {
char* pBlockData = pBlocks->data; char* pBlockData = pBlocks->data;
// todo. qsort is unstable, if timestamp is same, should get the last one
qsort(pBlockData, pBlocks->numOfRows, dataBuf->rowSize, rowDataCompar); qsort(pBlockData, pBlocks->numOfRows, dataBuf->rowSize, rowDataCompar);
int32_t i = 0; int32_t i = 0;
...@@ -350,6 +352,8 @@ int sortRemoveDataBlockDupRows(STableDataBlocks* dataBuf, SBlockKeyInfo* pBlkKey ...@@ -350,6 +352,8 @@ int sortRemoveDataBlockDupRows(STableDataBlocks* dataBuf, SBlockKeyInfo* pBlkKey
if (!dataBuf->ordered) { if (!dataBuf->ordered) {
pBlkKeyTuple = pBlkKeyInfo->pKeyTuple; pBlkKeyTuple = pBlkKeyInfo->pKeyTuple;
// todo. qsort is unstable, if timestamp is same, should get the last one
qsort(pBlkKeyTuple, nRows, sizeof(SBlockKeyTuple), rowDataCompar); qsort(pBlkKeyTuple, nRows, sizeof(SBlockKeyTuple), rowDataCompar);
pBlkKeyTuple = pBlkKeyInfo->pKeyTuple; pBlkKeyTuple = pBlkKeyInfo->pKeyTuple;
......
source/libs/sync/src/syncMain.c
浏览文件 @ cdb8cae0
...@@ -1658,14 +1658,24 @@ static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) { ...@@ -1658,14 +1658,24 @@ static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {
syncTimeout2RpcMsg(pSyncMsg, &rpcMsg); syncTimeout2RpcMsg(pSyncMsg, &rpcMsg);
syncRpcMsgLog2((char*)"==syncNodeEqHeartbeatTimer==", &rpcMsg); syncRpcMsgLog2((char*)"==syncNodeEqHeartbeatTimer==", &rpcMsg);
if (pSyncNode->FpEqMsg != NULL) { if (pSyncNode->FpEqMsg != NULL) {
pSyncNode->FpEqMsg(pSyncNode->msgcb, &rpcMsg); int32_t code = pSyncNode->FpEqMsg(pSyncNode->msgcb, &rpcMsg);
if (code != 0) {
sError("vgId:%d sync enqueue timer msg error, code:%d", pSyncNode->vgId, code);
rpcFreeCont(rpcMsg.pCont);
return;
}
} else { } else {
sTrace("syncNodeEqHeartbeatTimer pSyncNode->FpEqMsg is NULL"); sTrace("syncNodeEqHeartbeatTimer pSyncNode->FpEqMsg is NULL");
} }
syncTimeoutDestroy(pSyncMsg); syncTimeoutDestroy(pSyncMsg);
if (gSyncEnv != NULL) {
taosTmrReset(syncNodeEqHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager, taosTmrReset(syncNodeEqHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager,
&pSyncNode->pHeartbeatTimer); &pSyncNode->pHeartbeatTimer);
} else {
sError("sync env is already stop");
}
} else { } else {
sTrace("==syncNodeEqHeartbeatTimer== heartbeatTimerLogicClock:%" PRIu64 ", heartbeatTimerLogicClockUser:%" PRIu64 sTrace("==syncNodeEqHeartbeatTimer== heartbeatTimerLogicClock:%" PRIu64 ", heartbeatTimerLogicClockUser:%" PRIu64
"", "",
......
source/libs/transport/src/transCli.c
浏览文件 @ cdb8cae0
...@@ -983,6 +983,9 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) { ...@@ -983,6 +983,9 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
SEpSet epSet = {0}; SEpSet epSet = {0};
tDeserializeSEpSet(pResp->pCont, pResp->contLen, &epSet); tDeserializeSEpSet(pResp->pCont, pResp->contLen, &epSet);
pCtx->epSet = epSet; pCtx->epSet = epSet;
if (!transEpSetIsEqual(&epSet, &pCtx->epSet)) {
pCtx->retryCount = 0;
}
} }
addConnToPool(pThrd->pool, pConn); addConnToPool(pThrd->pool, pConn);
tTrace("use remote epset, current in use: %d, retry count:%d, try limit: %d", pEpSet->inUse, pCtx->retryCount + 1, tTrace("use remote epset, current in use: %d, retry count:%d, try limit: %d", pEpSet->inUse, pCtx->retryCount + 1,
......
source/util/src/talgo.c
浏览文件 @ cdb8cae0
...@@ -175,7 +175,13 @@ void *taosbsearch(const void *key, const void *base, int32_t nmemb, int32_t size ...@@ -175,7 +175,13 @@ void *taosbsearch(const void *key, const void *base, int32_t nmemb, int32_t size
c = compar(key, p); c = compar(key, p);
if (c == 0) { if (c == 0) {
break; if (flags == TD_GT){
lidx = midx + 1;
} else if(flags == TD_LT){
ridx = midx - 1;
}else{
break;
}
} else if (c < 0) { } else if (c < 0) {
ridx = midx - 1; ridx = midx - 1;
} else { } else {
...@@ -189,6 +195,10 @@ void *taosbsearch(const void *key, const void *base, int32_t nmemb, int32_t size ...@@ -189,6 +195,10 @@ void *taosbsearch(const void *key, const void *base, int32_t nmemb, int32_t size
return (c <= 0) ? p : (midx + 1 < nmemb ? p + size : NULL); return (c <= 0) ? p : (midx + 1 < nmemb ? p + size : NULL);
} else if (flags == TD_LE) { } else if (flags == TD_LE) {
return (c >= 0) ? p : (midx > 0 ? p - size : NULL); return (c >= 0) ? p : (midx > 0 ? p - size : NULL);
} else if (flags == TD_GT) {
return (c < 0) ? p : (midx + 1 < nmemb ? p + size : NULL);
} else if (flags == TD_LT) {
return (c > 0) ? p : (midx > 0 ? p - size : NULL);
} else { } else {
ASSERT(0); ASSERT(0);
} }
......
source/util/test/taosbsearchTest.cpp
浏览文件 @ cdb8cae0
...@@ -248,6 +248,170 @@ TEST(testCase, taosbsearch_greater_or_equal) { ...@@ -248,6 +248,170 @@ TEST(testCase, taosbsearch_greater_or_equal) {
ASSERT_EQ(pRet, nullptr); ASSERT_EQ(pRet, nullptr);
} }
TEST(testCase, taosbsearch_greater) {
// For equal test
int key = 3;
void *pRet = NULL;
pRet = taosbsearch((void *)&key, NULL, 0, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(pRet, nullptr);
// 1 element
int array1[1] = {5};
key = 1;
pRet = taosbsearch((void *)&key, (void *)array1, 1, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 5);
key = 6;
pRet = taosbsearch((void *)&key, (void *)array1, 1, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(pRet, nullptr);
key = 5;
pRet = taosbsearch((void *)&key, (void *)array1, 1, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(pRet, nullptr);
// 2 element
int array2[2] = {3, 6};
key = 1;
pRet = taosbsearch((void *)&key, (void *)array2, 2, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 3);
key = 3;
pRet = taosbsearch((void *)&key, (void *)array2, 2, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 6);
key = 4;
pRet = taosbsearch((void *)&key, (void *)array2, 2, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 6);
key = 6;
pRet = taosbsearch((void *)&key, (void *)array2, 2, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(pRet, nullptr);
key = 7;
pRet = taosbsearch((void *)&key, (void *)array2, 2, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(pRet, nullptr);
// 3 element
int array3[3] = {3, 6, 8};
key = 1;
pRet = taosbsearch((void *)&key, (void *)array3, 3, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 3);
key = 3;
pRet = taosbsearch((void *)&key, (void *)array3, 3, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 6);
key = 4;
pRet = taosbsearch((void *)&key, (void *)array3, 3, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 6);
key = 6;
pRet = taosbsearch((void *)&key, (void *)array3, 3, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 8);
key = 7;
pRet = taosbsearch((void *)&key, (void *)array3, 3, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 8);
key = 8;
pRet = taosbsearch((void *)&key, (void *)array3, 3, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(pRet, nullptr);
key = 9;
pRet = taosbsearch((void *)&key, (void *)array3, 3, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(pRet, nullptr);
// 4 element
int array4[4] = {3, 6, 8, 11};
key = 1;
pRet = taosbsearch((void *)&key, (void *)array4, 4, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 3);
key = 3;
pRet = taosbsearch((void *)&key, (void *)array4, 4, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 6);
key = 4;
pRet = taosbsearch((void *)&key, (void *)array4, 4, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 6);
key = 6;
pRet = taosbsearch((void *)&key, (void *)array4, 4, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 8);
key = 7;
pRet = taosbsearch((void *)&key, (void *)array4, 4, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 8);
key = 8;
pRet = taosbsearch((void *)&key, (void *)array4, 4, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 11);
key = 9;
pRet = taosbsearch((void *)&key, (void *)array4, 4, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 11);
key = 11;
pRet = taosbsearch((void *)&key, (void *)array4, 4, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(pRet, nullptr);
key = 13;
pRet = taosbsearch((void *)&key, (void *)array4, 4, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(pRet, nullptr);
// 5 element
int array5[5] = {3, 6, 8, 11, 15};
key = 1;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 3);
key = 3;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 6);
key = 4;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 6);
key = 6;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 8);
key = 7;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 8);
key = 8;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 11);
key = 9;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 11);
key = 11;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 15);
key = 13;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(*(int *)pRet, 15);
key = 15;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(pRet, nullptr);
key = 17;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_GT);
ASSERT_EQ(pRet, nullptr);
}
TEST(testCase, taosbsearch_less_or_equal) { TEST(testCase, taosbsearch_less_or_equal) {
// For equal test // For equal test
int key = 3; int key = 3;
...@@ -411,4 +575,168 @@ TEST(testCase, taosbsearch_less_or_equal) { ...@@ -411,4 +575,168 @@ TEST(testCase, taosbsearch_less_or_equal) {
key = 17; key = 17;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_LE); pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_LE);
ASSERT_EQ(*(int *)pRet, 15); ASSERT_EQ(*(int *)pRet, 15);
}
TEST(testCase, taosbsearch_less) {
// For equal test
int key = 3;
void *pRet = NULL;
pRet = taosbsearch((void *)&key, NULL, 0, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(pRet, nullptr);
// 1 element
int array1[1] = {5};
key = 1;
pRet = taosbsearch((void *)&key, (void *)array1, 1, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(pRet, nullptr);
key = 6;
pRet = taosbsearch((void *)&key, (void *)array1, 1, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 5);
key = 5;
pRet = taosbsearch((void *)&key, (void *)array1, 1, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(pRet, nullptr);
// 2 element
int array2[2] = {3, 6};
key = 1;
pRet = taosbsearch((void *)&key, (void *)array2, 2, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(pRet, nullptr);
key = 3;
pRet = taosbsearch((void *)&key, (void *)array2, 2, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(pRet, nullptr);
key = 4;
pRet = taosbsearch((void *)&key, (void *)array2, 2, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 3);
key = 6;
pRet = taosbsearch((void *)&key, (void *)array2, 2, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 3);
key = 7;
pRet = taosbsearch((void *)&key, (void *)array2, 2, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 6);
// 3 element
int array3[3] = {3, 6, 8};
key = 1;
pRet = taosbsearch((void *)&key, (void *)array3, 3, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(pRet, nullptr);
key = 3;
pRet = taosbsearch((void *)&key, (void *)array3, 3, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(pRet, nullptr);
key = 4;
pRet = taosbsearch((void *)&key, (void *)array3, 3, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 3);
key = 6;
pRet = taosbsearch((void *)&key, (void *)array3, 3, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 3);
key = 7;
pRet = taosbsearch((void *)&key, (void *)array3, 3, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 6);
key = 8;
pRet = taosbsearch((void *)&key, (void *)array3, 3, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 6);
key = 9;
pRet = taosbsearch((void *)&key, (void *)array3, 3, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 8);
// 4 element
int array4[4] = {3, 6, 8, 11};
key = 1;
pRet = taosbsearch((void *)&key, (void *)array4, 4, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(pRet, nullptr);
key = 3;
pRet = taosbsearch((void *)&key, (void *)array4, 4, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(pRet, nullptr);
key = 4;
pRet = taosbsearch((void *)&key, (void *)array4, 4, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 3);
key = 6;
pRet = taosbsearch((void *)&key, (void *)array4, 4, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 3);
key = 7;
pRet = taosbsearch((void *)&key, (void *)array4, 4, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 6);
key = 8;
pRet = taosbsearch((void *)&key, (void *)array4, 4, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 6);
key = 9;
pRet = taosbsearch((void *)&key, (void *)array4, 4, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 8);
key = 11;
pRet = taosbsearch((void *)&key, (void *)array4, 4, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 8);
key = 13;
pRet = taosbsearch((void *)&key, (void *)array4, 4, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 11);
// 5 element
int array5[5] = {3, 6, 8, 11, 15};
key = 1;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(pRet, nullptr);
key = 3;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(pRet, nullptr);
key = 4;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 3);
key = 6;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 3);
key = 7;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 6);
key = 8;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 6);
key = 9;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 8);
key = 11;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 8);
key = 13;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 11);
key = 15;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 11);
key = 17;
pRet = taosbsearch((void *)&key, (void *)array5, 5, sizeof(int), compareFunc, TD_LT);
ASSERT_EQ(*(int *)pRet, 15);
} }
\ No newline at end of file
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或