sql: support prepared statements

This patch adds the support of prepared statements.
Added a new type for handling prepared statements.
Added new IPROTO-constants for support of prepared statements
in const.go.

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

@@ -117,7 +117,6 @@ local function push_func(cnt)
     return cnt
 end
 rawset(_G, 'push_func', push_func)
-
 box.space.test:truncate()
 
 --box.schema.user.revoke('guest', 'read,write,execute', 'universe')

connection.go

@@ -1009,3 +1009,13 @@ func (conn *Connection) OverrideSchema(s *Schema) {
 		conn.Schema = s
 	}
 }
+
+// NewPrepareStatement passes a sql statement to Tarantool for preparation synchronously.
+func (conn *Connection) NewPrepareStatement(expr string) (*PreparedStatement, error) {
+	req := NewPrepareRequest(expr)
+	resp, err := conn.Do(req).Get()
+	if err != nil {
+		return nil, err
+	}
+	return newPreparedStatement(conn, resp), nil
+}

connection_pool/config.lua

@@ -21,6 +21,21 @@ box.once("init", function()
         parts = {{ field = 1, type = 'string' }},
         if_not_exists = true
     })
+
+    local sp = box.schema.space.create('SQL_TEST', {
+        id = 521,
+        if_not_exists = true,
+        format = {
+            {name = "NAME0", type = "unsigned"},
+            {name = "NAME1", type = "string"},
+            {name = "NAME2", type = "string"},
+        }
+    })
+    sp:create_index('primary', {type = 'tree', parts = {1, 'uint'}, if_not_exists = true})
+    sp:insert{1, "test", "test"}
+    -- grants for sql tests
+    box.schema.user.grant('test', 'create,read,write,drop,alter', 'space')
+    box.schema.user.grant('test', 'create', 'sequence')
 end)
 
 local function simple_incr(a)

connection_pool/connection_pool.go

@@ -526,6 +526,9 @@ func (connPool *ConnectionPool) EvalAsync(expr string, args interface{}, userMod
 
 // Do sends the request and returns a future.
 func (connPool *ConnectionPool) Do(req tarantool.Request, userMode Mode) *tarantool.Future {
+	if stickyReq, ok := req.(tarantool.StickyRequest); ok {
+		return stickyReq.Conn().Do(req)
+	}
 	conn, err := connPool.getNextConnection(userMode)
 	if err != nil {
 		return newErrorFuture(err)
@@ -788,3 +791,12 @@ func newErrorFuture(err error) *tarantool.Future {
 	fut.SetError(err)
 	return fut
 }
+
+// NewPrepareStatement passes a sql statement to Tarantool for preparation synchronously.
+func (connPool *ConnectionPool) NewPrepareStatement(expr string, userMode Mode) (*tarantool.PreparedStatement, error) {
+	conn, err := connPool.getNextConnection(userMode)
+	if err != nil {
+		return nil, err
+	}
+	return conn.NewPrepareStatement(expr)
+}

connection_pool/connection_pool_test.go

@@ -3,6 +3,7 @@ package connection_pool_test
 import (
 	"log"
 	"os"
+	"reflect"
 	"strings"
 	"testing"
 	"time"
@@ -1276,6 +1277,74 @@ func TestDo(t *testing.T) {
 	require.NotNilf(t, resp, "response is nil after Ping")
 }
 
+func TestNewPrepareStatement(t *testing.T) {
+	// Tarantool supports SQL since version 2.0.0
+	isLess, err := test_helpers.IsTarantoolVersionLess(2, 0, 0)
+	if err != nil {
+		t.Fatalf("Could not check the Tarantool version")
+	}
+	if isLess {
+		t.Skip()
+	}
+
+	roles := []bool{true, true, false, true, false}
+
+	err = test_helpers.SetClusterRO(servers, connOpts, roles)
+	require.Nilf(t, err, "fail to set roles for cluster")
+
+	connPool, err := connection_pool.Connect(servers, connOpts)
+	require.Nilf(t, err, "failed to connect")
+	require.NotNilf(t, connPool, "conn is nil after Connect")
+
+	defer connPool.Close()
+
+	stmt, err := connPool.NewPrepareStatement("SELECT NAME0, NAME1 FROM SQL_TEST WHERE NAME0=:id AND NAME1=:name;", connection_pool.ANY)
+	require.Nilf(t, err, "fail to prepare statement: %v", err)
+
+	executeReq := tarantool.NewExecutePreparedRequest(stmt)
+	unprepareReq := tarantool.NewUnprepareRequest(stmt)
+
+	resp, err := connPool.Do(executeReq.Args([]interface{}{1, "test"}), connection_pool.ANY).Get()
+	if err != nil {
+		t.Fatalf("failed to execute prepared: %v", err)
+	}
+	if resp == nil {
+		t.Fatalf("nil response")
+	}
+	if resp.Code != tarantool.OkCode {
+		t.Fatalf("failed to execute prepared: code %d", resp.Code)
+	}
+	if reflect.DeepEqual(resp.Data[0], []interface{}{1, "test"}) {
+		t.Error("Select with named arguments failed")
+	}
+	if resp.MetaData[0].FieldType != "unsigned" ||
+		resp.MetaData[0].FieldName != "NAME0" ||
+		resp.MetaData[1].FieldType != "string" ||
+		resp.MetaData[1].FieldName != "NAME1" {
+		t.Error("Wrong metadata")
+	}
+
+	resp, err = connPool.Do(unprepareReq, connection_pool.ANY).Get()
+	if err != nil {
+		t.Errorf("failed to unprepare prepared statement: %v", err)
+	}
+	if resp.Code != tarantool.OkCode {
+		t.Errorf("failed to unprepare prepared statement: code %d", resp.Code)
+	}
+
+	_, err = connPool.Do(unprepareReq, connection_pool.ANY).Get()
+	if err == nil {
+		t.Errorf("the statement must be already unprepared")
+	}
+	require.Contains(t, err.Error(), "Prepared statement with id")
+
+	_, err = connPool.Do(executeReq, connection_pool.ANY).Get()
+	if err == nil {
+		t.Errorf("the statement must be already unprepared")
+	}
+	require.Contains(t, err.Error(), "Prepared statement with id")
+}
+
 // runTestMain is a body of TestMain function
 // (see https://pkg.go.dev/testing#hdr-Main).
 // Using defer + os.Exit is not works so TestMain body

connection_pool/example_test.go

@@ -548,3 +548,28 @@ func ExampleConnectionPool_Do() {
 	// Ping Data []
 	// Ping Error <nil>
 }
+
+func ExampleConnectionPool_NewPrepareStatement() {
+	pool, err := examplePool(testRoles)
+	if err != nil {
+		fmt.Println(err)
+	}
+	defer pool.Close()
+
+	stmt, err := pool.NewPrepareStatement("SELECT 1", connection_pool.ANY)
+	if err != nil {
+		fmt.Println(err)
+	}
+
+	executeReq := tarantool.NewExecutePreparedRequest(stmt)
+	unprepareReq := tarantool.NewUnprepareRequest(stmt)
+
+	_, err = pool.Do(executeReq, connection_pool.ANY).Get()
+	if err != nil {
+		fmt.Printf("Failed to execute prepared stmt")
+	}
+	_, err = pool.Do(unprepareReq, connection_pool.ANY).Get()
+	if err != nil {
+		fmt.Printf("Failed to prepare")
+	}
+}

connector.go

@@ -30,6 +30,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 +42,7 @@ 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
 
 	Do(req Request) (fut *Future)
 }

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,43 @@ 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_Do() {
+	// 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",
+	}
+	conn, err := tarantool.Connect(server, opts)
+	if err != nil {
+		fmt.Printf("Failed to connect: %s", err.Error())
+	}
+
+	stmt, err := conn.NewPrepareStatement("SELECT 1")
+	if err != nil {
+		fmt.Printf("Failed to connect: %s", err.Error())
+	}
+
+	executeReq := tarantool.NewExecutePreparedRequest(stmt)
+	unprepareReq := tarantool.NewUnprepareRequest(stmt)
+
+	_, err = conn.Do(executeReq).Get()
+	if err != nil {
+		fmt.Printf("Failed to execute prepared stmt")
+	}
+
+	_, err = conn.Do(unprepareReq).Get()
+	if err != nil {
+		fmt.Printf("Failed to prepare")
+	}
+}

future.go

@@ -239,3 +239,16 @@ func (fut *Future) Err() error {
 	fut.wait()
 	return fut.err
 }
+
+func (fut *Future) GetPreparedStatement(conn *Connection) (*PreparedStatement, error) {
+	resp, err := fut.Get()
+	if err != nil {
+		return nil, err
+	}
+	return &PreparedStatement{
+		StatementID: PreparedStatementID(resp.StmtID),
+		ParamCount:  resp.BindCount,
+		Conn:        conn,
+		MetaData:    resp.MetaData,
+	}, nil
+}

multi/multi.go

@@ -419,6 +419,11 @@ func (connMulti *ConnectionMulti) EvalTyped(expr string, args interface{}, resul
 	return connMulti.getCurrentConnection().EvalTyped(expr, args, result)
 }
 
+// ExecuteTyped passes sql expression to Tarantool 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)
@@ -482,6 +487,16 @@ func (connMulti *ConnectionMulti) EvalAsync(expr string, args interface{}) *tara
 	return connMulti.getCurrentConnection().EvalAsync(expr, args)
 }
 
+// ExecuteAsync passes sql expression to Tarantool for execution.
+func (connMulti *ConnectionMulti) ExecuteAsync(expr string, args interface{}) *tarantool.Future {
+	return connMulti.getCurrentConnection().ExecuteAsync(expr, args)
+}
+
+// NewPrepareStatement passes a sql statement to Tarantool for preparation synchronously.
+func (connMulti *ConnectionMulti) NewPrepareStatement(expr string) (*tarantool.PreparedStatement, error) {
+	return connMulti.getCurrentConnection().NewPrepareStatement(expr)
+}
+
 // Do sends the request and returns a future.
 func (connMulti *ConnectionMulti) Do(req tarantool.Request) *tarantool.Future {
 	return connMulti.getCurrentConnection().Do(req)