sql: introduce connExecutor, the query execution orchestrator

Release note: None

The connExecutor is the top dog that interfaces with a pgwire connection
(through the clientComm and CommandResult interfaces), consumes a stream
of queries and produces a stream of results. It encapsulates the
connection state machine for which it produces events. It interfaces
with the two SQL execution engines for actually running queries. Its
main responsibility is to dispatch statements based on the current state
(in txn, not in txn, in aborted txn, etc) and to handle execution in all
states but Open (in Open it talks to an execution engine). It also
handles other commands than executiong queries: preparing, binding, etc
and it maintains the session state associated with prepares statements.

Author: andreimatei

pkg/sql/advancecode_string.go

@@ -164,7 +164,7 @@ func (a *appStats) getStrForStmt(stmt Statement) string {
 }
 
 // sqlStats carries per-application statistics for all applications on
-// each node. It hangs off Executor.
+// each node.
 type sqlStats struct {
 	st *cluster.Settings
 	syncutil.Mutex

pkg/sql/app_stats.go

@@ -0,0 +1,28 @@
+// Copyright 2017 The Cockroach Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied. See the License for the specific language governing
+// permissions and limitations under the License.
+
+package sql
+
+import (
+	"context"
+
+	"github.com/cockroachdb/cockroach/pkg/util/fsm"
+)
+
+// StatementFilter2 is the type of callback that
+// ExecutorTestingKnobs.StatementFilter takes.
+// If the callback returns an error, the session is terminated.
+//
+// WIP(andrei): rename
+type StatementFilter2 func(ctx context.Context, query string, ev fsm.Event) error

pkg/sql/conn_executor.go

@@ -22,32 +22,60 @@ package sql
 
 import (
 	"context"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/storage/engine/enginepb"
 	// We dot-import fsm to use common names such as fsm.True/False. State machine
 	// implementations using that library are weird beasts intimately inter-twined
 	// with that package; therefor this file should stay as small as possible.
 	. "github.com/cockroachdb/cockroach/pkg/util/fsm"
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 )
 
 /// States.
 
 type stateNoTxn struct{}
+
+func (stateNoTxn) String() string {
+	return "NoTxn"
+}
+
 type stateOpen struct {
 	ImplicitTxn Bool
 	// RetryIntent, if set, means the user declared the intention to retry the txn
 	// in case of retriable errors by running a SAVEPOINT cockroach_restart. The
 	// txn will enter a RestartWait state in case of such errors.
 	RetryIntent Bool
 }
+
+func (stateOpen) String() string {
+	return "Open"
+}
+
 type stateAborted struct {
 	// RetryIntent carries over the setting from stateOpen, in case we move back
 	// to Open.
 	RetryIntent Bool
 }
+
+func (stateAborted) String() string {
+	return "Aborted"
+}
+
 type stateRestartWait struct{}
+
+func (stateRestartWait) String() string {
+	return "RestartWait"
+}
+
 type stateCommitWait struct{}
 
+func (stateCommitWait) String() string {
+	return "CommitWait"
+}
+
 func (stateNoTxn) State()       {}
 func (stateOpen) State()        {}
 func (stateAborted) State()     {}
@@ -60,13 +88,51 @@ type eventTxnStart struct {
 	ImplicitTxn Bool
 }
 type eventTxnStartPayload struct {
-	tranCtx transitionCtx
+	iso enginepb.IsolationType
+	pri roachpb.UserPriority
+	// txnSQLTimestamp is the timestamp that statements executed in the
+	// transaction that is started by this event will report for now(),
+	// current_timestamp(), transaction_timestamp().
+	txnSQLTimestamp time.Time
+	tranCtx         transitionCtx
+	readOnly        tree.ReadWriteMode
+}
+
+func makeEventTxnStartPayload(
+	iso enginepb.IsolationType,
+	pri roachpb.UserPriority,
+	txnSQLTimestamp time.Time,
+	tranCtx transitionCtx,
+	readOnly tree.ReadWriteMode,
+) eventTxnStartPayload {
+	return eventTxnStartPayload{
+		iso:             iso,
+		pri:             pri,
+		readOnly:        readOnly,
+		txnSQLTimestamp: txnSQLTimestamp,
+		tranCtx:         tranCtx,
+	}
 }
 
 // eventRetryIntentSet is generated in the Open state when a SAVEPOINT
 // cockroach_restart is seen.
 type eventRetryIntentSet struct{}
 type eventTxnFinish struct{}
+
+// eventTxnFinishPayload represents the payload for eventTxnFinish.
+type eventTxnFinishPayload struct {
+	// commit is set if the transaction committed, false if it was aborted.
+	commit bool
+}
+
+// toEvent turns the eventTxnFinishPayload into a txnEvent.
+func (e eventTxnFinishPayload) toEvent() txnEvent {
+	if e.commit {
+		return txnCommit
+	}
+	return txnAborted
+}
+
 type eventTxnRestart struct{}
 
 type eventNonRetriableErr struct {
@@ -148,7 +214,7 @@ var TxnStateTransitions = Compile(Pattern{
 			Action: func(args Args) error {
 				return args.Extended.(*txnState2).noTxnToOpen(
 					args.Ctx, args.Event.(eventTxnStart),
-					args.Payload.(eventTxnStartPayload).tranCtx)
+					args.Payload.(eventTxnStartPayload))
 			},
 		},
 	},
@@ -161,7 +227,8 @@ var TxnStateTransitions = Compile(Pattern{
 			Action: func(args Args) error {
 				ts := args.Extended.(*txnState2)
 				ts.finishSQLTxn(args.Ctx)
-				ts.setAdvanceInfo(advanceInfo{code: advanceOne, flush: true})
+				ts.setAdvanceInfo(
+					advanceOne, flush, noRewind, args.Payload.(eventTxnFinishPayload).toEvent())
 				return nil
 			},
 		},
@@ -190,10 +257,10 @@ var TxnStateTransitions = Compile(Pattern{
 			Action: func(args Args) error {
 				// The caller will call rewCap.rewindAndUnlock().
 				args.Extended.(*txnState2).setAdvanceInfo(
-					advanceInfo{
-						code:   rewind,
-						rewCap: args.Payload.(eventRetriableErrPayload).rewCap,
-						flush:  false})
+					rewind,
+					noFlush,
+					args.Payload.(eventRetriableErrPayload).rewCap,
+					txnRestart)
 				return nil
 			},
 		},
@@ -216,7 +283,7 @@ var TxnStateTransitions = Compile(Pattern{
 			Action: func(args Args) error {
 				ts := args.Extended.(*txnState2)
 				ts.mu.txn.CleanupOnError(ts.Ctx, args.Payload.(payloadWithError).errorCause())
-				ts.setAdvanceInfo(advanceInfo{code: skipQueryStr, flush: true})
+				ts.setAdvanceInfo(skipBatch, flush, noRewind, txnAborted)
 				return nil
 			},
 		},
@@ -230,7 +297,7 @@ var TxnStateTransitions = Compile(Pattern{
 			Action: func(args Args) error {
 				ts := args.Extended.(*txnState2)
 				ts.mu.txn.CleanupOnError(ts.Ctx, args.Payload.(payloadWithError).errorCause())
-				ts.setAdvanceInfo(advanceInfo{code: skipQueryStr, flush: true})
+				ts.setAdvanceInfo(skipBatch, flush, noRewind, txnAborted)
 				return nil
 			},
 		},
@@ -241,7 +308,7 @@ var TxnStateTransitions = Compile(Pattern{
 			Action: func(args Args) error {
 				// We flush after setting the retry intent; we know what statement
 				// caused this event and we don't need to rewind past it.
-				args.Extended.(*txnState2).setAdvanceInfo(advanceInfo{code: advanceOne, flush: true})
+				args.Extended.(*txnState2).setAdvanceInfo(advanceOne, flush, noRewind, noEvent)
 				return nil
 			},
 		},
@@ -252,15 +319,36 @@ var TxnStateTransitions = Compile(Pattern{
 			Action: func(args Args) error {
 				// Note: Preparing the KV txn for restart has already happened by this
 				// point.
-				args.Extended.(*txnState2).setAdvanceInfo(advanceInfo{code: skipQueryStr, flush: true})
+				args.Extended.(*txnState2).setAdvanceInfo(skipBatch, flush, noRewind, txnRestart)
 				return nil
 			},
 		},
 		eventTxnReleased{}: {
 			Description: "RELEASE SAVEPOINT cockroach_restart",
 			Next:        stateCommitWait{},
 			Action: func(args Args) error {
-				args.Extended.(*txnState2).setAdvanceInfo(advanceInfo{code: advanceOne, flush: true})
+				args.Extended.(*txnState2).setAdvanceInfo(advanceOne, flush, noRewind, txnCommit)
+				return nil
+			},
+		},
+		// ROLLBACK TO SAVEPOINT
+		eventTxnRestart{}: {
+			Description: "ROLLBACK TO SAVEPOINT cockroach_restart",
+			Next:        stateOpen{ImplicitTxn: False, RetryIntent: True},
+			Action: func(args Args) error {
+				state := args.Extended.(*txnState2)
+				// If commands have already been sent through the transaction, restart
+				// the client txn's proto to increment the epoch.
+				if state.mu.txn.GetTxnCoordMeta().CommandCount > 0 {
+					// NOTE: We don't bump the txn timestamp on this restart. If this
+					// we generally supported savepoints and one would issue a rollback to
+					// a regular savepoint, clearly we couldn't bump the timestamp. But in
+					// the special case of the cockroach_restart savepoint, it's not clear
+					// to me what a user's expectation might be.
+					state.mu.txn.Proto().Restart(
+						0 /* userPriority */, 0 /* upgradePriority */, hlc.Timestamp{})
+				}
+				args.Extended.(*txnState2).setAdvanceInfo(advanceOne, flush, noRewind, noEvent)
 				return nil
 			},
 		},
@@ -277,7 +365,8 @@ var TxnStateTransitions = Compile(Pattern{
 			Action: func(args Args) error {
 				ts := args.Extended.(*txnState2)
 				ts.finishSQLTxn(ts.Ctx)
-				ts.setAdvanceInfo(advanceInfo{code: advanceOne, flush: true})
+				ts.setAdvanceInfo(
+					advanceOne, flush, noRewind, args.Payload.(eventTxnFinishPayload).toEvent())
 				return nil
 			},
 		},
@@ -293,13 +382,15 @@ var TxnStateTransitions = Compile(Pattern{
 				ts := args.Extended.(*txnState2)
 				ts.finishSQLTxn(args.Ctx)
 
+				payload := args.Payload.(eventTxnStartPayload)
+
 				ts.resetForNewSQLTxn(
 					args.Ctx,
 					explicitTxn,
-					ts.sqlTimestamp, ts.isolation, ts.priority,
+					payload.txnSQLTimestamp, payload.iso, payload.pri, payload.readOnly,
 					args.Payload.(eventTxnStartPayload).tranCtx,
 				)
-				ts.setAdvanceInfo(advanceInfo{code: advanceOne, flush: true})
+				ts.setAdvanceInfo(advanceOne, flush, noRewind, noEvent)
 				return nil
 			},
 		},
@@ -313,7 +404,8 @@ var TxnStateTransitions = Compile(Pattern{
 			Action: func(args Args) error {
 				ts := args.Extended.(*txnState2)
 				ts.finishSQLTxn(args.Ctx)
-				ts.setAdvanceInfo(advanceInfo{code: advanceOne, flush: true})
+				ts.setAdvanceInfo(
+					advanceOne, flush, noRewind, args.Payload.(eventTxnFinishPayload).toEvent())
 				return nil
 			},
 		},
@@ -322,7 +414,7 @@ var TxnStateTransitions = Compile(Pattern{
 			Description: "ROLLBACK TO SAVEPOINT cockroach_restart",
 			Next:        stateOpen{ImplicitTxn: False, RetryIntent: True},
 			Action: func(args Args) error {
-				args.Extended.(*txnState2).setAdvanceInfo(advanceInfo{code: advanceOne, flush: true})
+				args.Extended.(*txnState2).setAdvanceInfo(advanceOne, flush, noRewind, noEvent)
 				return nil
 			},
 		},
@@ -332,7 +424,7 @@ var TxnStateTransitions = Compile(Pattern{
 				ts := args.Extended.(*txnState2)
 				ts.mu.txn.CleanupOnError(ts.Ctx, args.Payload.(eventNonRetriableErrPayload).err)
 				ts.mu.txn = nil
-				ts.setAdvanceInfo(advanceInfo{code: skipQueryStr, flush: true})
+				ts.setAdvanceInfo(skipBatch, flush, noRewind, noEvent)
 				return nil
 			},
 		},
@@ -345,7 +437,8 @@ var TxnStateTransitions = Compile(Pattern{
 			Action: func(args Args) error {
 				ts := args.Extended.(*txnState2)
 				ts.finishSQLTxn(args.Ctx)
-				ts.setAdvanceInfo(advanceInfo{code: advanceOne, flush: true})
+				ts.setAdvanceInfo(
+					advanceOne, flush, noRewind, args.Payload.(eventTxnFinishPayload).toEvent())
 				return nil
 			},
 		},
@@ -355,7 +448,7 @@ var TxnStateTransitions = Compile(Pattern{
 			Description: "any other statement",
 			Next:        stateAborted{RetryIntent: True},
 			Action: func(args Args) error {
-				args.Extended.(*txnState2).setAdvanceInfo(advanceInfo{code: skipQueryStr, flush: true})
+				args.Extended.(*txnState2).setAdvanceInfo(skipBatch, flush, noRewind, noEvent)
 				return nil
 			},
 		},
@@ -367,27 +460,32 @@ func cleanupAndFinish(args Args) error {
 	ts := args.Extended.(*txnState2)
 	ts.mu.txn.CleanupOnError(ts.Ctx, args.Payload.(payloadWithError).errorCause())
 	ts.finishSQLTxn(args.Ctx)
-	ts.setAdvanceInfo(advanceInfo{code: skipQueryStr, flush: true})
+	ts.setAdvanceInfo(skipBatch, flush, noRewind, txnAborted)
 	return nil
 }
 
 // noTxnToOpen implements the side effects of starting a txn. It also calls
-// setAdvanceInfo(), telling the execution to stayInPlace.
+// setAdvanceInfo().
 func (ts *txnState2) noTxnToOpen(
-	connCtx context.Context, ev eventTxnStart, tranCtx transitionCtx,
+	connCtx context.Context, ev eventTxnStart, payload eventTxnStartPayload,
 ) error {
 	txnTyp := explicitTxn
+	advCode := advanceOne
 	if ev.ImplicitTxn.Get() {
 		txnTyp = implicitTxn
+		// For an implicit txn, we want the statement that produced the event to be
+		// executed again (this time in state Open).
+		advCode = stayInPlace
 	}
 
 	ts.resetForNewSQLTxn(
 		connCtx,
 		txnTyp,
-		tranCtx.clock.PhysicalTime(), /* sqlTimestamp */
-		tranCtx.defaultIsolationLevel,
-		roachpb.NormalUserPriority,
-		tranCtx,
+		payload.txnSQLTimestamp,
+		payload.iso,
+		payload.pri,
+		payload.readOnly,
+		payload.tranCtx,
 	)
 	// When starting a transaction from the NoTxn state, we don't advance the
 	// cursor - we want the same statement to be executed again. Note that this is
@@ -399,6 +497,6 @@ func (ts *txnState2) noTxnToOpen(
 	// flushed (since we're not in a txn), and we shouldn't have generated any
 	// results for the statement that generated this event (which is why we can
 	// stayInPlace here).
-	ts.setAdvanceInfo(advanceInfo{code: stayInPlace, flush: true})
+	ts.setAdvanceInfo(advCode, flush, noRewind, noEvent)
 	return nil
 }