Merge branch 'master' into UUID-clone

Author: lance6716

.github/workflows/ci.yml

@@ -0,0 +1,45 @@
+name: runTestsAndLinters
+on: [push, pull_request]
+
+jobs:
+  test:
+    strategy:
+      matrix:
+        go: [ 1.16, 1.15 ]
+    name: Tests Go ${{ matrix.go }}
+    runs-on: ubuntu-18.04
+
+    steps:
+      - name: Setup MySQL
+        run: |
+          echo -n "mysql -V: " ; mysql -V
+          echo -n "mysqldump -V: " ; mysqldump -V
+
+          echo -e '[mysqld]\nserver-id=1\nlog-bin=mysql\nbinlog-format=row\ngtid-mode=ON\nenforce_gtid_consistency=ON\n' | sudo tee /etc/mysql/conf.d/replication.cnf
+          sudo service mysql start
+          sudo mysql -h 127.0.0.1 -uroot -proot -e "use mysql; update user set authentication_string=PASSWORD('') where User='root'; update user set plugin='mysql_native_password'; FLUSH PRIVILEGES;"
+          # create ssl/rsa files for mysql ssl support
+          sudo mysql_ssl_rsa_setup --uid=mysql
+          mysql -e "CREATE DATABASE IF NOT EXISTS test;" -uroot
+          mysql -e "SHOW VARIABLES LIKE 'log_bin'" -uroot
+      - name: Prepare for Go
+        run: |
+          sudo apt-get install -y make gcc
+      - name: Install Go
+        uses: actions/setup-go@v2
+        with:
+          go-version: ${{ matrix.go }}
+      - name: Checkout code
+        uses: actions/checkout@v1
+      - name: Run tests
+        run: go test ./...
+
+  golangci:
+    name: golangci
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: golangci-lint
+        uses: golangci/golangci-lint-action@v2
+        with:
+          version: latest

.golangci.yml

@@ -0,0 +1,28 @@
+linters:
+  disable-all: true
+  enable:
+    # All code is ready for:
+    - deadcode
+    - errcheck
+    - staticcheck
+    - structcheck
+    - typecheck
+    - unused
+    - varcheck
+    - misspell
+    - nolintlint
+    - goimports
+    - nakedret
+    - unconvert
+    - whitespace
+    # ToDo:
+    #- gosimple
+    #- govet
+    #- ineffassign
+    #- gocritic
+    #- golint
+linters-settings:
+  nolintlint:
+    allow-unused: false
+    allow-leading-space: false
+    require-specific: true

.travis.yml

@@ -0,0 +1,19 @@
+### Tag v1.1.1 (2021.04.08)
+* Restructured dump/ tests. [#563](https://github.com/go-mysql-org/go-mysql/pull/563) ([atercattus](https://github.com/atercattus))
+* Replace magic numbers in canal/canal_test.go by constants. [#562](https://github.com/go-mysql-org/go-mysql/pull/562) ([atercattus](https://github.com/atercattus))
+* Fix parsing GTIDs from mysqlpdump. [#561](https://github.com/go-mysql-org/go-mysql/pull/561) ([dobegor](https://github.com/dobegor))
+* Streaming of SELECT responses. `client/Conn.ExecuteSelectStreaming()` added. [#560](https://github.com/go-mysql-org/go-mysql/pull/560) ([atercattus](https://github.com/atercattus))
+* Migation from travis.ci to github actions. [#559](https://github.com/go-mysql-org/go-mysql/pull/559) ([atercattus](https://github.com/atercattus))
+* Output sorted mysql gtid. [#500](https://github.com/go-mysql-org/go-mysql/pull/500) ([zr-hebo](https://github.com/zr-hebo))
+* Add skipped columns information. [#505](https://github.com/go-mysql-org/go-mysql/pull/505) ([laskoviymishka](https://github.com/laskoviymishka))
+* Feat: support disable retry sync for canal. [#507](https://github.com/go-mysql-org/go-mysql/pull/507) ([everpcpc](https://github.com/everpcpc))
+* Update README.md. [#511](https://github.com/go-mysql-org/go-mysql/pull/511) ([TennyZhuang](https://github.com/TennyZhuang))
+* Add function to extend replication options. [#508](https://github.com/go-mysql-org/go-mysql/pull/508) ([wefen](https://github.com/wefen))
+
+### Tag v1.1.0 (2020.07.17)
+* Update .travis.yml (go 1.14 and tip). [#510](https://github.com/go-mysql-org/go-mysql/pull/510) ([atercattus](https://github.com/atercattus))
+* Update README.md. [#509](https://github.com/go-mysql-org/go-mysql/pull/509) ([atercattus](https://github.com/atercattus))
+* A lot of memory allocation optimizations. Changed public API for `mysql/Resultset` type. [#466](https://github.com/go-mysql-org/go-mysql/pull/466) ([atercattus](https://github.com/atercattus))
+
+### Tag v1.0.0 (2020.07.17)
+Add SemVer

CHANGELOG.md

@@ -2,16 +2,29 @@
 
 A pure go library to handle MySQL network protocol and replication.
 
-## Call for Committer/Maintainer
+![semver](https://img.shields.io/github/v/tag/go-mysql-org/go-mysql)
+![example workflow](https://github.com/go-mysql-org/go-mysql/actions/workflows/ci.yml/badge.svg)
+![gomod version](https://img.shields.io/github/go-mod/go-version/go-mysql-org/go-mysql/master)
 
-Sorry that I have no enough time to maintain this project wholly, if you like this project and want to help me improve it continuously, please contact me through email ([email protected]).
+## How to migrate to this repo
+To change the used package in your repo it's enough to add this `replace` directive to your `go.mod`:
+```
+replace github.com/siddontang/go-mysql => github.com/go-mysql-org/go-mysql v1.1.1
+```
 
-Requirement: In the email, you should list somethings(including but not limited to below) to make me believe we can work together.
+v.1.1.1 - is the last tag in repo, feel free to choose what you want.
 
-+ Your GitHub ID
-+ The contributions to go-mysql before, including PRs or Issues. 
-+ The reason why you can improve go-mysql.
- 
+## Changelog
+This repo uses [Changelog](CHANGELOG.md).
+
+---
+# Content
+* [Slave replication](#replication)
+* [Incremental dumping](#canal)
+* [Client](#client)
+* [Fake server](#server)
+* [Failover](#failover)
+* [database/sql like driver](#driver)
 
 ## Replication
 
@@ -197,6 +210,27 @@ Tested MySQL versions for the client include:
 - 5.7.x
 - 8.0.x
 
+### Example for SELECT streaming (v.1.1.1)
+You can use also streaming for large SELECT responses.
+The callback function will be called for every result row without storing the whole resultset in memory.
+`result.Fields` will be filled before the first callback call.
+
+```go
+// ...
+var result mysql.Result
+err := conn.ExecuteSelectStreaming(`select id, name from table LIMIT 100500`, &result, func(row []mysql.FieldValue) error {
+    for idx, val := range row {
+    	field := result.Fields[idx]
+    	// You must not save FieldValue.AsString() value after this callback is done.
+    	// Copy it if you need.
+    	// ...
+    }
+    return false, nil
+})
+
+// ...
+```
+
 ## Server
 
 Server package supplies a framework to implement a simple MySQL server which can handle the packets from the MySQL client. 

README.md

@@ -249,7 +249,7 @@ func (c *Canal) Close() {
 	c.conn = nil
 	c.connLock.Unlock()
 
-	c.eventHandler.OnPosSynced(c.master.Position(), c.master.GTIDSet(), true)
+	_ = c.eventHandler.OnPosSynced(c.master.Position(), c.master.GTIDSet(), true)
 }
 
 func (c *Canal) WaitDumpDone() <-chan struct{} {

canal/canal.go

@@ -26,6 +26,16 @@ type canalTestSuite struct {
 
 var _ = Suite(&canalTestSuite{})
 
+const (
+	miA = 0
+	miB = -1
+	miC = 1
+
+	umiA = 0
+	umiB = 1
+	umiC = 16777215
+)
+
 func (s *canalTestSuite) SetUpSuite(c *C) {
 	cfg := NewDefaultConfig()
 	cfg.Addr = fmt.Sprintf("%s:3306", *testHost)
@@ -62,7 +72,11 @@ func (s *canalTestSuite) SetUpSuite(c *C) {
 	s.execute(c, sql)
 
 	s.execute(c, "DELETE FROM test.canal_test")
-	s.execute(c, "INSERT INTO test.canal_test (content, name, mi, umi) VALUES (?, ?, ?, ?), (?, ?, ?, ?), (?, ?, ?, ?)", "1", "a", 0, 0, `\0\ndsfasdf`, "b", 1, 16777215, "", "c", -1, 1)
+	s.execute(c, "INSERT INTO test.canal_test (content, name, mi, umi) VALUES (?, ?, ?, ?), (?, ?, ?, ?), (?, ?, ?, ?)",
+		"1", "a", miA, umiA,
+		`\0\ndsfasdf`, "b", miC, umiC,
+		"", "c", miB, umiB,
+	)
 
 	s.execute(c, "SET GLOBAL binlog_format = 'ROW'")
 
@@ -99,7 +113,7 @@ type testEventHandler struct {
 func (h *testEventHandler) OnRow(e *RowsEvent) error {
 	log.Infof("OnRow %s %v\n", e.Action, e.Rows)
 	umi, ok := e.Rows[0][4].(uint32) // 4th col is umi. mysqldump gives uint64 instead of uint32
-	if ok && (umi != 0 && umi != 1 && umi != 16777215) {
+	if ok && (umi != umiA && umi != umiB && umi != umiC) {
 		return fmt.Errorf("invalid unsigned medium int %d", umi)
 	}
 	return nil
@@ -119,7 +133,11 @@ func (s *canalTestSuite) TestCanal(c *C) {
 	for i := 1; i < 10; i++ {
 		s.execute(c, "INSERT INTO test.canal_test (name) VALUES (?)", fmt.Sprintf("%d", i))
 	}
-	s.execute(c, "INSERT INTO test.canal_test (mi,umi) VALUES (?,?), (?,?), (?,?)", 0, 0, -1, 16777215, 1, 1)
+	s.execute(c, "INSERT INTO test.canal_test (mi,umi) VALUES (?,?), (?,?), (?,?)",
+		miA, umiA,
+		miC, umiC,
+		miB, umiB,
+	)
 	s.execute(c, "ALTER TABLE test.canal_test ADD `age` INT(5) NOT NULL AFTER `name`")
 	s.execute(c, "INSERT INTO test.canal_test (name,age) VALUES (?,?)", "d", "18")
 
@@ -283,8 +301,8 @@ func TestDropTableExp(t *testing.T) {
 		}
 	}
 }
-func TestWithoutSchemeExp(t *testing.T) {
 
+func TestWithoutSchemeExp(t *testing.T) {
 	cases := []replication.QueryEvent{
 		replication.QueryEvent{
 			Schema: []byte("test"),

canal/canal_test.go

@@ -227,7 +227,7 @@ func parseStmt(stmt ast.StmtNode) (ns []*node) {
 		}
 		ns = []*node{n}
 	}
-	return
+	return ns
 }
 
 func (c *Canal) updateTable(db, table string) (err error) {

canal/sync.go

@@ -46,7 +46,7 @@ func (c *Conn) readInitialHandshake() error {
 	pos := 1 + bytes.IndexByte(data[1:], 0x00) + 1
 
 	// connection id length is 4
-	c.connectionID = uint32(binary.LittleEndian.Uint32(data[pos : pos+4]))
+	c.connectionID = binary.LittleEndian.Uint32(data[pos : pos+4])
 	pos += 4
 
 	c.salt = []byte{}
@@ -106,7 +106,7 @@ func (c *Conn) readInitialHandshake() error {
 // generate auth response data according to auth plugin
 //
 // NOTE: the returned boolean value indicates whether to add a \NUL to the end of data.
-//       it is quite tricky because MySQl server expects different formats of responses in different auth situations.
+//       it is quite tricky because MySQL server expects different formats of responses in different auth situations.
 //       here the \NUL needs to be added when sending back the empty password or cleartext password in 'sha256_password'
 //       authentication.
 func (c *Conn) genAuthResponse(authData []byte) ([]byte, bool, error) {
@@ -199,7 +199,7 @@ func (c *Conn) writeAuthHandshake() error {
 
 	// Charset [1 byte]
 	// use default collation id 33 here, is utf-8
-	data[12] = byte(DEFAULT_COLLATION_ID)
+	data[12] = DEFAULT_COLLATION_ID
 
 	// SSL Connection Request Packet
 	// http://dev.mysql.com/doc/internals/en/connection-phase-packets.html#packet-Protocol::SSLRequest

client/auth.go

@@ -289,7 +289,6 @@ func (s *clientTestSuite) TestStmt_Select(c *C) {
 
 	e, _ = result.GetStringByName(0, "e")
 	c.Assert(e, Equals, "test1")
-
 }
 
 func (s *clientTestSuite) TestStmt_NULL(c *C) {
@@ -309,11 +308,13 @@ func (s *clientTestSuite) TestStmt_NULL(c *C) {
 
 	str = `select * from mixer_test_stmt where id = ?`
 	stmt, err = s.c.Prepare(str)
+	c.Assert(err, IsNil)
+
 	defer stmt.Close()
 
+	result, err = stmt.Execute(2)
 	c.Assert(err, IsNil)
 
-	result, err = stmt.Execute(2)
 	b, err := result.IsNullByName(0, "id")
 	c.Assert(err, IsNil)
 	c.Assert(b, Equals, false)

client/client_test.go

@@ -33,6 +33,9 @@ type Conn struct {
 	connectionID uint32
 }
 
+// This function will be called for every row in resultset from ExecuteSelectStreaming.
+type SelectPerRowCallback func(row []FieldValue) error
+
 func getNetProto(addr string) string {
 	proto := "tcp"
 	if strings.Contains(addr, "/") {
@@ -165,6 +168,28 @@ func (c *Conn) Execute(command string, args ...interface{}) (*Result, error) {
 	}
 }
 
+// ExecuteSelectStreaming will call perRowCallback for every row in resultset
+//   WITHOUT saving any row data to Result.{Values/RawPkg/RowDatas} fields.
+//
+// ExecuteSelectStreaming should be used only for SELECT queries with a large response resultset for memory preserving.
+//
+// Example:
+//
+// 		var result mysql.Result
+// 		conn.ExecuteSelectStreaming(`SELECT ... LIMIT 100500`, &result, func(row []mysql.FieldValue) error {
+//   		// Use the row as you want.
+//   		// You must not save FieldValue.AsString() value after this callback is done. Copy it if you need.
+//   		return nil
+// 		})
+//
+func (c *Conn) ExecuteSelectStreaming(command string, result *Result, perRowCallback SelectPerRowCallback) error {
+	if err := c.writeCommandStr(COM_QUERY, command); err != nil {
+		return errors.Trace(err)
+	}
+
+	return c.readResultStreaming(false, result, perRowCallback)
+}
+
 func (c *Conn) Begin() error {
 	_, err := c.exec("BEGIN")
 	return errors.Trace(err)