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 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
 	}
 }
+
+// Prepare passes a sql statement to Tarantool for preparation synchronously.
+func (conn *Connection) Prepare(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
 }
+
+// Prepare passes a sql statement to Tarantool for preparation synchronously.
+func (connPool *ConnectionPool) Prepare(expr string, userMode Mode) (*tarantool.PreparedStatement, error) {
+	conn, err := connPool.getNextConnection(userMode)
+	if err != nil {
+		return nil, err
+	}
+	return conn.Prepare(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,65 @@ func TestDo(t *testing.T) {
 	require.NotNilf(t, resp, "response is nil after Ping")
 }
 
+func TestPrepare(t *testing.T) {
+	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.Prepare("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_Prepare() {
+	pool, err := examplePool(testRoles)
+	if err != nil {
+		fmt.Println(err)
+	}
+	defer pool.Close()
+
+	stmt, err := pool.Prepare("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,9 @@ 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
+
+	Prepare(expr string) (*PreparedStatement, error)
 
 	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.Prepare("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)
+}
+
+// Prepare passes a sql statement to Tarantool for preparation synchronously.
+func (connMulti *ConnectionMulti) Prepare(expr string) (*tarantool.PreparedStatement, error) {
+	return connMulti.getCurrentConnection().Prepare(expr)
+}
+
 // Do sends the request and returns a future.
 func (connMulti *ConnectionMulti) Do(req tarantool.Request) *tarantool.Future {
 	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)
 }
@@ -389,6 +409,27 @@ func (conn *Connection) ExecuteAsync(expr string, args interface{}) *Future {
 	return conn.Do(req)
 }
 
+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
+}
+
+func newPreparedStatement(conn *Connection, resp *Response) *PreparedStatement {
+	stmt := new(PreparedStatement)
+	stmt.Conn = conn
+	stmt.ParamCount = resp.BindCount
+	stmt.MetaData = resp.MetaData
+	stmt.StatementID = PreparedStatementID(resp.StmtID)
+	return stmt
+}
+
 // KeyValueBind is a type for encoding named SQL parameters
 type KeyValueBind struct {
 	Key   string
@@ -981,3 +1022,80 @@ func (req *ExecuteRequest) Args(args interface{}) *ExecuteRequest {
 func (req *ExecuteRequest) Body(res SchemaResolver, enc *msgpack.Encoder) error {
 	return fillExecute(enc, req.expr, req.args)
 }
+
+// StickyRequest is an interface that provides the info about a Connection
+// the request belongs to.
+type StickyRequest interface {
+	Conn() *Connection
+}
+
+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 {
+	StickyRequest
+	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
+}
+
+func (req *UnprepareRequest) Conn() *Connection {
+	return req.stmt.Conn
+}
+
+// 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 ExecutePreparedRequest struct {
+	StickyRequest
+	baseRequest
+	stmt *PreparedStatement
+	args interface{}
+}
+
+// NewPreparedExecuteRequest returns a new empty preparedExecuteRequest.
+func NewExecutePreparedRequest(stmt *PreparedStatement) *ExecutePreparedRequest {
+	req := new(ExecutePreparedRequest)
+	req.requestCode = ExecuteRequestCode
+	req.stmt = stmt
+	req.args = []interface{}{}
+	return req
+}
+
+func (req *ExecutePreparedRequest) Conn() *Connection {
+	return req.stmt.Conn
+}
+
+func (req *ExecutePreparedRequest) Args(args interface{}) *ExecutePreparedRequest {
+	req.args = args
+	return req
+}
+
+// Body fills an encoder with the execute request body.
+func (req *ExecutePreparedRequest) Body(res SchemaResolver, enc *msgpack.Encoder) error {
+	return fillPreparedExecute(enc, *req.stmt, req.args)
+}

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

tarantool_test.go

@@ -2,6 +2,7 @@ package tarantool_test
 
 import (
 	"fmt"
+	"github.com/stretchr/testify/require"
 	"log"
 	"os"
 	"reflect"
@@ -59,9 +60,9 @@ var spaceName = "test"
 var indexNo = uint32(0)
 var indexName = "primary"
 var opts = Opts{
-	Timeout: 500 * time.Millisecond,
-	User:    "test",
-	Pass:    "test",
+	//Timeout: 500 * time.Millisecond,
+	User: "test",
+	Pass: "test",
 	//Concurrency: 32,
 	//RateLimit: 4*1024,
 }
@@ -169,6 +170,50 @@ func BenchmarkClientSerialSQL(b *testing.B) {
 	}
 }
 
+func BenchmarkClientSerialSQLPrepared(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)
+	}
+
+	stmt, err := conn.Prepare("SELECT NAME0,NAME1,NAME2 FROM SQL_TEST WHERE NAME0=?")
+	if err != nil {
+		b.Fatalf("failed to prepare a SQL statement")
+	}
+	executeReq := NewExecutePreparedRequest(stmt)
+	unprepareReq := NewUnprepareRequest(stmt)
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_, err := conn.Do(executeReq.Args([]interface{}{uint(1111)})).Get()
+		if err != nil {
+			b.Errorf("Select failed: %s", err.Error())
+			break
+		}
+	}
+	_, err = conn.Do(unprepareReq).Get()
+	if err != nil {
+		b.Fatalf("failed to unprepare a SQL statement")
+	}
+}
+
 func BenchmarkClientFuture(b *testing.B) {
 	var err error
 
@@ -462,6 +507,48 @@ func BenchmarkClientParallelSQL(b *testing.B) {
 	})
 }
 
+func BenchmarkClientParallelSQLPrepared(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 := test_helpers.ConnectWithValidation(b, server, opts)
+	defer conn.Close()
+
+	spaceNo := 519
+	_, err = conn.Replace(spaceNo, []interface{}{uint(1111), "hello", "world"})
+	if err != nil {
+		b.Errorf("No connection available")
+	}
+
+	stmt, err := conn.Prepare("SELECT NAME0,NAME1,NAME2 FROM SQL_TEST WHERE NAME0=?")
+	if err != nil {
+		b.Fatalf("failed to prepare a SQL statement")
+	}
+	executeReq := NewExecutePreparedRequest(stmt)
+	unprepareReq := NewUnprepareRequest(stmt)
+
+	b.ResetTimer()
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			_, err := conn.Do(executeReq.Args([]interface{}{uint(1111)})).Get()
+			if err != nil {
+				b.Errorf("Select failed: %s", err.Error())
+				break
+			}
+		}
+	})
+	_, err = conn.Do(unprepareReq).Get()
+	if err != nil {
+		b.Fatalf("failed to unprepare a SQL statement")
+	}
+}
+
 func BenchmarkSQLSerial(b *testing.B) {
 	// Tarantool supports SQL since version 2.0.0
 	isLess, err := test_helpers.IsTarantoolVersionLess(2, 0, 0)
@@ -1341,6 +1428,58 @@ func TestStressSQL(t *testing.T) {
 	}
 }
 
+func TestPrepare(t *testing.T) {
+	var err error
+
+	conn := test_helpers.ConnectWithValidation(t, server, opts)
+	defer conn.Close()
+
+	stmt, err := conn.Prepare(selectNamedQuery2)
+	if err != nil {
+		t.Errorf("failed to prepare: %v", err)
+	}
+
+	executeReq := NewExecutePreparedRequest(stmt)
+	unprepareReq := NewUnprepareRequest(stmt)
+
+	resp, err := conn.Do(executeReq.Args([]interface{}{1, "test"})).Get()
+	if err != nil {
+		t.Errorf("failed to execute prepared: %v", err)
+	}
+	if resp.Code != OkCode {
+		t.Errorf("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 = conn.Do(unprepareReq).Get()
+	if err != nil {
+		t.Errorf("failed to unprepare prepared statement: %v", err)
+	}
+	if resp.Code != OkCode {
+		t.Errorf("failed to unprepare prepared statement: code %d", resp.Code)
+	}
+
+	_, err = conn.Do(unprepareReq).Get()
+	if err == nil {
+		t.Errorf("the statement must be already unprepared")
+	}
+	require.Contains(t, err.Error(), "Prepared statement with id")
+
+	_, err = conn.Do(executeReq).Get()
+	if err == nil {
+		t.Errorf("the statement must be already unprepared")
+	}
+	require.Contains(t, err.Error(), "Prepared statement with id")
+}
+
 func TestSchema(t *testing.T) {
 	var err error