sql: add support of prepared statements

vr009 · vr009 · commit 719ab2cbc0b5 · 2022-05-30T13:14:35.000+03:00
This patch adds the support of prepared statements. Changed the API of method Execute, the first argument may be either string or uint64. Added connectors interface methods in connector.go. Added new IPROTO-constants for support of prepared statements in const.go. Updated multi-package for corresponding to API of connector interface. Added Benchmarks for SQL-select prepared statement. Updated CHANGELOG.md. Follows up #62 Closes #117
diff --git a/CHANGELOG.md b/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
 
diff --git a/connector.go b/connector.go
@@ -17,7 +17,8 @@ type Connector interface {
 	Call(functionName string, args interface{}) (resp *Response, err error)
 	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)
+	Execute(expr interface{}, args interface{}) (resp *Response, err error)
+	Prepare(expr string) (resp *Response, 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 interface{}, 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 interface{}, args interface{}) *Future
+	PrepareAsync(expr string) *Future
 }
diff --git a/const.go b/const.go
@@ -12,6 +12,7 @@ const (
 	UpsertRequest    = 9
 	Call17Request    = 10
 	ExecuteRequest   = 11
+	PrepareRequest   = 13
 	PingRequest      = 64
 	SubscribeRequest = 66
 
@@ -34,6 +35,7 @@ const (
 	KeySQLText      = 0x40
 	KeySQLBind      = 0x41
 	KeySQLInfo      = 0x42
+	KeyStmtID       = 0x43
 
 	KeyFieldName               = 0x00
 	KeyFieldType               = 0x01
diff --git a/multi/multi.go b/multi/multi.go
@@ -343,10 +343,14 @@ func (connMulti *ConnectionMulti) Eval(expr string, args interface{}) (resp *tar
 // Execute passes sql expression to Tarantool for execution.
 //
 // Since 1.6.0
-func (connMulti *ConnectionMulti) Execute(expr string, args interface{}) (resp *tarantool.Response, err error) {
+func (connMulti *ConnectionMulti) Execute(expr interface{}, args interface{}) (resp *tarantool.Response, err error) {
 	return connMulti.getCurrentConnection().Execute(expr, args)
 }
 
+func (connMulti *ConnectionMulti) Prepare(expr string) (resp *tarantool.Response, 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 +405,11 @@ func (connMulti *ConnectionMulti) EvalTyped(expr string, args interface{}, resul
 	return connMulti.getCurrentConnection().EvalTyped(expr, args, result)
 }
 
+// ExecuteTyped passes SQL expression for execution.
+func (connMulti *ConnectionMulti) ExecuteTyped(expr interface{}, 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 +463,13 @@ 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 Lua expression for evaluation.
+func (connMulti *ConnectionMulti) ExecuteAsync(expr interface{}, args interface{}) *tarantool.Future {
+	return connMulti.getCurrentConnection().ExecuteAsync(expr, args)
+}
+
+// PrepareAsync passes SQL statement to prepare.
+func (connMulti *ConnectionMulti) PrepareAsync(expr string) *tarantool.Future {
+	return connMulti.getCurrentConnection().PrepareAsync(expr)
+}
diff --git a/request.go b/request.go
@@ -127,10 +127,18 @@ func (conn *Connection) Eval(expr string, args interface{}) (resp *Response, err
 //
 // It is equal to conn.ExecuteAsync(expr, args).Get().
 // Since 1.6.0
-func (conn *Connection) Execute(expr string, args interface{}) (resp *Response, err error) {
+func (conn *Connection) Execute(expr interface{}, args interface{}) (resp *Response, err error) {
 	return conn.ExecuteAsync(expr, args).Get()
 }
 
+// Prepare sends a sql expression to prepare.
+//
+// It is equal to conn.PrepareAsync(expr, args).Get().
+// Since 1.6.0
+func (conn *Connection) Prepare(expr string) (resp *Response, err error) {
+	return conn.PrepareAsync(expr).Get()
+}
+
 // single used for conn.GetTyped for decode one tuple.
 type single struct {
 	res   interface{}
@@ -227,7 +235,7 @@ func (conn *Connection) EvalTyped(expr string, args interface{}, result interfac
 //
 // In addition to error returns sql info and columns meta data
 // Since 1.6.0
-func (conn *Connection) ExecuteTyped(expr string, args interface{}, result interface{}) (SQLInfo, []ColumnMetaData, error) {
+func (conn *Connection) ExecuteTyped(expr interface{}, args interface{}, result interface{}) (SQLInfo, []ColumnMetaData, error) {
 	fut := conn.ExecuteAsync(expr, args)
 	err := fut.GetTyped(&result)
 	return fut.resp.SQLInfo, fut.resp.MetaData, err
@@ -369,17 +377,44 @@ func (conn *Connection) EvalAsync(expr string, args interface{}) *Future {
 
 // ExecuteAsync sends a sql expression for execution and returns Future.
 // Since 1.6.0
-func (conn *Connection) ExecuteAsync(expr string, args interface{}) *Future {
+func (conn *Connection) ExecuteAsync(expr interface{}, args interface{}) *Future {
 	future := conn.newFuture(ExecuteRequest)
+	var stmtID uint64
+	exprStr, ok := expr.(string)
+	if !ok {
+		stmtID, ok = expr.(uint64)
+		if !ok {
+			err := errors.New("expr argument must be either string or uint64")
+			return NewErrorFuture(err)
+		}
+		return future.send(conn, func(enc *msgpack.Encoder) error {
+			enc.EncodeMapLen(2)
+			enc.EncodeUint64(KeyStmtID)
+			enc.EncodeUint64(stmtID)
+			enc.EncodeUint64(KeySQLBind)
+			return encodeSQLBind(enc, args)
+		})
+	}
 	return future.send(conn, func(enc *msgpack.Encoder) error {
 		enc.EncodeMapLen(2)
 		enc.EncodeUint64(KeySQLText)
-		enc.EncodeString(expr)
+		enc.EncodeString(exprStr)
 		enc.EncodeUint64(KeySQLBind)
 		return encodeSQLBind(enc, args)
 	})
 }
 
+// PrepareAsync sends a sql expression to prepare and returns Future.
+// Since 1.6.0
+func (conn *Connection) PrepareAsync(expr string) *Future {
+	future := conn.newFuture(PrepareRequest)
+	return future.send(conn, func(enc *msgpack.Encoder) error {
+		enc.EncodeMapLen(1)
+		enc.EncodeUint64(KeySQLText)
+		return enc.EncodeString(expr)
+	})
+}
+
 // KeyValueBind is a type for encoding named SQL parameters
 type KeyValueBind struct {
 	Key   string
diff --git a/response.go b/response.go
@@ -14,6 +14,7 @@ type Response struct {
 	Data     []interface{}
 	MetaData []ColumnMetaData
 	SQLInfo  SQLInfo
+	StmtID   uint64
 	buf      smallBuf
 }
 
@@ -178,6 +179,10 @@ 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
+				}
 			default:
 				if err = d.Skip(); err != nil {
 					return err
diff --git a/tarantool_test.go b/tarantool_test.go
@@ -450,6 +450,49 @@ func BenchmarkSQLParallel(b *testing.B) {
 	})
 }
 
+func BenchmarkSQLPreparedParallel(b *testing.B) {
+	// Tarantool supports SQL since version 2.0.0
+	isLess, err := test_helpers.IsTarantoolVersionLess(2, 0, 0)
+	if err != nil {
+		b.Fatal("Could not check the Tarantool version")
+	}
+	if isLess {
+		b.Skip()
+	}
+
+	conn, err := Connect(server, opts)
+	if err != nil {
+		b.Errorf("No connection available")
+		return
+	}
+	defer conn.Close()
+
+	spaceNo := 519
+	_, err = conn.Replace(spaceNo, []interface{}{uint(1111), "hello", "world"})
+	if err != nil {
+		b.Errorf("No connection available")
+	}
+
+	res, err := conn.Prepare("SELECT NAME0,NAME1,NAME2 FROM SQL_TEST WHERE NAME0=?")
+	if err != nil {
+		b.Fatalf("failed to prepare statement: %s", err)
+	}
+	if res.Code != 0 {
+		b.Fatalf("failed to prepare statement: %s", err)
+	}
+
+	b.ResetTimer()
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			_, err := conn.Execute(res.StmtID, []interface{}{uint(1111)})
+			if err != nil {
+				b.Errorf("Select failed: %s", err.Error())
+				break
+			}
+		}
+	})
+}
+
 func BenchmarkSQLSerial(b *testing.B) {
 	// Tarantool supports SQL since version 2.0.0
 	isLess, err := test_helpers.IsTarantoolVersionLess(2, 0, 0)
@@ -483,6 +526,47 @@ func BenchmarkSQLSerial(b *testing.B) {
 	}
 }
 
+func BenchmarkSQLPreparedSerial(b *testing.B) {
+	// Tarantool supports SQL since version 2.0.0
+	isLess, err := test_helpers.IsTarantoolVersionLess(2, 0, 0)
+	if err != nil {
+		b.Fatal("Could not check the Tarantool version")
+	}
+	if isLess {
+		b.Skip()
+	}
+
+	conn, err := Connect(server, opts)
+	if err != nil {
+		b.Errorf("Failed to connect: %s", err)
+		return
+	}
+	defer conn.Close()
+
+	spaceNo := 519
+	_, err = conn.Replace(spaceNo, []interface{}{uint(1111), "hello", "world"})
+	if err != nil {
+		b.Errorf("Failed to replace: %s", err)
+	}
+
+	res, err := conn.Prepare("SELECT NAME0,NAME1,NAME2 FROM SQL_TEST WHERE NAME0=?")
+	if err != nil {
+		b.Fatalf("failed to prepare statement: %s", err)
+	}
+	if res.Code != 0 {
+		b.Fatalf("failed to prepare statement: %s", err)
+	}
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_, err := conn.Execute(res.StmtID, []interface{}{uint(1111)})
+		if err != nil {
+			b.Errorf("Select failed: %s", err.Error())
+			break
+		}
+	}
+}
+
 ///////////////////
 
 func TestClient(t *testing.T) {
@@ -1001,6 +1085,41 @@ func TestSQL(t *testing.T) {
 	}
 }
 
+func TestPreparedStmt(t *testing.T) {
+	// Tarantool supports SQL since version 2.0.0
+	isLess, err := test_helpers.IsTarantoolVersionLess(2, 0, 0)
+	if err != nil {
+		t.Fatal("Could not check the Tarantool version")
+	}
+	if isLess {
+		t.Skip()
+	}
+
+	var conn *Connection
+
+	conn, err = Connect(server, opts)
+	if err != nil {
+		t.Fatalf("Failed to connect: %s", err.Error())
+	}
+	if conn == nil {
+		t.Fatal("conn is nil after Connect")
+	}
+	defer conn.Close()
+
+	res, err := conn.Prepare(selectPosQuery2)
+	if err != nil {
+		t.Fatalf("failed to prepare request: %s", err)
+	}
+
+	res2, err := conn.Execute(res.StmtID, []interface{}{1, "test"})
+	if err != nil {
+		t.Fatalf("failed to execute with prepared statement: %s", err)
+	}
+	if res2.Code != 0 {
+		t.Fatalf("Failed to execute: code error %d", res.Code)
+	}
+}
+
 func TestSQLTyped(t *testing.T) {
 	// Tarantool supports SQL since version 2.0.0
 	isLess, err := test_helpers.IsTarantoolVersionLess(2, 0, 0)
