Rollback severity seems to end the connection (#64)

* Rollback seems to end the connection

Check for ROLLBACK severity and
guard against it in Close

Fixes #61
Fixes #48

* ROLLBACK kills connection too

Ping would try to receive a message during the
session reset which won't ever happen.

Co-authored-by: huebnerr <45869321+huebnerr@users.noreply.github.com>

Author: Michael

connection.go

@@ -71,6 +71,7 @@ type connection struct {
 	scratch          [512]byte
 	sessionID        string
 	serverTZOffset   string
+	dead             bool // used if a ROLLBACK severity error is encountered
 	sessMutex        sync.Mutex
 }
 
@@ -153,6 +154,9 @@ func (v *connection) Ping(ctx context.Context) error {
 // ResetSession implements the SessionResetter interface for connection. This allows the sql
 // package to evaluate the connection state when managing the connection pool.
 func (v *connection) ResetSession(ctx context.Context) error {
+	if v.dead {
+		return driver.ErrBadConn
+	}
 	return v.Ping(ctx)
 }
 

driver_test.go

@@ -801,6 +801,30 @@ func TestConnectionClosure(t *testing.T) {
 	rows.Close()
 }
 
+func TestConcurrentStatementQuery(t *testing.T) {
+	connDB := openConnection(t, "test_stmt_ordering_threads_pre")
+	defer closeConnection(t, connDB, "test_stmt_ordering_threads_post")
+	stmt, err := connDB.PrepareContext(ctx, "SELECT a FROM stmt_thread_test")
+	assertNoErr(t, err)
+	wg := new(sync.WaitGroup)
+	wg.Add(3)
+	for i := 0; i < 3; i++ {
+		go func() {
+			defer wg.Done()
+			_, err := stmt.QueryContext(ctx)
+			assertNoErr(t, err)
+		}()
+	}
+	wg.Wait()
+}
+
+func TestInvalidDDLStatement(t *testing.T) {
+	connDB := openConnection(t)
+	defer closeConnection(t, connDB)
+	_, err := connDB.Exec("DROP VIEW DOESNOTEXISTVIEW")
+	assertErr(t, err, "does not exist")
+}
+
 var verticaUserName = flag.String("user", "dbadmin", "the user name to connect to Vertica")
 var verticaPassword = flag.String("password", os.Getenv("VERTICA_TEST_PASSWORD"), "Vertica password for this user")
 var verticaHostPort = flag.String("locator", "localhost:5433", "Vertica's host and port")

msgs/beerrormsg.go

@@ -46,7 +46,7 @@ type BEErrorMsg struct {
 	Routine  string
 }
 
-// InitFromMsgBody docs
+// CreateFromMsgBody docs
 func (b *BEErrorMsg) CreateFromMsgBody(buf *msgBuffer) (BackEndMsg, error) {
 
 	res := &BEErrorMsg{}

msgs/msg.go

@@ -36,25 +36,30 @@ import (
 	"fmt"
 )
 
+// CmdTargetType describes the target of a command
 type CmdTargetType byte
 
-var (
+// Possible command targets
+const (
 	CmdTargetTypePortal    CmdTargetType = 'P'
 	CmdTargetTypeStatement CmdTargetType = 'S'
 )
 
-// FrontEndMsg docs
+// FrontEndMsg is sent from the adapter to the database
 type FrontEndMsg interface {
 	Flatten() ([]byte, byte)
 	String() string
 }
 
-// BackEndMsg docs
+// BackEndMsg is received from the database
 type BackEndMsg interface {
 	CreateFromMsgBody(*msgBuffer) (BackEndMsg, error)
 	String() string
 }
 
+// backEndMsgTypeMap is a global map of message descriptor bytes to instances
+// of that message. The instances are not used directly, but instead are used to
+// construct new values of that message type. This is populated on init.
 var backEndMsgTypeMap = make(map[byte]BackEndMsg)
 
 func registerBackEndMsgType(msgType byte, bem BackEndMsg) {

stmt.go

@@ -72,6 +72,8 @@ type stmt struct {
 	posArgCnt    int
 	paramTypes   []common.ParameterType
 	lastRowDesc  *msgs.BERowDescMsg
+	// set if Vertica issues an error of ROLLBACK severity
+	rolledBack bool
 }
 
 func newStmt(connection *connection, command string) (*stmt, error) {
@@ -99,40 +101,41 @@ func (s *stmt) pushNamed(name string) {
 
 // Close closes this statement.
 func (s *stmt) Close() error {
-	if s.parseState == parseStateParsed {
-		closeMsg := &msgs.FECloseMsg{TargetType: msgs.CmdTargetTypeStatement, TargetName: s.preparedName}
+	if s.parseState != parseStateParsed {
+		return nil
+	}
+	if s.rolledBack {
+		s.parseState = parseStateUnparsed
+		s.conn.dead = true
+		return nil
+	}
+	closeMsg := &msgs.FECloseMsg{TargetType: msgs.CmdTargetTypeStatement, TargetName: s.preparedName}
 
-		//s.conn.lockSessionMutex()
-		//defer s.conn.unlockSessionMutex()
+	if err := s.conn.sendMessage(closeMsg); err != nil {
+		return err
+	}
 
-		if err := s.conn.sendMessage(closeMsg); err != nil {
-			return err
-		}
+	if err := s.conn.sendMessage(&msgs.FEFlushMsg{}); err != nil {
+		return err
+	}
+
+	for {
+		bMsg, err := s.conn.recvMessage()
 
-		if err := s.conn.sendMessage(&msgs.FEFlushMsg{}); err != nil {
+		if err != nil {
 			return err
 		}
 
-		for {
-			bMsg, err := s.conn.recvMessage()
-
-			if err != nil {
-				return err
-			}
-
-			switch bMsg.(type) {
-			case *msgs.BECloseCompleteMsg:
-				s.parseState = parseStateUnparsed
-				return nil
-			case *msgs.BECmdDescriptionMsg:
-				continue
-			default:
-				s.conn.defaultMessageHandler(bMsg)
-			}
+		switch bMsg.(type) {
+		case *msgs.BECloseCompleteMsg:
+			s.parseState = parseStateUnparsed
+			return nil
+		case *msgs.BECmdDescriptionMsg:
+			continue
+		default:
+			s.conn.defaultMessageHandler(bMsg)
 		}
 	}
-
-	return nil
 }
 
 // NumInput is used by database/sql to sanity check the number of arguments given
@@ -280,7 +283,7 @@ func (s *stmt) QueryContextRaw(ctx context.Context, baseArgs []driver.NamedValue
 		case *msgs.BECmdCompleteMsg:
 			break
 		case *msgs.BEErrorMsg:
-			return newEmptyRows(), msg.ToErrorType()
+			return newEmptyRows(), s.evaluateErrorMsg(msg)
 		case *msgs.BEEmptyQueryResponseMsg:
 			return newEmptyRows(), nil
 		case *msgs.BEReadyForQueryMsg, *msgs.BEPortalSuspendedMsg:
@@ -388,6 +391,13 @@ func (s *stmt) interpolate(args []driver.NamedValue) (string, error) {
 	return result, nil
 }
 
+func (s *stmt) evaluateErrorMsg(msg *msgs.BEErrorMsg) error {
+	if msg.Severity == "ROLLBACK" {
+		s.rolledBack = true
+	}
+	return msg.ToErrorType()
+}
+
 func (s *stmt) prepareAndDescribe() error {
 
 	parseMsg := &msgs.FEParseMsg{
@@ -493,7 +503,7 @@ func (s *stmt) collectResults(ctx context.Context) (*rows, error) {
 			s.lastRowDesc = msg
 			rows = newRows(ctx, s.lastRowDesc, s.conn.serverTZOffset)
 		case *msgs.BEErrorMsg:
-			return newEmptyRows(), msg.ToErrorType()
+			return newEmptyRows(), s.evaluateErrorMsg(msg)
 		case *msgs.BEEmptyQueryResponseMsg:
 			return newEmptyRows(), nil
 		case *msgs.BEBindCompleteMsg, *msgs.BECmdDescriptionMsg: