Exec: bind values; Fix 'INSERT INTO ...SELECT' parsing

clickhouse.go

@@ -168,10 +168,7 @@ func (ch *clickhouse) Rollback() error {
 	}
 	ch.data = nil
 	ch.inTransaction = false
-	if err := ch.cancel(); err != nil {
-		return err
-	}
-	return driver.ErrBadConn
+	return nil
 }
 
 func (ch *clickhouse) Close() error {

clickhouse_test.go

@@ -2,7 +2,6 @@ package clickhouse_test
 
 import (
 	"database/sql"
-	"database/sql/driver"
 	"fmt"
 	"strings"
 	"testing"
@@ -704,7 +703,7 @@ func Test_Tx(t *testing.T) {
 		if tx, err := connect.Begin(); assert.NoError(t, err) {
 			_, err = tx.Query("SELECT 1")
 			if assert.NoError(t, err) {
-				if !assert.Equal(t, driver.ErrBadConn, tx.Rollback()) {
+				if !assert.NoError(t, tx.Rollback()) {
 					return
 				}
 			}

helpers.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"database/sql/driver"
 	"fmt"
+	"regexp"
 	"strings"
 	"time"
 )
@@ -96,9 +97,11 @@ func paramParser(reader *bytes.Reader) string {
 	return name.String()
 }
 
+var selectRe = regexp.MustCompile(`\s+SELECT\s+`)
+
 func isInsert(query string) bool {
 	if f := strings.Fields(query); len(f) > 2 {
-		return strings.EqualFold("INSERT", f[0]) && strings.EqualFold("INTO", f[1]) && strings.Index(strings.ToUpper(query), " SELECT ") == -1
+		return strings.EqualFold("INSERT", f[0]) && strings.EqualFold("INTO", f[1]) && !selectRe.MatchString(strings.ToUpper(query))
 	}
 	return false
 }

stmt.go

@@ -43,7 +43,8 @@ func (stmt *stmt) execContext(ctx context.Context, args []driver.Value) (driver.
 		}
 		return emptyResult, nil
 	}
-	if err := stmt.ch.sendQuery(stmt.query); err != nil {
+
+	if err := stmt.ch.sendQuery(stmt.bind(convertOldArgs(args))); err != nil {
 		return nil, err
 	}
 	if _, err := stmt.ch.receiveData(); err != nil {
@@ -57,16 +58,33 @@ func (stmt *stmt) Query(args []driver.Value) (driver.Rows, error) {
 }
 
 func (stmt *stmt) queryContext(ctx context.Context, args []namedValue) (driver.Rows, error) {
+	if finish := stmt.ch.watchCancel(ctx); finish != nil {
+		defer finish()
+	}
+
+	if err := stmt.ch.sendQuery(stmt.bind(args)); err != nil {
+		return nil, err
+	}
+
+	rows, err := stmt.ch.receiveData()
+	if err != nil {
+		return nil, err
+	}
+
+	return rows, nil
+}
+
+func (stmt *stmt) Close() error {
+	stmt.ch.logf("[stmt] close")
+	return nil
+}
+
+func (stmt *stmt) bind(args []namedValue) string {
 	var (
 		buf     bytes.Buffer
 		index   int
 		keyword bool
 	)
-
-	if finish := stmt.ch.watchCancel(ctx); finish != nil {
-		defer finish()
-	}
-
 	switch {
 	case stmt.NumInput() != 0:
 		reader := bytes.NewReader([]byte(stmt.query))
@@ -110,22 +128,7 @@ func (stmt *stmt) queryContext(ctx context.Context, args []namedValue) (driver.R
 	default:
 		buf.WriteString(stmt.query)
 	}
-
-	if err := stmt.ch.sendQuery(buf.String()); err != nil {
-		return nil, err
-	}
-
-	rows, err := stmt.ch.receiveData()
-	if err != nil {
-		return nil, err
-	}
-
-	return rows, nil
-}
-
-func (stmt *stmt) Close() error {
-	stmt.ch.logf("[stmt] close")
-	return nil
+	return buf.String()
 }
 
 type namedValue struct {