Merge remote-tracking branch 'origin/3.0' into fix/valgrind

cmake/cmake.options

@@ -163,18 +163,6 @@ option(
     ON 
 )
 
-option(
-    BUILD_WITH_CRAFT
-    "If build with canonical-raft"
-    OFF
-)
-
-option(
-    BUILD_WITH_TRAFT
-    "If build with traft"
-    OFF
-)
-
 IF(${TD_LINUX} MATCHES TRUE)
 
 option(

cmake/craft_CMakeLists.txt.in

-
-# canonical-raft
-ExternalProject_Add(craft
-        GIT_REPOSITORY https://github.com/canonical/raft.git
-        GIT_TAG v0.11.2
-        SOURCE_DIR "${TD_CONTRIB_DIR}/craft"
-        BINARY_DIR "${TD_CONTRIB_DIR}/craft"
-        #BUILD_IN_SOURCE TRUE
-        # https://answers.ros.org/question/333125/how-to-include-external-automakeautoconf-projects-into-ament_cmake/
-        CONFIGURE_COMMAND COMMAND autoreconf -i COMMAND ./configure --enable-example
-        BUILD_COMMAND "$(MAKE)"
-        INSTALL_COMMAND ""
-        TEST_COMMAND ""
-)

cmake/traft_CMakeLists.txt.in

-
-# traft
-ExternalProject_Add(traft
-        GIT_REPOSITORY https://github.com/taosdata/traft.git
-        GIT_TAG for_3.0
-        SOURCE_DIR "${TD_CONTRIB_DIR}/traft"
-        BINARY_DIR "${TD_CONTRIB_DIR}/traft"
-        #BUILD_IN_SOURCE TRUE
-        # https://answers.ros.org/question/333125/how-to-include-external-automakeautoconf-projects-into-ament_cmake/
-        CONFIGURE_COMMAND COMMAND autoreconf -i COMMAND ./configure
-        BUILD_COMMAND "$(MAKE)"
-        INSTALL_COMMAND ""
-        TEST_COMMAND ""
-)

docs/examples/go/connect/afconn/main.go

@@ -3,7 +3,7 @@ package main
 import (
 	"fmt"
 
-	"github.com/taosdata/driver-go/v2/af"
+	"github.com/taosdata/driver-go/v3/af"
 )
 
 func main() {

docs/examples/go/connect/cgoexample/main.go

@@ -4,7 +4,7 @@ import (
 	"database/sql"
 	"fmt"
 
-	_ "github.com/taosdata/driver-go/v2/taosSql"
+	_ "github.com/taosdata/driver-go/v3/taosSql"
 )
 
 func main() {

docs/examples/go/connect/restexample/main.go

@@ -4,7 +4,7 @@ import (
 	"database/sql"
 	"fmt"
 
-	_ "github.com/taosdata/driver-go/v2/taosRestful"
+	_ "github.com/taosdata/driver-go/v3/taosRestful"
 )
 
 func main() {

docs/examples/go/connect/wrapper/main.go

@@ -3,7 +3,7 @@ package main
 import (
 	"fmt"
 
-	"github.com/taosdata/driver-go/v2/wrapper"
+	"github.com/taosdata/driver-go/v3/wrapper"
 )
 
 func main() {

docs/examples/go/go.mod

@@ -2,5 +2,5 @@ module goexample
 
 go 1.17
 
-require github.com/taosdata/driver-go/v2 develop
+require github.com/taosdata/driver-go/v3 3.0
 

docs/examples/go/insert/json/main.go

@@ -3,7 +3,7 @@ package main
 import (
 	"fmt"
 
-	"github.com/taosdata/driver-go/v2/af"
+	"github.com/taosdata/driver-go/v3/af"
 )
 
 func prepareDatabase(conn *af.Connector) {

docs/examples/go/insert/line/main.go

@@ -3,7 +3,7 @@ package main
 import (
 	"fmt"
 
-	"github.com/taosdata/driver-go/v2/af"
+	"github.com/taosdata/driver-go/v3/af"
 )
 
 func prepareDatabase(conn *af.Connector) {

docs/examples/go/insert/sql/main.go

@@ -4,7 +4,7 @@ import (
 	"database/sql"
 	"fmt"
 
-	_ "github.com/taosdata/driver-go/v2/taosRestful"
+	_ "github.com/taosdata/driver-go/v3/taosRestful"
 )
 
 func createStable(taos *sql.DB) {

docs/examples/go/insert/stmt/main.go

@@ -4,9 +4,9 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/taosdata/driver-go/v2/af"
-	"github.com/taosdata/driver-go/v2/af/param"
-	"github.com/taosdata/driver-go/v2/common"
+	"github.com/taosdata/driver-go/v3/af"
+	"github.com/taosdata/driver-go/v3/af/param"
+	"github.com/taosdata/driver-go/v3/common"
 )
 
 func checkErr(err error, prompt string) {

docs/examples/go/insert/telnet/main.go

@@ -3,7 +3,7 @@ package main
 import (
 	"fmt"
 
-	"github.com/taosdata/driver-go/v2/af"
+	"github.com/taosdata/driver-go/v3/af"
 )
 
 func prepareDatabase(conn *af.Connector) {

docs/examples/go/query/sync/main.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 	"time"
 
-	_ "github.com/taosdata/driver-go/v2/taosRestful"
+	_ "github.com/taosdata/driver-go/v3/taosRestful"
 )
 
 func main() {

docs/examples/go/sub/main.go

 package main
 
 import (
-	"database/sql/driver"
+	"context"
+	"encoding/json"
 	"fmt"
-	"io"
-	"os"
+	"strconv"
 	"time"
 
-	taos "github.com/taosdata/driver-go/v2/af"
+	"github.com/taosdata/driver-go/v3/af"
+	"github.com/taosdata/driver-go/v3/af/tmq"
+	"github.com/taosdata/driver-go/v3/common"
+	"github.com/taosdata/driver-go/v3/errors"
+	"github.com/taosdata/driver-go/v3/wrapper"
 )
 
 func main() {
-	db, err := taos.Open("", "", "", "log", 0)
+	db, err := af.Open("", "root", "taosdata", "", 0)
 	if err != nil {
-		fmt.Fprintln(os.Stderr, err)
-		os.Exit(1)
+		panic(err)
 	}
 	defer db.Close()
-	topic, err := db.Subscribe(false, "taoslogtail", "select ts, level, ipaddr, content from log", time.Second)
+	_, err = db.Exec("create database if not exists example_tmq")
 	if err != nil {
-		fmt.Fprintln(os.Stderr, err)
-		os.Exit(2)
+		panic(err)
+	}
+	_, err = db.Exec("create topic if not exists example_tmq_topic with meta as DATABASE example_tmq")
+	if err != nil {
+		panic(err)
+	}
+	config := tmq.NewConfig()
+	defer config.Destroy()
+	err = config.SetGroupID("test")
+	if err != nil {
+		panic(err)
+	}
+	err = config.SetAutoOffsetReset("earliest")
+	if err != nil {
+		panic(err)
+	}
+	err = config.SetConnectIP("127.0.0.1")
+	if err != nil {
+		panic(err)
+	}
+	err = config.SetConnectUser("root")
+	if err != nil {
+		panic(err)
+	}
+	err = config.SetConnectPass("taosdata")
+	if err != nil {
+		panic(err)
+	}
+	err = config.SetConnectPort("6030")
+	if err != nil {
+		panic(err)
+	}
+	err = config.SetMsgWithTableName(true)
+	if err != nil {
+		panic(err)
+	}
+	err = config.EnableHeartBeat()
+	if err != nil {
+		panic(err)
+	}
+	err = config.EnableAutoCommit(func(result *wrapper.TMQCommitCallbackResult) {
+		if result.ErrCode != 0 {
+			errStr := wrapper.TMQErr2Str(result.ErrCode)
+			err := errors.NewError(int(result.ErrCode), errStr)
+			panic(err)
+		}
+	})
+	if err != nil {
+		panic(err)
+	}
+	consumer, err := tmq.NewConsumer(config)
+	if err != nil {
+		panic(err)
+	}
+	err = consumer.Subscribe([]string{"example_tmq_topic"})
+	if err != nil {
+		panic(err)
+	}
+	_, err = db.Exec("create table example_tmq.t1 (ts timestamp,v int)")
+	if err != nil {
+		panic(err)
 	}
-	defer topic.Unsubscribe(true)
 	for {
-		func() {
-			rows, err := topic.Consume()
-			defer func() { rows.Close(); time.Sleep(time.Second) }()
-			if err != nil {
-				fmt.Println(err)
-				os.Exit(3)
-			}
-			for {
-				values := make([]driver.Value, 4)
-				err := rows.Next(values)
-				if err == io.EOF {
-					break
-				} else if err != nil {
-					fmt.Fprintln(os.Stderr, err)
-					os.Exit(4)
-				}
-				ts := values[0].(time.Time)
-				level := values[1].(int8)
-				ipaddr := values[2].(string)
-				content := values[3].(string)
-				fmt.Printf("%s %d %s %s\n", ts.Format(time.StampMilli), level, ipaddr, content)
-			}
-		}()
+		result, err := consumer.Poll(time.Second)
+		if err != nil {
+			panic(err)
+		}
+		if result.Type != common.TMQ_RES_TABLE_META {
+			panic("want message type 2 got " + strconv.Itoa(int(result.Type)))
+		}
+		data, _ := json.Marshal(result.Meta)
+		fmt.Println(string(data))
+		consumer.Commit(context.Background(), result.Message)
+		consumer.FreeMessage(result.Message)
+		break
+	}
+	_, err = db.Exec("insert into example_tmq.t1 values(now,1)")
+	if err != nil {
+		panic(err)
 	}
+	for {
+		result, err := consumer.Poll(time.Second)
+		if err != nil {
+			panic(err)
+		}
+		if result.Type != common.TMQ_RES_DATA {
+			panic("want message type 1 got " + strconv.Itoa(int(result.Type)))
+		}
+		data, _ := json.Marshal(result.Data)
+		fmt.Println(string(data))
+		consumer.Commit(context.Background(), result.Message)
+		consumer.FreeMessage(result.Message)
+		break
+	}
+	consumer.Close()
 }
-
-// 未完成

docs/zh/14-reference/03-connector/go.mdx

@@ -65,7 +65,7 @@ REST 连接支持所有能运行 Go 的平台。
 
 ### 使用 go get 安装
 
-`go get -u github.com/taosdata/driver-go/v2@develop`
+`go get -u github.com/taosdata/driver-go/v3@latest`
 
 ### 使用 go mod 管理
 
@@ -80,7 +80,7 @@ REST 连接支持所有能运行 Go 的平台。
   ```go
   import (
     "database/sql"
-    _ "github.com/taosdata/driver-go/v2/taosSql"
+    _ "github.com/taosdata/driver-go/v3/taosSql"
   )
   ```
 
@@ -132,7 +132,7 @@ import (
     "database/sql"
     "fmt"
 
-    _ "github.com/taosdata/driver-go/v2/taosSql"
+    _ "github.com/taosdata/driver-go/v3/taosSql"
 )
 
 func main() {
@@ -164,7 +164,7 @@ import (
     "database/sql"
     "fmt"
 
-    _ "github.com/taosdata/driver-go/v2/taosRestful"
+    _ "github.com/taosdata/driver-go/v3/taosRestful"
 )
 
 func main() {
@@ -205,14 +205,14 @@ func main() {
 
 ### 更多示例程序
 
-* [示例程序](https://github.com/taosdata/TDengine/tree/develop/examples/go)
+* [示例程序](https://github.com/taosdata/driver-go/tree/3.0/examples)
 * [视频教程](https://www.taosdata.com/blog/2020/11/11/1951.html)。
 
 ## 使用限制
 
 由于 REST 接口无状态所以 `use db` 语法不会生效，需要将 db 名称放到 SQL 语句中，如：`create table if not exists tb1 (ts timestamp, a int)`改为`create table if not exists test.tb1 (ts timestamp, a int)`否则将报错`[0x217] Database not specified or available`。
 
-也可以将 db 名称放到 DSN 中，将 `root:taosdata@http(localhost:6041)/` 改为 `root:taosdata@http(localhost:6041)/test`，此方法在 TDengine 2.4.0.5 版本的 taosAdapter 开始支持。当指定的 db 不存在时执行 `create database` 语句不会报错，而执行针对该 db 的其他查询或写入操作会报错。
+也可以将 db 名称放到 DSN 中，将 `root:taosdata@http(localhost:6041)/` 改为 `root:taosdata@http(localhost:6041)/test`。当指定的 db 不存在时执行 `create database` 语句不会报错，而执行针对该 db 的其他查询或写入操作会报错。
 
 完整示例如下：
 
@@ -224,7 +224,7 @@ import (
     "fmt"
     "time"
 
-    _ "github.com/taosdata/driver-go/v2/taosRestful"
+    _ "github.com/taosdata/driver-go/v3/taosRestful"
 )
 
 func main() {
@@ -266,35 +266,27 @@ func main() {
 
 ## 常见问题
 
-1. 无法找到包 `github.com/taosdata/driver-go/v2/taosRestful`
-
-  将 `go.mod` 中 require 块对`github.com/taosdata/driver-go/v2`的引用改为`github.com/taosdata/driver-go/v2 develop`，之后执行 `go mod tidy`。
-
-2. database/sql 中 stmt（参数绑定）相关接口崩溃
+1. database/sql 中 stmt（参数绑定）相关接口崩溃
 
   REST 不支持参数绑定相关接口，建议使用`db.Exec`和`db.Query`。
 
-3. 使用 `use db` 语句后执行其他语句报错 `[0x217] Database not specified or available`
+2. 使用 `use db` 语句后执行其他语句报错 `[0x217] Database not specified or available`
 
   在 REST 接口中 SQL 语句的执行无上下文关联，使用 `use db` 语句不会生效，解决办法见上方使用限制章节。
 
-4. 使用 taosSql 不报错使用 taosRestful 报错 `[0x217] Database not specified or available`
+3. 使用 taosSql 不报错使用 taosRestful 报错 `[0x217] Database not specified or available`
 
   因为 REST 接口无状态，使用 `use db` 语句不会生效，解决办法见上方使用限制章节。
 
-5. 升级 `github.com/taosdata/driver-go/v2/taosRestful`
-
-  将 `go.mod` 文件中对 `github.com/taosdata/driver-go/v2` 的引用改为 `github.com/taosdata/driver-go/v2 develop`，之后执行 `go mod tidy`。
-
-6. `readBufferSize` 参数调大后无明显效果
+4. `readBufferSize` 参数调大后无明显效果
 
   `readBufferSize` 调大后会减少获取结果时 `syscall` 的调用。如果查询结果的数据量不大，修改该参数不会带来明显提升，如果该参数修改过大，瓶颈会在解析 JSON 数据。如果需要优化查询速度，需要根据实际情况调整该值来达到查询效果最优。
 
-7. `disableCompression` 参数设置为 `false` 时查询效率降低
+5. `disableCompression` 参数设置为 `false` 时查询效率降低
 
   当 `disableCompression` 参数设置为 `false` 时查询结果会使用 `gzip` 压缩后传输，拿到数据后要先进行 `gzip` 解压。
 
-8. `go get` 命令无法获取包，或者获取包超时
+6. `go get` 命令无法获取包，或者获取包超时
 
   设置 Go 代理 `go env -w GOPROXY=https://goproxy.cn,direct`。
 
@@ -334,17 +326,33 @@ func main() {
 
 #### 订阅
 
-* `func (conn *Connector) Subscribe(restart bool, topic string, sql string, interval time.Duration) (Subscriber, error)`
+* `func NewConsumer(conf *Config) (*Consumer, error)`
+
+创建消费者。
+
+* `func (c *Consumer) Subscribe(topics []string) error`
+
+订阅主题。
 
-  订阅数据。
+* `func (c *Consumer) Poll(timeout time.Duration) (*Result, error)`
 
-* `func (s *taosSubscriber) Consume() (driver.Rows, error)`
+轮询消息。
 
-  消费订阅数据，返回 `database/sql/driver` 包的 `Rows` 结构。
+* `func (c *Consumer) Commit(ctx context.Context, message unsafe.Pointer) error`
 
-* `func (s *taosSubscriber) Unsubscribe(keepProgress bool)`
+提交消息。
 
-  取消订阅数据。
+* `func (c *Consumer) FreeMessage(message unsafe.Pointer)`
+
+释放消息。
+
+* `func (c *Consumer) Unsubscribe() error`
+
+取消订阅。
+
+* `func (c *Consumer) Close() error`
+
+关闭消费者。
 
 #### schemaless
 
@@ -366,10 +374,6 @@ func main() {
 
   参数绑定单行插入。
 
-* `func (conn *Connector) StmtQuery(sql string, params *param.Param) (rows driver.Rows, err error)`
-
-  参数绑定查询，返回 `database/sql/driver` 包的 `Rows` 结构。
-
 * `func (conn *Connector) InsertStmt() *insertstmt.InsertStmt`
 
   初始化参数。
@@ -408,4 +412,4 @@ func main() {
 
 ## API 参考
 
-全部 API 见 [driver-go 文档](https://pkg.go.dev/github.com/taosdata/driver-go/v2)
+全部 API 见 [driver-go 文档](https://pkg.go.dev/github.com/taosdata/driver-go/v3)

source/dnode/mnode/impl/src/mndProfile.c

@@ -687,6 +687,7 @@ static int32_t mndRetrieveConns(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
   int32_t   numOfRows = 0;
   int32_t   cols = 0;
   SConnObj *pConn = NULL;
+  int32_t   keepTime = tsShellActivityTimer * 3;
 
   if (pShow->pIter == NULL) {
     SProfileMgmt *pMgmt = &pMnode->profileMgmt;
@@ -700,6 +701,10 @@ static int32_t mndRetrieveConns(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
       break;
     }
 
+    if ((taosGetTimestampMs() - pConn->lastAccessTimeMs) > (keepTime * 1000)) {
+      continue;
+    }
+
     cols = 0;
 
     SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

source/libs/scalar/src/scalar.c

@@ -140,13 +140,23 @@ int32_t scalarGenerateSetFromList(void **data, void *pNode, uint32_t type) {
       SCL_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
     }
 
+    colDataDestroy(out.columnData);
+    taosMemoryFreeClear(out.columnData);
+    out.columnData = taosMemoryCalloc(1, sizeof(SColumnInfoData));
+
     cell = cell->pNext;
   }
 
   *data = pObj;
+
+  colDataDestroy(out.columnData);
+  taosMemoryFreeClear(out.columnData);
   return TSDB_CODE_SUCCESS;
 
 _return:
+
+  colDataDestroy(out.columnData);
+  taosMemoryFreeClear(out.columnData);
   taosHashCleanup(pObj);
   SCL_RET(code);
 }

source/libs/sync/test/sh/auto_bench.sh

 #!/bin/bash
 
-if [ $# != 5 ] ; then
-  echo "Uasge: $0 instances vgroups replica ctables rows"
+if [ $# != 6 ] ; then
+  echo "Uasge: $0 instances vgroups replica ctables rows weak"
   echo ""
   exit 1
 fi
@@ -11,11 +11,12 @@ vgroups=$2
 replica=$3
 ctables=$4
 rows=$5
+weak=$6
 
 echo "params: instances:${instances}, vgroups:${vgroups}, replica:${replica}, ctables:${ctables}, rows:${rows}"
 
 dt=`date "+%Y-%m-%d-%H-%M-%S"`
-casedir=instances_${instances}_vgroups_${vgroups}_replica_${replica}_ctables_${ctables}_rows_${rows}_${dt}
+casedir=instances_${instances}_vgroups_${vgroups}_replica_${replica}_ctables_${ctables}_rows_${rows}_weak_${weak}_${dt}
 mkdir ${casedir}
 cp ./insert.tpl.json ${casedir}
 cd ${casedir}

source/libs/sync/test/sh/insert.tpl.json

@@ -16,7 +16,10 @@
         {
             "dbinfo": {
                 "name": "db1",
-                "drop": "yes",
+                "drop": "no",
+		"wal_retention_period": -1,
+		"wal_retention_size": -1,
+		"drop": "no",
                 "vgroups": tpl_vgroups_tpl,
 	        "replica": tpl_replica_tpl 	
             },