sql: support prepared statements

This patch adds the support of prepared statements.
Added a new type for handling prepared statements.
Added new methods for connector's interface in connector.go.
Added new IPROTO-constants for support of prepared statements
in const.go. Updated multi-package for corresponding it to connector's
interface.

Added a new function for checking the count of prepared statements in
config.lua in tests for Prepare and Unprepare.
Added benchmarks for SQL-select prepared statement.
Added examples of using Prepare in example_test.go.
Fixed some grammar inconsistencies for the method Execute.
Updated CHANGELOG.md.

Follows up #62
Closes #117

Author: vr009

CHANGELOG.md

@@ -18,6 +18,7 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
 - queue-utube handling (#85)
 - Master discovery (#113)
 - SQL support (#62)
+- Prepared statements (#117)
 
 ### Fixed
 

config.lua

@@ -91,6 +91,7 @@ box.once("init", function()
     -- grants for sql tests
     box.schema.user.grant('test', 'create,read,write,drop,alter', 'space')
     box.schema.user.grant('test', 'create', 'sequence')
+    box.schema.user.grant('guest', 'super')
 end)
 
 local function func_name()
@@ -110,11 +111,15 @@ local function simple_incr(a)
 end
 rawset(_G, 'simple_incr', simple_incr)
 
+local function test_stmt_count_check()
+    return box.info().sql().cache.stmt_count
+end
+rawset(_G, 'test_stmt_count_check', test_stmt_count_check)
 box.space.test:truncate()
 
 --box.schema.user.revoke('guest', 'read,write,execute', 'universe')
 
 -- Set listen only when every other thing is configured.
 box.cfg{
-    listen = os.getenv("TEST_TNT_LISTEN"),
+    listen = 3013,
 }

connector.go

@@ -18,6 +18,7 @@ type Connector interface {
 	Call17(functionName string, args interface{}) (resp *Response, err error)
 	Eval(expr string, args interface{}) (resp *Response, err error)
 	Execute(expr string, args interface{}) (resp *Response, err error)
+	Prepare(expr string) (stmt *PreparedStatement, err error)
 
 	GetTyped(space, index interface{}, key interface{}, result interface{}) (err error)
 	SelectTyped(space, index interface{}, offset, limit, iterator uint32, key interface{}, result interface{}) (err error)
@@ -28,6 +29,7 @@ type Connector interface {
 	CallTyped(functionName string, args interface{}, result interface{}) (err error)
 	Call17Typed(functionName string, args interface{}, result interface{}) (err error)
 	EvalTyped(expr string, args interface{}, result interface{}) (err error)
+	ExecuteTyped(expr string, args interface{}, result interface{}) (SQLInfo, []ColumnMetaData, error)
 
 	SelectAsync(space, index interface{}, offset, limit, iterator uint32, key interface{}) *Future
 	InsertAsync(space interface{}, tuple interface{}) *Future
@@ -38,4 +40,6 @@ type Connector interface {
 	CallAsync(functionName string, args interface{}) *Future
 	Call17Async(functionName string, args interface{}) *Future
 	EvalAsync(expr string, args interface{}) *Future
+	ExecuteAsync(expr string, args interface{}) *Future
+	PrepareAsync(expr string) *PreparedStatement
 }

const.go

@@ -12,6 +12,7 @@ const (
 	UpsertRequest    = 9
 	Call17Request    = 10
 	ExecuteRequest   = 11
+	PrepareRequest   = 13
 	PingRequest      = 64
 	SubscribeRequest = 66
 
@@ -31,9 +32,11 @@ const (
 	KeyData         = 0x30
 	KeyError        = 0x31
 	KeyMetaData     = 0x32
+	KeyBindCount    = 0x34
 	KeySQLText      = 0x40
 	KeySQLBind      = 0x41
 	KeySQLInfo      = 0x42
+	KeyStmtID       = 0x43
 
 	KeyFieldName               = 0x00
 	KeyFieldType               = 0x01

example_test.go

@@ -499,3 +499,39 @@ func ExampleConnection_Execute() {
 	fmt.Println("MetaData", resp.MetaData)
 	fmt.Println("SQL Info", resp.SQLInfo)
 }
+
+// To use prepared statements to query a tarantool instance, call Prepare.
+func ExampleConnection_Prepare() {
+	// Tarantool supports SQL since version 2.0.0
+	isLess, _ := test_helpers.IsTarantoolVersionLess(2, 0, 0)
+	if isLess {
+		return
+	}
+	server := "127.0.0.1:3013"
+	opts := tarantool.Opts{
+		Timeout:       500 * time.Millisecond,
+		Reconnect:     1 * time.Second,
+		MaxReconnects: 3,
+		User:          "test",
+		Pass:          "test",
+	}
+	client, err := tarantool.Connect(server, opts)
+	if err != nil {
+		fmt.Printf("Failed to connect: %s", err.Error())
+	}
+
+	// pass a query to prepare
+	stmtResp, err := client.Prepare("SELECT id FROM SQL_TEST WHERE id=? AND name=?")
+	fmt.Println("Prepare")
+	fmt.Println("Error", err)
+	fmt.Println("Statement ID", stmtResp.StatementID)
+
+	// pass the id of the statement to Execute
+	resp, err := stmtResp.Execute([]interface{}{2, "test"})
+	fmt.Println("Execute")
+	fmt.Println("Error", err)
+	fmt.Println("Code", resp.Code)
+	fmt.Println("Data", resp.Data)
+	fmt.Println("MetaData", resp.MetaData)
+	fmt.Println("SQL Info", resp.SQLInfo)
+}

multi/multi.go

@@ -340,13 +340,22 @@ func (connMulti *ConnectionMulti) Eval(expr string, args interface{}) (resp *tar
 	return connMulti.getCurrentConnection().Eval(expr, args)
 }
 
-// Execute passes sql expression to Tarantool for execution.
+// Execute passes a sql expression to Tarantool for execution.
 //
 // Since 1.6.0
 func (connMulti *ConnectionMulti) Execute(expr string, args interface{}) (resp *tarantool.Response, err error) {
 	return connMulti.getCurrentConnection().Execute(expr, args)
 }
 
+// Prepare sends a sql statement to prepare.
+//
+// Returns a PreparedStatement object for the current connection.
+//
+// Since 1.6.0
+func (connMulti *ConnectionMulti) Prepare(expr string) (resp *tarantool.PreparedStatement, err error) {
+	return connMulti.getCurrentConnection().Prepare(expr)
+}
+
 // GetTyped performs select (with limit = 1 and offset = 0) to box space and
 // fills typed result.
 func (connMulti *ConnectionMulti) GetTyped(space, index interface{}, key interface{}, result interface{}) (err error) {
@@ -401,6 +410,11 @@ func (connMulti *ConnectionMulti) EvalTyped(expr string, args interface{}, resul
 	return connMulti.getCurrentConnection().EvalTyped(expr, args, result)
 }
 
+// ExecuteTyped passes a sql expression for execution.
+func (connMulti *ConnectionMulti) ExecuteTyped(expr string, args interface{}, result interface{}) (tarantool.SQLInfo, []tarantool.ColumnMetaData, error) {
+	return connMulti.getCurrentConnection().ExecuteTyped(expr, args, result)
+}
+
 // SelectAsync sends select request to Tarantool and returns Future.
 func (connMulti *ConnectionMulti) SelectAsync(space, index interface{}, offset, limit, iterator uint32, key interface{}) *tarantool.Future {
 	return connMulti.getCurrentConnection().SelectAsync(space, index, offset, limit, iterator, key)
@@ -454,3 +468,17 @@ func (connMulti *ConnectionMulti) Call17Async(functionName string, args interfac
 func (connMulti *ConnectionMulti) EvalAsync(expr string, args interface{}) *tarantool.Future {
 	return connMulti.getCurrentConnection().EvalAsync(expr, args)
 }
+
+// ExecuteAsync passes a sql expression for execution.
+//
+// Since 1.6.0
+func (connMulti *ConnectionMulti) ExecuteAsync(expr string, args interface{}) *tarantool.Future {
+	return connMulti.getCurrentConnection().ExecuteAsync(expr, args)
+}
+
+// PrepareAsync passes a SQL statement to prepare.
+//
+// Since 1.6.0
+func (connMulti *ConnectionMulti) PrepareAsync(expr string) *tarantool.PreparedStatement {
+	return connMulti.getCurrentConnection().PrepareAsync(expr)
+}

request.go

@@ -131,6 +131,16 @@ func (conn *Connection) Execute(expr string, args interface{}) (resp *Response,
 	return conn.ExecuteAsync(expr, args).Get()
 }
 
+// Prepare sends a sql statement to prepare.
+//
+// It is equal to conn.PrepareAsync(expr).Get().
+// Since 1.6.0
+func (conn *Connection) Prepare(expr string) (stmt *PreparedStatement, err error) {
+	stmt = conn.PrepareAsync(expr)
+	err = stmt.wait()
+	return stmt, err
+}
+
 // single used for conn.GetTyped for decode one tuple.
 type single struct {
 	res   interface{}
@@ -380,6 +390,87 @@ func (conn *Connection) ExecuteAsync(expr string, args interface{}) *Future {
 	})
 }
 
+// PrepareAsync sends a sql statement to prepare and returns PreparedStatement object.
+//
+// Since 1.6.0
+func (conn *Connection) PrepareAsync(expr string) *PreparedStatement {
+	future := conn.newFuture(PrepareRequest)
+	stmt := newPreparedStatement(future.send(conn, func(enc *msgpack.Encoder) error {
+		enc.EncodeMapLen(1)
+		enc.EncodeUint64(KeySQLText)
+		return enc.EncodeString(expr)
+	}), conn)
+	return stmt
+}
+
+type PreparedStatementID uint64
+
+// PreparedStatement is a type for handling prepared statements
+//
+// Since 1.6.0
+type PreparedStatement struct {
+	StatementID PreparedStatementID
+	MetaData    []ColumnMetaData
+	ParamCount  uint64
+	conn        *Connection
+	fut         *Future
+}
+
+func newPreparedStatement(fut *Future, conn *Connection) *PreparedStatement {
+	stmt := new(PreparedStatement)
+	stmt.fut = fut
+	stmt.conn = conn
+	return stmt
+}
+
+// wait until the prepared statement is ready and fill the statement object
+func (stmt *PreparedStatement) wait() error {
+	resp, err := stmt.fut.Get()
+	stmt.StatementID = PreparedStatementID(resp.StmtID)
+	stmt.MetaData = resp.MetaData
+	stmt.ParamCount = resp.BindCount
+	return err
+}
+
+// UnprepareAsync sends an undo request and returns Future
+func (stmt *PreparedStatement) UnprepareAsync() *Future {
+	err := stmt.wait()
+	if err != nil {
+		return NewErrorFuture(err)
+	}
+	future := stmt.conn.newFuture(PrepareRequest)
+	return future.send(stmt.conn, func(enc *msgpack.Encoder) error {
+		enc.EncodeMapLen(1)
+		enc.EncodeUint64(KeyStmtID)
+		return enc.EncodeUint64(uint64(stmt.StatementID))
+	})
+}
+
+// Unprepare undo the prepared statement
+func (stmt *PreparedStatement) Unprepare() (resp *Response, err error) {
+	return stmt.UnprepareAsync().Get()
+}
+
+// ExecuteAsync sends the prepared SQL statement for execution and returns Future
+func (stmt *PreparedStatement) ExecuteAsync(args interface{}) *Future {
+	err := stmt.wait()
+	if err != nil {
+		return NewErrorFuture(err)
+	}
+	return stmt.fut.send(stmt.conn, func(enc *msgpack.Encoder) error {
+		enc.EncodeMapLen(2)
+		enc.EncodeUint64(KeyStmtID)
+		enc.EncodeUint64(uint64(stmt.StatementID))
+		enc.EncodeUint64(KeySQLBind)
+		return encodeSQLBind(enc, args)
+	})
+}
+
+// Execute sends the prepared SQL statement for execution
+func (stmt *PreparedStatement) Execute(args interface{}) (resp *Response, err error) {
+	return stmt.ExecuteAsync(args).Get()
+}
+
 // KeyValueBind is a type for encoding named SQL parameters
 type KeyValueBind struct {
 	Key   string

response.go

@@ -11,10 +11,12 @@ type Response struct {
 	Code      uint32
 	Error     string // error message
 	// Data contains deserialized data for untyped requests
-	Data     []interface{}
-	MetaData []ColumnMetaData
-	SQLInfo  SQLInfo
-	buf      smallBuf
+	Data      []interface{}
+	MetaData  []ColumnMetaData
+	SQLInfo   SQLInfo
+	StmtID    uint64
+	BindCount uint64
+	buf       smallBuf
 }
 
 type ColumnMetaData struct {
@@ -178,6 +180,14 @@ func (resp *Response) decodeBody() (err error) {
 				if err = d.Decode(&resp.MetaData); err != nil {
 					return err
 				}
+			case KeyStmtID:
+				if resp.StmtID, err = d.DecodeUint64(); err != nil {
+					return err
+				}
+			case KeyBindCount:
+				if resp.BindCount, err = d.DecodeUint64(); err != nil {
+					return err
+				}
 			default:
 				if err = d.Skip(); err != nil {
 					return err