update to radon upstream

Author: BohuTANG

.travis.yml

@@ -4,6 +4,7 @@ go:
   - 1.x
 
 before_install:
+  - go get github.com/shopspring/decimal
   - go get github.com/pierrre/gotestcover
   - go get github.com/stretchr/testify/assert
 

driver/mock.go

@@ -170,6 +170,11 @@ func (th *TestHandler) ServerVersion() string {
 	return "FakeDB"
 }
 
+// SetServerVersion implements the interface.
+func (th *TestHandler) SetServerVersion() {
+	return
+}
+
 // NewSession implements the interface.
 func (th *TestHandler) NewSession(s *Session) {
 	th.mu.Lock()

driver/server.go

@@ -19,16 +19,16 @@ import (
 
 	"github.com/xelabs/go-mysqlstack/proto"
 	"github.com/xelabs/go-mysqlstack/sqldb"
-	"github.com/xelabs/go-mysqlstack/xlog"
-
 	"github.com/xelabs/go-mysqlstack/sqlparser/depends/common"
 	querypb "github.com/xelabs/go-mysqlstack/sqlparser/depends/query"
 	"github.com/xelabs/go-mysqlstack/sqlparser/depends/sqltypes"
+	"github.com/xelabs/go-mysqlstack/xlog"
 )
 
 // Handler interface.
 type Handler interface {
 	ServerVersion() string
+	SetServerVersion()
 	NewSession(session *Session)
 	SessionInc(session *Session)
 	SessionDec(session *Session)
@@ -54,8 +54,6 @@ type Listener struct {
 
 	// Incrementing ID for connection id.
 	connectionID uint32
-
-	serverVersion string
 }
 
 // NewListener creates a new Listener.
@@ -66,12 +64,11 @@ func NewListener(log *xlog.Log, address string, handler Handler) (*Listener, err
 	}
 
 	return &Listener{
-		log:           log,
-		address:       address,
-		handler:       handler,
-		listener:      listener,
-		connectionID:  1,
-		serverVersion: handler.ServerVersion(),
+		log:          log,
+		address:      address,
+		handler:      handler,
+		listener:     listener,
+		connectionID: 1,
 	}, nil
 }
 
@@ -86,7 +83,7 @@ func (l *Listener) Accept() {
 		}
 		ID := l.connectionID
 		l.connectionID++
-		go l.handle(conn, ID, l.serverVersion)
+		go l.handle(conn, ID)
 	}
 }
 
@@ -123,16 +120,21 @@ func (l *Listener) parserComStatementExecute(data []byte, session *Session) (*St
 	if err != nil {
 		return nil, err
 	}
-	protoStmt, err := proto.UnPackStatementExecute(data[1:], stmt.ParamCount, sqltypes.ParseMySQLValues)
-	if err != nil {
+
+	protoStmt := &proto.Statement{
+		ID:         stmt.ID,
+		ParamCount: stmt.ParamCount,
+		ParamsType: stmt.ParamsType,
+		BindVars:   stmt.BindVars,
+	}
+	if err = proto.UnPackStatementExecute(data[1:], protoStmt, sqltypes.ParseMySQLValues); err != nil {
 		return nil, err
 	}
-	stmt.BindVars = protoStmt.BindVars
 	return stmt, nil
 }
 
 // handle is called in a go routine for each client connection.
-func (l *Listener) handle(conn net.Conn, ID uint32, serverVersion string) {
+func (l *Listener) handle(conn net.Conn, ID uint32) {
 	var err error
 	var data []byte
 	var authPkt []byte
@@ -146,7 +148,11 @@ func (l *Listener) handle(conn net.Conn, ID uint32, serverVersion string) {
 			log.Error("server.handle.panic:\n%v\n%s", x, debug.Stack())
 		}
 	}()
-	session := newSession(log, ID, l.serverVersion, conn)
+
+	// set server version if backend MySQL version is different.
+	l.handler.SetServerVersion()
+
+	session := newSession(log, ID, l.handler.ServerVersion(), conn)
 	// Session check.
 	if err = l.handler.SessionCheck(session); err != nil {
 		log.Warning("session[%v].check.failed.error:%+v", ID, err)
@@ -252,6 +258,7 @@ func (l *Listener) handle(conn net.Conn, ID uint32, serverVersion string) {
 				ID:          id,
 				PrepareStmt: query,
 				ParamCount:  paramCount,
+				ParamsType:  make([]int32, paramCount),
 				BindVars:    make(map[string]*querypb.BindVariable, paramCount),
 			}
 			for i := uint16(0); i < paramCount; i++ {
@@ -274,6 +281,7 @@ func (l *Listener) handle(conn net.Conn, ID uint32, serverVersion string) {
 					return
 				}
 			}
+
 			if err = l.handler.ComQuery(session, stmt.PrepareStmt, sqltypes.CopyBindVariables(stmt.BindVars), func(qr *sqltypes.Result) error {
 				return session.writeBinaryRows(qr)
 			}); err != nil {

driver/statement.go

@@ -12,7 +12,6 @@ package driver
 import (
 	"github.com/xelabs/go-mysqlstack/proto"
 	"github.com/xelabs/go-mysqlstack/sqldb"
-
 	querypb "github.com/xelabs/go-mysqlstack/sqlparser/depends/query"
 	"github.com/xelabs/go-mysqlstack/sqlparser/depends/sqltypes"
 )
@@ -23,6 +22,7 @@ type Statement struct {
 	ID          uint32
 	ParamCount  uint16
 	PrepareStmt string
+	ParamsType  []int32
 	ColumnNames []string
 	BindVars    map[string]*querypb.BindVariable
 }

makefile

@@ -5,6 +5,7 @@ fmt:
 	go vet ./...
 
 test:
+	go get github.com/shopspring/decimal
 	go get github.com/stretchr/testify/assert
 	@echo "--> Testing..."
 	@$(MAKE) testxlog

proto/auth.go

@@ -97,7 +97,10 @@ func (a *Auth) UnPack(payload []byte) error {
 			return fmt.Errorf("auth.unpack: can't read authResponse")
 		}
 	} else {
-		if a.authResponse, err = buf.ReadBytesNUL(); err != nil {
+		if a.authResponse, err = buf.ReadBytes(20); err != nil {
+			return fmt.Errorf("auth.unpack: can't read authResponse")
+		}
+		if err = buf.ReadZero(1); err != nil {
 			return fmt.Errorf("auth.unpack: can't read authResponse")
 		}
 	}

proto/auth_test.go

@@ -179,19 +179,19 @@ func TestAuthWithoutSecure(t *testing.T) {
 	want.authResponseLen = 20
 	want.clientFlags = DefaultClientCapability &^ sqldb.CLIENT_SECURE_CONNECTION &^ sqldb.CLIENT_PLUGIN_AUTH_LENENC_CLIENT_DATA
 	want.clientFlags |= sqldb.CLIENT_CONNECT_WITH_DB
-	want.authResponse = nativePassword("sbtest", DefaultSalt)
-	want.user = "sbtest"
-	want.database = "sbtest"
+	want.authResponse = nativePassword("password", DefaultSalt)
+	want.user = "root"
+	want.database = "test_db"
 	want.pluginName = DefaultAuthPluginName
 
 	got := NewAuth()
 	err := got.UnPack(want.Pack(
 		DefaultClientCapability&^sqldb.CLIENT_SECURE_CONNECTION,
 		0x02,
-		"sbtest",
-		"sbtest",
+		"root",
+		"password",
 		DefaultSalt,
-		"sbtest",
+		"test_db",
 	))
 	got.authResponseLen = 20
 	assert.Nil(t, err)

proto/statement.go

@@ -13,7 +13,6 @@ import (
 	"fmt"
 
 	"github.com/xelabs/go-mysqlstack/sqldb"
-
 	"github.com/xelabs/go-mysqlstack/sqlparser/depends/common"
 	querypb "github.com/xelabs/go-mysqlstack/sqlparser/depends/query"
 	"github.com/xelabs/go-mysqlstack/sqlparser/depends/sqltypes"
@@ -26,6 +25,7 @@ type Statement struct {
 	ColumnCount uint16
 	ParamCount  uint16
 	Warnings    uint16
+	ParamsType  []int32
 	ColumnNames []string
 
 	BindVars map[string]*querypb.BindVariable
@@ -156,87 +156,79 @@ func PackStatementExecute(stmtID uint32, parameters []sqltypes.Value) ([]byte, e
 }
 
 // UnPackStatementExecute -- unpack the stmt-execute packet from client.
-func UnPackStatementExecute(data []byte, paramsCount uint16, parseValueFn func(*common.Buffer, querypb.Type) (interface{}, error)) (*Statement, error) {
+func UnPackStatementExecute(data []byte, prepare *Statement, parseValueFn func(*common.Buffer, querypb.Type) (interface{}, error)) error {
 	var err error
-	var paramsType []int32
-
-	stmt := &Statement{}
 	bitMap := make([]byte, 0)
 	buf := common.ReadBuffer(data)
 
-	// statement ID
-	if stmt.ID, err = buf.ReadU32(); err != nil {
-		return nil, sqldb.NewSQLErrorf(sqldb.ER_MALFORMED_PACKET, "reading statement ID failed")
+	if _, err = buf.ReadU32(); err != nil {
+		return sqldb.NewSQLErrorf(sqldb.ER_MALFORMED_PACKET, "reading statement ID failed")
 	}
 
 	// cursor type flags
 	if _, err = buf.ReadU8(); err != nil {
-		return nil, sqldb.NewSQLErrorf(sqldb.ER_MALFORMED_PACKET, "reading cursor type flags failed")
+		return sqldb.NewSQLErrorf(sqldb.ER_MALFORMED_PACKET, "reading cursor type flags failed")
 	}
 
 	// iteration count
 	var itercount uint32
 	if itercount, err = buf.ReadU32(); err != nil {
-		return nil, sqldb.NewSQLErrorf(sqldb.ER_MALFORMED_PACKET, "reading iteration count failed")
+		return sqldb.NewSQLErrorf(sqldb.ER_MALFORMED_PACKET, "reading iteration count failed")
 	}
 	if itercount != 1 {
-		return nil, sqldb.NewSQLErrorf(sqldb.ER_MALFORMED_PACKET, "iteration count is not equal to 1")
+		return sqldb.NewSQLErrorf(sqldb.ER_MALFORMED_PACKET, "iteration count is not equal to 1")
 	}
 
-	if paramsCount > 0 {
-		// Init.
-		paramsType = make([]int32, paramsCount)
-		stmt.BindVars = make(map[string]*querypb.BindVariable)
-
-		if bitMap, err = buf.ReadBytes(int((paramsCount + 7) / 8)); err != nil {
-			return nil, sqldb.NewSQLErrorf(sqldb.ER_MALFORMED_PACKET, "reading NULL-bitmap failed")
+	if prepare.ParamCount > 0 {
+		if bitMap, err = buf.ReadBytes(int((prepare.ParamCount + 7) / 8)); err != nil {
+			return sqldb.NewSQLErrorf(sqldb.ER_MALFORMED_PACKET, "reading NULL-bitmap failed")
 		}
 
 		var newParamsBoundFlag byte
 		if newParamsBoundFlag, err = buf.ReadU8(); err != nil {
-			return nil, sqldb.NewSQLErrorf(sqldb.ER_MALFORMED_PACKET, "reading NULL-bitmap failed")
+			return sqldb.NewSQLErrorf(sqldb.ER_MALFORMED_PACKET, "reading NULL-bitmap failed")
 		}
 		if newParamsBoundFlag == 0x01 {
 			var mysqlType, flags byte
-			for i := uint16(0); i < paramsCount; i++ {
+			for i := uint16(0); i < prepare.ParamCount; i++ {
 				if mysqlType, err = buf.ReadU8(); err != nil {
-					return nil, sqldb.NewSQLErrorf(sqldb.ER_MALFORMED_PACKET, "reading parameter type failed")
+					return sqldb.NewSQLErrorf(sqldb.ER_MALFORMED_PACKET, "reading parameter type failed")
 				}
 
 				if flags, err = buf.ReadU8(); err != nil {
-					return nil, sqldb.NewSQLErrorf(sqldb.ER_MALFORMED_PACKET, "reading parameter flags failed")
+					return sqldb.NewSQLErrorf(sqldb.ER_MALFORMED_PACKET, "reading parameter flags failed")
 				}
 				// Convert MySQL type to Vitess type.
 				valType, err := sqltypes.MySQLToType(int64(mysqlType), int64(flags))
 				if err != nil {
-					return nil, sqldb.NewSQLErrorf(sqldb.ER_MALFORMED_PACKET, fmt.Sprintf("MySQLToType(%v,%v) failed: %v", mysqlType, flags, err))
+					return sqldb.NewSQLErrorf(sqldb.ER_MALFORMED_PACKET, fmt.Sprintf("MySQLToType(%v,%v) failed: %v", mysqlType, flags, err))
 				}
-				paramsType[i] = int32(valType)
+				prepare.ParamsType[i] = int32(valType)
 			}
 		}
 
-		for i := uint16(0); i < paramsCount; i++ {
+		for i := uint16(0); i < prepare.ParamCount; i++ {
 			var val interface{}
-			if paramsType[i] == int32(sqltypes.Text) || paramsType[i] == int32(sqltypes.Blob) {
+			if prepare.ParamsType[i] == int32(sqltypes.Text) || prepare.ParamsType[i] == int32(sqltypes.Blob) {
 				continue
 			}
 
 			if (bitMap[i/8] & (1 << uint(i%8))) > 0 {
 				val, err = parseValueFn(buf, sqltypes.Null)
 			} else {
-				val, err = parseValueFn(buf, querypb.Type(paramsType[i]))
+				val, err = parseValueFn(buf, querypb.Type(prepare.ParamsType[i]))
 			}
 			if err != nil {
-				return nil, sqldb.NewSQLErrorf(sqldb.ER_MALFORMED_PACKET, fmt.Sprintf("decoding parameter value failed(%v) failed: %v", paramsType[i], err))
+				return sqldb.NewSQLErrorf(sqldb.ER_MALFORMED_PACKET, fmt.Sprintf("decoding parameter value failed(%v) failed: %v", prepare.ParamsType[i], err))
 			}
 
 			// If value is nil, must set bind variables to nil.
 			bv, err := sqltypes.BuildBindVariable(val)
 			if err != nil {
-				return nil, sqldb.NewSQLErrorf(sqldb.ER_MALFORMED_PACKET, fmt.Sprintf("build converted parameters value failed: %v", err))
+				return sqldb.NewSQLErrorf(sqldb.ER_MALFORMED_PACKET, fmt.Sprintf("build converted parameters value failed: %v", err))
 			}
-			stmt.BindVars[fmt.Sprintf("v%d", i+1)] = bv
+			prepare.BindVars[fmt.Sprintf("v%d", i+1)] = bv
 		}
 	}
-	return stmt, nil
+	return nil
 }

proto/statement_test.go

@@ -93,9 +93,15 @@ func TestStatementExecute(t *testing.T) {
 	parseFn := func(*common.Buffer, querypb.Type) (interface{}, error) {
 		return nil, nil
 	}
-	got, err := UnPackStatementExecute(datas, 4, parseFn)
+
+	protoStmt := &Statement{
+		ID:         id,
+		ParamCount: uint16(len(values)),
+		ParamsType: make([]int32, len(values)),
+		BindVars:   make(map[string]*querypb.BindVariable, len(values)),
+	}
+	err = UnPackStatementExecute(datas, protoStmt, parseFn)
 	assert.Nil(t, err)
-	assert.NotNil(t, got)
 }
 
 func TestStatementExecuteUnPackError(t *testing.T) {
@@ -140,8 +146,44 @@ func TestStatementExecuteUnPackError(t *testing.T) {
 	buff := common.NewBuffer(32)
 	fs := []func(buff *common.Buffer){f0, f1, f2, f3, f4, f5, f6}
 	for i := 0; i < len(fs); i++ {
-		_, err := UnPackStatementExecute(buff.Datas(), 1, parseFn)
+
+		protoStmt := &Statement{
+			ID:         1,
+			ParamCount: 2,
+			ParamsType: make([]int32, 2),
+			BindVars:   make(map[string]*querypb.BindVariable, 2),
+		}
+
+		err := UnPackStatementExecute(buff.Datas(), protoStmt, parseFn)
 		assert.NotNil(t, err)
 		fs[i](buff)
 	}
 }
+
+// issue 462.
+// https://dev.mysql.com/doc/internals/en/com-stmt-execute.html
+// test about new-params-bound-flag about 0 1
+func TestStatementExecuteBatchUnPackStatementExecute(t *testing.T) {
+	data := []byte{ /*23,*/ 18, 0, 0, 0, 128, 1, 0, 0, 0, 0, 1, 1, 128, 1}
+	data2 := []byte{ /*23,*/ 18, 0, 0, 0, 128, 1, 0, 0, 0, 0, 0, 1, 128, 1}
+
+	var dataBatch [][]byte
+	dataBatch = append(dataBatch, data)
+	dataBatch = append(dataBatch, data2)
+
+	parseFn := func(*common.Buffer, querypb.Type) (interface{}, error) {
+		return nil, nil
+	}
+
+	protoStmt := &Statement{
+		ID:         23,
+		ParamCount: 1,
+		ParamsType: make([]int32, 1),
+		BindVars:   make(map[string]*querypb.BindVariable, 1),
+	}
+	err := UnPackStatementExecute(dataBatch[0], protoStmt, parseFn)
+	assert.Nil(t, err)
+
+	err = UnPackStatementExecute(dataBatch[1], protoStmt, parseFn)
+	assert.Nil(t, err)
+}