Merge pull request vitessio#2516 from alainjobart/mysqlconn

Mysqlconn migration

Author: alainjobart

go/mysql/mysql.go

@@ -32,6 +32,10 @@ const (
 func init() {
 	// This needs to be called before threads begin to spawn.
 	C.vt_library_init()
+	sqldb.Register("libmysqlclient", Connect)
+
+	// Comment this out and uncomment call to sqldb.RegisterDefault in
+	// go/mysqlconn/sqldb_conn.go to make it the default.
 	sqldb.RegisterDefault(Connect)
 }
 

go/mysqlconn/client.go

@@ -72,7 +72,7 @@ func Connect(ctx context.Context, params *sqldb.ConnParams) (*Conn, error) {
 		}
 		if err != nil {
 			status <- connectResult{
-				err: sqldb.NewSQLError(CRConnectionError, "", "net.Dial(%v,%v) failed: %v", netProto, addr, err),
+				err: sqldb.NewSQLError(CRConnHostError, "", "net.Dial(%v,%v) failed: %v", netProto, addr, err),
 			}
 			return
 		}

go/mysqlconn/client_test.go

@@ -1,6 +1,7 @@
 package mysqlconn
 
 import (
+	"fmt"
 	"io/ioutil"
 	"net"
 	"os"
@@ -103,7 +104,58 @@ func TestConnectTimeout(t *testing.T) {
 	ctx = context.Background()
 	_, err = Connect(ctx, params)
 	os.Remove(name)
-	assertSQLError(t, err, CRConnectionError, SSSignalException, "connection refused")
+	assertSQLError(t, err, CRConnHostError, SSSignalException, "connection refused")
+}
+
+// testKillWithRealDatabase opens a connection, issues a command that
+// will sleep for a few seconds, waits a bit for MySQL to start
+// executing it, then kills the connection (using another
+// connection). We make sure we get the right error code.
+func testKillWithRealDatabase(t *testing.T, params *sqldb.ConnParams) {
+	ctx := context.Background()
+	conn, err := Connect(ctx, params)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	errChan := make(chan error)
+	go func() {
+		_, err = conn.ExecuteFetch("select sleep(10) from dual", 1000, false)
+		errChan <- err
+		close(errChan)
+	}()
+
+	killConn, err := Connect(ctx, params)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer killConn.Close()
+
+	if _, err := killConn.ExecuteFetch(fmt.Sprintf("kill %v", conn.ConnectionID), 1000, false); err != nil {
+		t.Fatalf("Kill(%v) failed: %v", conn.ConnectionID, err)
+	}
+
+	err = <-errChan
+	assertSQLError(t, err, CRServerLost, SSSignalException, "EOF")
+}
+
+// testDupEntryWithRealDatabase tests a duplicate key is properly raised.
+func testDupEntryWithRealDatabase(t *testing.T, params *sqldb.ConnParams) {
+	ctx := context.Background()
+	conn, err := Connect(ctx, params)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	if _, err := conn.ExecuteFetch("create table dup_entry(id int, name int, primary key(id), unique index(name))", 0, false); err != nil {
+		t.Fatalf("create table failed: %v", err)
+	}
+	if _, err := conn.ExecuteFetch("insert into dup_entry(id, name) values(1, 10)", 0, false); err != nil {
+		t.Fatalf("first insert failed: %v", err)
+	}
+	_, err = conn.ExecuteFetch("insert into dup_entry(id, name) values(2, 10)", 0, false)
+	assertSQLError(t, err, ERDupEntry, SSDupKey, "Duplicate entry")
 }
 
 // TestWithRealDatabase runs a real MySQL database, and runs all kinds
@@ -127,6 +179,16 @@ func TestWithRealDatabase(t *testing.T) {
 		t.Error(err)
 	}
 
+	// Kill tests the query part of the API.
+	t.Run("Kill", func(t *testing.T) {
+		testKillWithRealDatabase(t, &params)
+	})
+
+	// DupEntry tests a duplicate key returns the right error.
+	t.Run("DupEntry", func(t *testing.T) {
+		testDupEntryWithRealDatabase(t, &params)
+	})
+
 	// Queries tests the query part of the API.
 	t.Run("Queries", func(t *testing.T) {
 		testQueriesWithRealDatabase(t, &params)

go/mysqlconn/conn.go

@@ -99,12 +99,12 @@ func (c *Conn) readOnePacket() ([]byte, error) {
 	var header [4]byte
 
 	if _, err := io.ReadFull(c.reader, header[:]); err != nil {
-		return nil, sqldb.NewSQLError(CRServerLost, SSUnknownComError, "io.ReadFull(header size) failed: %v", err)
+		return nil, sqldb.NewSQLError(CRServerLost, SSSignalException, "io.ReadFull(header size) failed: %v", err)
 	}
 
 	sequence := uint8(header[3])
 	if sequence != c.sequence {
-		return nil, sqldb.NewSQLError(CRServerLost, SSUnknownComError, "invalid sequence, expected %v got %v", c.sequence, sequence)
+		return nil, sqldb.NewSQLError(CRServerLost, SSSignalException, "invalid sequence, expected %v got %v", c.sequence, sequence)
 	}
 
 	c.sequence++
@@ -118,7 +118,7 @@ func (c *Conn) readOnePacket() ([]byte, error) {
 
 	data := make([]byte, length)
 	if _, err := io.ReadFull(c.reader, data); err != nil {
-		return nil, sqldb.NewSQLError(CRServerLost, SSUnknownComError, "io.ReadFull(packet body of length %v) failed: %v", length, err)
+		return nil, sqldb.NewSQLError(CRServerLost, SSSignalException, "io.ReadFull(packet body of length %v) failed: %v", length, err)
 	}
 	return data, nil
 }
@@ -191,16 +191,16 @@ func (c *Conn) writePacket(data []byte) error {
 		header[2] = byte(packetLength >> 16)
 		header[3] = c.sequence
 		if n, err := c.writer.Write(header[:]); err != nil {
-			return sqldb.NewSQLError(CRServerLost, SSUnknownComError, "Write(header) failed: %v", err)
+			return sqldb.NewSQLError(CRServerLost, SSSignalException, "Write(header) failed: %v", err)
 		} else if n != 4 {
-			return sqldb.NewSQLError(CRServerLost, SSUnknownComError, "Write(header) returned a short write: %v < 4", n)
+			return sqldb.NewSQLError(CRServerLost, SSSignalException, "Write(header) returned a short write: %v < 4", n)
 		}
 
 		// Write the body.
 		if n, err := c.writer.Write(data[index : index+packetLength]); err != nil {
-			return sqldb.NewSQLError(CRServerLost, SSUnknownComError, "Write(packet) failed: %v", err)
+			return sqldb.NewSQLError(CRServerLost, SSSignalException, "Write(packet) failed: %v", err)
 		} else if n != packetLength {
-			return sqldb.NewSQLError(CRServerLost, SSUnknownComError, "Write(packet) returned a short write: %v < %v", n, packetLength)
+			return sqldb.NewSQLError(CRServerLost, SSSignalException, "Write(packet) returned a short write: %v < %v", n, packetLength)
 		}
 
 		// Update our state.
@@ -216,9 +216,9 @@ func (c *Conn) writePacket(data []byte) error {
 				header[2] = 0
 				header[3] = c.sequence
 				if n, err := c.writer.Write(header[:]); err != nil {
-					return sqldb.NewSQLError(CRServerLost, SSUnknownComError, "Write(empty header) failed: %v", err)
+					return sqldb.NewSQLError(CRServerLost, SSSignalException, "Write(empty header) failed: %v", err)
 				} else if n != 4 {
-					return sqldb.NewSQLError(CRServerLost, SSUnknownComError, "Write(empty header) returned a short write: %v < 4", n)
+					return sqldb.NewSQLError(CRServerLost, SSSignalException, "Write(empty header) returned a short write: %v < 4", n)
 				}
 				c.sequence++
 			}
@@ -230,7 +230,7 @@ func (c *Conn) writePacket(data []byte) error {
 
 func (c *Conn) flush() error {
 	if err := c.writer.Flush(); err != nil {
-		return sqldb.NewSQLError(CRServerLost, SSUnknownComError, "Flush() failed: %v", err)
+		return sqldb.NewSQLError(CRServerLost, SSSignalException, "Flush() failed: %v", err)
 	}
 	return nil
 }

go/mysqlconn/constants.go

@@ -172,6 +172,9 @@ const (
 	// ERUnknownComError is ER_UNKNOWN_COM_ERROR
 	ERUnknownComError = 1047
 
+	// ERDupEntry is ER_DUP_ENTRY
+	ERDupEntry = 1062
+
 	// ERUnknownError is ER_UNKNOWN_ERROR
 	ERUnknownError = 1105
 
@@ -186,6 +189,9 @@ const (
 	// SSSignalException is ER_SIGNAL_EXCEPTION
 	SSSignalException = "HY000"
 
+	// SSDupKey is ER_DUP_KEY
+	SSDupKey = "23000"
+
 	// SSAccessDeniedError is ER_ACCESS_DENIED_ERROR
 	SSAccessDeniedError = "28000"
 
@@ -262,3 +268,12 @@ var CharacterSetMap = map[string]uint8{
 	"cp932":    95,
 	"eucjpms":  97,
 }
+
+// IsNum returns true if a MySQL type is a numeric value.
+// It is the same as IS_NUM defined in mysql.h.
+//
+// FIXME(alainjobart) This needs to use the constants in
+// replication/constants.go, so we are using numerical values here.
+func IsNum(typ uint8) bool {
+	return ((typ <= 9 /* MYSQL_TYPE_INT24 */ && typ != 7 /* MYSQL_TYPE_TIMESTAMP */) || typ == 13 /* MYSQL_TYPE_YEAR */ || typ == 246 /* MYSQL_TYPE_NEWDECIMAL */)
+}

go/mysqlconn/doc.go

@@ -88,4 +88,20 @@ We should add the following protections for the server:
   Should start during initial handshake, maybe have a shorter value during
   handshake.
 
+--
+NUM_FLAG flag:
+
+It is added by the C client library if the field is numerical.
+
+  if (IS_NUM(client_field->type))
+    client_field->flags|= NUM_FLAG;
+
+This is somewhat useless. Also, that flag overlaps with GROUP_FLAG
+(which seems to be used by the server only for temporary tables in
+some cases, so it's not a big deal).
+
+But eventually, we probably want to remove it entirely, as it is not
+transmitted over the wire. For now, we keep it for backward
+compatibility with the C client.
+
 */

go/mysqlconn/query.go

@@ -117,6 +117,16 @@ func (c *Conn) readColumnDefinition(field *querypb.Field, index int) error {
 	// a Field without that data, so we don't return the flags.
 	if field.ColumnLength != 0 || field.Charset != 0 {
 		field.Flags = uint32(flags)
+
+		// FIXME(alainjobart): This is something the MySQL
+		// client library does: If the type is numerical, it
+		// adds a NUM_FLAG to the flags.  We're doing it here
+		// only to be compatible with the C library. Once
+		// we're not using that library any more, we'll remove this.
+		// See doc.go.
+		if IsNum(t) {
+			field.Flags |= uint32(querypb.MySqlFlag_NUM_FLAG)
+		}
 	}
 
 	return nil
@@ -297,6 +307,7 @@ func (c *Conn) ExecuteFetch(query string, maxrows int, wantfields bool) (*sqltyp
 			if !wantfields {
 				result.Fields = nil
 			}
+			result.RowsAffected = uint64(len(result.Rows))
 			return result, nil
 		case ErrPacket:
 			// Error packet.

go/mysqlconn/query_test.go

@@ -77,6 +77,7 @@ func TestQueries(t *testing.T) {
 				sqltypes.NULL,
 			},
 		},
+		RowsAffected: 2,
 	})
 
 	// Typicall Select with TYPE_AND_NAME.
@@ -179,6 +180,7 @@ func TestQueries(t *testing.T) {
 				sqltypes.NULL,
 			},
 		},
+		RowsAffected: 2,
 	})
 
 	// Typicall Select with TYPE_AND_NAME.
@@ -198,6 +200,7 @@ func TestQueries(t *testing.T) {
 				sqltypes.MakeTrusted(querypb.Type_VARCHAR, []byte("nice name")),
 			},
 		},
+		RowsAffected: 2,
 	})
 
 	// Typicall Select with TYPE_ONLY.
@@ -215,6 +218,7 @@ func TestQueries(t *testing.T) {
 				sqltypes.MakeTrusted(querypb.Type_INT64, []byte("20")),
 			},
 		},
+		RowsAffected: 2,
 	})
 
 	// Typicall Select with ALL.
@@ -230,7 +234,10 @@ func TestQueries(t *testing.T) {
 				ColumnLength: 0x80020304,
 				Charset:      0x1234,
 				Decimals:     36,
-				Flags:        16387, // NOT_NULL_FLAG, PRI_KEY_FLAG, PART_KEY_FLAG
+				Flags: uint32(querypb.MySqlFlag_NOT_NULL_FLAG |
+					querypb.MySqlFlag_PRI_KEY_FLAG |
+					querypb.MySqlFlag_PART_KEY_FLAG |
+					querypb.MySqlFlag_NUM_FLAG),
 			},
 		},
 		Rows: [][]sqltypes.Value{
@@ -244,6 +251,7 @@ func TestQueries(t *testing.T) {
 				sqltypes.MakeTrusted(querypb.Type_INT64, []byte("30")),
 			},
 		},
+		RowsAffected: 3,
 	})
 }
 
@@ -313,8 +321,8 @@ func checkQueryInternal(t *testing.T, query string, sConn, cConn *Conn, result *
 		if !reflect.DeepEqual(got, &expected) {
 			for i, f := range got.Fields {
 				if !reflect.DeepEqual(f, expected.Fields[i]) {
-					t.Errorf("Got      field(%v) = %v", i, f)
-					t.Errorf("Expected field(%v) = %v", i, expected.Fields[i])
+					t.Logf("Got      field(%v) = %v", i, f)
+					t.Logf("Expected field(%v) = %v", i, expected.Fields[i])
 				}
 			}
 			t.Fatalf("ExecuteFetch(wantfields=%v) returned:\n%v\nBut was expecting:\n%v", wantfields, got, expected)
@@ -409,12 +417,16 @@ func testQueriesWithRealDatabase(t *testing.T, params *sqldb.ConnParams) {
 	}
 
 	// Try a simple DDL.
-	if _, err := conn.ExecuteFetch("create table a(id int, name varchar(128), primary key(id))", 0, false); err != nil {
+	result, err := conn.ExecuteFetch("create table a(id int, name varchar(128), primary key(id))", 0, false)
+	if err != nil {
 		t.Fatalf("create table failed: %v", err)
 	}
+	if result.RowsAffected != 0 {
+		t.Errorf("create table returned RowsAffected %v, was expecting 0", result.RowsAffected)
+	}
 
 	// Try a simple insert.
-	result, err := conn.ExecuteFetch("insert into a(id, name) values(10, 'nice name')", 1000, true)
+	result, err = conn.ExecuteFetch("insert into a(id, name) values(10, 'nice name')", 1000, true)
 	if err != nil {
 		t.Fatalf("insert failed: %v", err)
 	}
@@ -440,7 +452,8 @@ func testQueriesWithRealDatabase(t *testing.T, params *sqldb.ConnParams) {
 				Charset:      CharacterSetBinary,
 				Flags: uint32(querypb.MySqlFlag_NOT_NULL_FLAG |
 					querypb.MySqlFlag_PRI_KEY_FLAG |
-					querypb.MySqlFlag_PART_KEY_FLAG),
+					querypb.MySqlFlag_PART_KEY_FLAG |
+					querypb.MySqlFlag_NUM_FLAG),
 			},
 			{
 				Name:         "name",
@@ -459,6 +472,7 @@ func testQueriesWithRealDatabase(t *testing.T, params *sqldb.ConnParams) {
 				sqltypes.MakeTrusted(querypb.Type_VARCHAR, []byte("nice name")),
 			},
 		},
+		RowsAffected: 1,
 	}
 	if !reflect.DeepEqual(result, expectedResult) {
 		// MySQL 5.7 is adding the NO_DEFAULT_VALUE_FLAG to Flags.
@@ -470,10 +484,13 @@ func testQueriesWithRealDatabase(t *testing.T, params *sqldb.ConnParams) {
 
 	// Insert a few rows.
 	for i := 0; i < 100; i++ {
-		_, err := conn.ExecuteFetch(fmt.Sprintf("insert into a(id, name) values(%v, 'nice name %v')", 1000+i, i), 1000, true)
+		result, err := conn.ExecuteFetch(fmt.Sprintf("insert into a(id, name) values(%v, 'nice name %v')", 1000+i, i), 1000, true)
 		if err != nil {
 			t.Fatalf("ExecuteFetch(%v) failed: %v", i, err)
 		}
+		if result.RowsAffected != 1 {
+			t.Errorf("insert into returned RowsAffected %v, was expecting 1", result.RowsAffected)
+		}
 	}
 
 	// And use a streaming query to read them back.
@@ -482,9 +499,13 @@ func testQueriesWithRealDatabase(t *testing.T, params *sqldb.ConnParams) {
 	readRowsUsingStream(t, conn, 101)
 
 	// And drop the table.
-	if _, err := conn.ExecuteFetch("drop table a", 0, false); err != nil {
+	result, err = conn.ExecuteFetch("drop table a", 0, false)
+	if err != nil {
 		t.Fatalf("drop table failed: %v", err)
 	}
+	if result.RowsAffected != 0 {
+		t.Errorf("insert into returned RowsAffected %v, was expecting 0", result.RowsAffected)
+	}
 }
 
 func readRowsUsingStream(t *testing.T, conn *Conn, expectedCount int) {
@@ -506,7 +527,8 @@ func readRowsUsingStream(t *testing.T, conn *Conn, expectedCount int) {
 			Charset:      CharacterSetBinary,
 			Flags: uint32(querypb.MySqlFlag_NOT_NULL_FLAG |
 				querypb.MySqlFlag_PRI_KEY_FLAG |
-				querypb.MySqlFlag_PART_KEY_FLAG),
+				querypb.MySqlFlag_PART_KEY_FLAG |
+				querypb.MySqlFlag_NUM_FLAG),
 		},
 		{
 			Name:         "name",