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

@@ -15,6 +15,7 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
 - Public API with request object types (#126)
 - Support decimal type in msgpack (#96)
 - Support datetime type in msgpack (#118)
+- Prepared SQL statements (#117)
 
 ### Changed
 

config.lua

@@ -118,6 +118,10 @@ local function push_func(cnt)
 end
 rawset(_G, 'push_func', push_func)
 
+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')

connection_pool/connection_pool.go

@@ -314,6 +314,18 @@ func (connPool *ConnectionPool) Execute(expr string, args interface{}, userMode
 	return conn.Execute(expr, args)
 }
 
+// Prepare sends a sql statement to prepare.
+//
+// Since 1.7.0
+func (connPool *ConnectionPool) Prepare(expr string, userMode Mode) (stmt *tarantool.PreparedStatement, err error) {
+	conn, err := connPool.getNextConnection(userMode)
+	if err != nil {
+		return nil, err
+	}
+
+	return conn.Prepare(expr)
+}
+
 // GetTyped performs select (with limit = 1 and offset = 0)
 // to box space and fills typed result.
 func (connPool *ConnectionPool) GetTyped(space, index interface{}, key interface{}, result interface{}, userMode ...Mode) (err error) {
@@ -424,7 +436,7 @@ func (connPool *ConnectionPool) EvalTyped(expr string, args interface{}, result
 	return conn.EvalTyped(expr, args, result)
 }
 
-// ExecuteTyped passes a sql expression for execution.
+// ExecuteTyped passes an sql expression for execution.
 //
 // Since 1.7.0
 func (connPool *ConnectionPool) ExecuteTyped(expr string, args interface{}, result interface{}, userMode Mode) (tarantool.SQLInfo, []tarantool.ColumnMetaData, error) {
@@ -560,6 +572,18 @@ func (connPool *ConnectionPool) ExecuteAsync(expr string, args interface{}, user
 	return conn.ExecuteAsync(expr, args)
 }
 
+// PrepareAsync sends a sql statement to prepare and returns PreparedStatement object.
+//
+// Since 1.7.0
+func (connPool *ConnectionPool) PrepareAsync(expr string, userMode Mode) *tarantool.PreparedStatement {
+	conn, err := connPool.getNextConnection(userMode)
+	if err != nil {
+		return nil
+	}
+
+	return conn.PrepareAsync(expr)
+}
+
 // Do sends the request and returns a response.
 func (connPool *ConnectionPool) Do(req tarantool.Request, userMode Mode) (*tarantool.Response, error) {
 	conn, err := connPool.getNextConnection(userMode)

connector.go

@@ -19,6 +19,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)
@@ -30,6 +31,7 @@ type Connector interface {
 	Call16Typed(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
@@ -41,6 +43,8 @@ type Connector interface {
 	Call16Async(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
 
 	Do(req Request) (resp *Response, err error)
 	DoTyped(req Request, result interface{}) (err error)

const.go

@@ -12,6 +12,7 @@ const (
 	UpsertRequestCode    = 9
 	Call17RequestCode    = 10 /* call in >= 1.7 format */
 	ExecuteRequestCode   = 11
+	PrepareRequestCode   = 13
 	PingRequestCode      = 64
 	SubscribeRequestCode = 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

@@ -651,3 +651,48 @@ 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{}{1, "test_1"})
+	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)
+
+	// use the same object for execute with other arguments
+	resp, err = stmtResp.Execute([]interface{}{2, "test_2"})
+	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

@@ -349,13 +349,22 @@ func (connMulti *ConnectionMulti) Eval(expr string, args interface{}) (resp *tar
 	return connMulti.getCurrentConnection().Eval(expr, args)
 }
 
-// Execute passes a sql expression to Tarantool for execution.
+// Execute passes an 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.7.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) {
@@ -419,7 +428,7 @@ func (connMulti *ConnectionMulti) EvalTyped(expr string, args interface{}, resul
 	return connMulti.getCurrentConnection().EvalTyped(expr, args, result)
 }
 
-// ExecuteTyped passes a sql expression for execution.
+// ExecuteTyped passes an 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)
 }
@@ -494,6 +503,13 @@ func (connMulti *ConnectionMulti) ExecuteAsync(expr string, args interface{}) *t
 	return connMulti.getCurrentConnection().ExecuteAsync(expr, args)
 }
 
+// PrepareAsync passes a sql statement to prepare.
+//
+// Since 1.7.0
+func (connMulti *ConnectionMulti) PrepareAsync(expr string) *tarantool.PreparedStatement {
+	return connMulti.getCurrentConnection().PrepareAsync(expr)
+}
+
 // Do sends the request and returns a response.
 func (connMulti *ConnectionMulti) Do(req tarantool.Request) (*tarantool.Response, error) {
 	return connMulti.getCurrentConnection().Do(req)

request.go

@@ -91,6 +91,26 @@ func fillExecute(enc *msgpack.Encoder, expr string, args interface{}) error {
 	return encodeSQLBind(enc, args)
 }
 
+func fillPrepare(enc *msgpack.Encoder, expr string) error {
+	enc.EncodeMapLen(1)
+	enc.EncodeUint64(KeySQLText)
+	return enc.EncodeString(expr)
+}
+
+func fillUnprepare(enc *msgpack.Encoder, stmt PreparedStatement) error {
+	enc.EncodeMapLen(1)
+	enc.EncodeUint64(KeyStmtID)
+	return enc.EncodeUint64(uint64(stmt.StatementID))
+}
+
+func fillPreparedExecute(enc *msgpack.Encoder, stmt PreparedStatement, args interface{}) error {
+	enc.EncodeMapLen(2)
+	enc.EncodeUint64(KeyStmtID)
+	enc.EncodeUint64(uint64(stmt.StatementID))
+	enc.EncodeUint64(KeySQLBind)
+	return encodeSQLBind(enc, args)
+}
+
 func fillPing(enc *msgpack.Encoder) error {
 	return enc.EncodeMapLen(0)
 }
@@ -190,6 +210,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.7.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{}
@@ -389,6 +419,78 @@ func (conn *Connection) ExecuteAsync(expr string, args interface{}) *Future {
 	return conn.DoAsync(req)
 }
 
+// PrepareAsync sends a sql statement to prepare and returns PreparedStatement object.
+//
+// Since 1.7.0
+func (conn *Connection) PrepareAsync(expr string) *PreparedStatement {
+	req := newPrepareRequest(expr)
+	fut := conn.DoAsync(req)
+	stmt := newPreparedStatement(fut, conn)
+	return stmt
+}
+
+type PreparedStatementID uint64
+
+// PreparedStatement is a type for handling prepared statements
+//
+// Since 1.7.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)
+	}
+	req := newUnprepareRequest(*stmt)
+	fut := stmt.conn.DoAsync(req)
+	return fut
+}
+
+// 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)
+	}
+	req := newPreparedExecuteRequest(*stmt)
+	req.Args(args)
+	fut := stmt.conn.DoAsync(req)
+	return fut
+}
+
+// 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
@@ -981,3 +1083,63 @@ func (req *ExecuteRequest) Args(args interface{}) *ExecuteRequest {
 func (req *ExecuteRequest) Body(res SchemaResolver, enc *msgpack.Encoder) error {
 	return fillExecute(enc, req.expr, req.args)
 }
+
+type prepareRequest struct {
+	baseRequest
+	expr string
+}
+
+// newPrepareRequest returns a new empty prepareRequest.
+func newPrepareRequest(expr string) *prepareRequest {
+	req := new(prepareRequest)
+	req.requestCode = PrepareRequestCode
+	req.expr = expr
+	return req
+}
+
+// Body fills an encoder with the execute request body.
+func (req *prepareRequest) Body(res SchemaResolver, enc *msgpack.Encoder) error {
+	return fillPrepare(enc, req.expr)
+}
+
+type unprepareRequest struct {
+	baseRequest
+	stmt PreparedStatement
+}
+
+// newUnprepareRequest returns a new empty unprepareRequest.
+func newUnprepareRequest(stmt PreparedStatement) *unprepareRequest {
+	req := new(unprepareRequest)
+	req.requestCode = PrepareRequestCode
+	req.stmt = stmt
+	return req
+}
+
+// Body fills an encoder with the execute request body.
+func (req *unprepareRequest) Body(res SchemaResolver, enc *msgpack.Encoder) error {
+	return fillUnprepare(enc, req.stmt)
+}
+
+type preparedExecuteRequest struct {
+	baseRequest
+	stmt PreparedStatement
+	args interface{}
+}
+
+// newPreparedExecuteRequest returns a new empty preparedExecuteRequest.
+func newPreparedExecuteRequest(stmt PreparedStatement) *preparedExecuteRequest {
+	req := new(preparedExecuteRequest)
+	req.requestCode = ExecuteRequestCode
+	req.stmt = stmt
+	return req
+}
+
+func (req *preparedExecuteRequest) Args(args interface{}) *preparedExecuteRequest {
+	req.args = args
+	return req
+}
+
+// Body fills an encoder with the execute request body.
+func (req *preparedExecuteRequest) Body(res SchemaResolver, enc *msgpack.Encoder) error {
+	return fillPreparedExecute(enc, req.stmt, req.args)
+}