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/cli/cliflags/flags.go

@@ -690,4 +690,10 @@ in the history of the cluster.`,
 When no node ID is specified, also lists all nodes that have been decommissioned
 in the history of the cluster.`,
 	}
+
+	// !!!
+	UseFrontendV2 = FlagInfo{
+		Name:        "use-frontend-v2",
+		Description: CertsDir.Description,
+	}
 )

pkg/cli/flags.go

@@ -211,6 +211,7 @@ func init() {
 
 		VarFlag(f, &serverCfg.Stores, cliflags.Store)
 		VarFlag(f, &serverCfg.MaxOffset, cliflags.MaxOffset)
+		BoolFlag(f, &serverCfg.UseFrontendV2, cliflags.UseFrontendV2, serverCfg.UseFrontendV2)
 
 		// Usage for the unix socket is odd as we use a real file, whereas
 		// postgresql and clients consider it a directory and build a filename

pkg/gossip/gossip.go

@@ -959,6 +959,7 @@ func (g *Gossip) GetSystemConfig() (config.SystemConfig, bool) {
 // RegisterSystemConfigChannel registers a channel to signify updates for the
 // system config. It is notified after registration, and whenever a new
 // system config is successfully unmarshaled.
+// !!! comment
 func (g *Gossip) RegisterSystemConfigChannel() <-chan struct{} {
 	g.systemConfigMu.Lock()
 	defer g.systemConfigMu.Unlock()

pkg/internal/client/txn.go

@@ -797,6 +797,7 @@ func (txn *Txn) Exec(
 // book-keeping.
 //
 // TODO(andrei): I think this is called in the wrong place. See #18170.
+// !!! still used?
 func (txn *Txn) PrepareForRetry(ctx context.Context, err error) {
 	txn.commitTriggers = nil
 	log.VEventf(ctx, 2, "automatically retrying transaction: %s because of error: %s",

pkg/server/admin.go

@@ -157,7 +157,7 @@ func (s *adminServer) NewContextAndSessionForRPC(
 ) (context.Context, *sql.Session) {
 	ctx = s.server.AnnotateCtx(ctx)
 	session := sql.NewSession(
-		ctx, args, s.server.sqlExecutor, nil /* remote */, s.memMetrics, nil /* conn */)
+		ctx, args, s.server.sqlExecutor, s.memMetrics, nil /* conn */)
 	session.StartMonitor(&s.memMonitor, mon.BoundAccount{})
 	return ctx, session
 }

pkg/server/admin_test.go

@@ -240,7 +240,7 @@ func TestAdminAPIDatabases(t *testing.T) {
 	const testdb = "test"
 	session := sql.NewSession(
 		ctx, sql.SessionArgs{User: security.RootUser}, ts.sqlExecutor,
-		nil /* remote */, &sql.MemoryMetrics{}, nil /* conn */)
+		&sql.MemoryMetrics{}, nil /* conn */)
 	session.StartUnlimitedMonitor()
 	defer session.Finish(ts.sqlExecutor)
 	query := "CREATE DATABASE " + testdb
@@ -412,7 +412,7 @@ func TestAdminAPITableDetails(t *testing.T) {
 
 			session := sql.NewSession(
 				ctx, sql.SessionArgs{User: security.RootUser}, ts.sqlExecutor,
-				nil /* remote */, &sql.MemoryMetrics{}, nil /* conn */)
+				&sql.MemoryMetrics{}, nil /* conn */)
 			session.StartUnlimitedMonitor()
 			defer session.Finish(ts.sqlExecutor)
 			setupQueries := []string{
@@ -557,7 +557,7 @@ func TestAdminAPIZoneDetails(t *testing.T) {
 	defer span.Finish()
 	session := sql.NewSession(
 		ctx, sql.SessionArgs{User: security.RootUser}, ts.sqlExecutor,
-		nil /* remote */, &sql.MemoryMetrics{}, nil /* conn */)
+		&sql.MemoryMetrics{}, nil /* conn */)
 	session.StartUnlimitedMonitor()
 	setupQueries := []string{
 		"CREATE DATABASE test",
@@ -668,7 +668,7 @@ func TestAdminAPIUsers(t *testing.T) {
 	defer span.Finish()
 	session := sql.NewSession(
 		ctx, sql.SessionArgs{User: security.RootUser}, ts.sqlExecutor,
-		nil /* remote */, &sql.MemoryMetrics{}, nil /* conn */)
+		&sql.MemoryMetrics{}, nil /* conn */)
 	session.StartUnlimitedMonitor()
 	defer session.Finish(ts.sqlExecutor)
 	query := `
@@ -713,7 +713,7 @@ func TestAdminAPIEvents(t *testing.T) {
 	defer span.Finish()
 	session := sql.NewSession(
 		ctx, sql.SessionArgs{User: security.RootUser}, ts.sqlExecutor,
-		nil /* remote */, &sql.MemoryMetrics{}, nil /* conn */)
+		&sql.MemoryMetrics{}, nil /* conn */)
 	session.StartUnlimitedMonitor()
 	defer session.Finish(ts.sqlExecutor)
 	setupQueries := []string{

pkg/server/authentication.go

@@ -238,7 +238,7 @@ func (s *authenticationServer) verifyPassword(
 	ctx context.Context, username string, password string,
 ) (bool, error) {
 	exists, hashedPassword, err := sql.GetUserHashedPassword(
-		ctx, s.server.sqlExecutor, s.memMetrics, username,
+		ctx, s.server.execCfg, s.memMetrics, username,
 	)
 	if err != nil {
 		return false, err

pkg/server/config.go

@@ -71,6 +71,7 @@ const (
 	TempDirsRecordFilename                = "temp-dirs-record.txt"
 	defaultEventLogEnabled                = true
 	defaultEnableWebSessionAuthentication = false
+	defaultUseFrontendV2                  = true
 
 	maximumMaxClockOffset = 5 * time.Second
 
@@ -234,6 +235,9 @@ type Config struct {
 	// the Admin API's HTTP endpoints.
 	EnableWebSessionAuthentication bool
 
+	// !!!
+	UseFrontendV2 bool
+
 	enginesCreated bool
 }
 
@@ -366,6 +370,7 @@ func MakeConfig(ctx context.Context, st *cluster.Settings) Config {
 		ScanMaxIdleTime:                defaultScanMaxIdleTime,
 		EventLogEnabled:                defaultEventLogEnabled,
 		EnableWebSessionAuthentication: defaultEnableWebSessionAuthentication,
+		UseFrontendV2:                  defaultUseFrontendV2,
 		Stores: base.StoreSpecList{
 			Specs: []base.StoreSpec{storeSpec},
 		},

pkg/server/server.go

@@ -521,7 +521,9 @@ func NewServer(cfg Config, stopper *stop.Stopper) (*Server, error) {
 		execCfg.EvalContextTestingKnobs = *sqlEvalContext.(*tree.EvalContextTestingKnobs)
 	}
 	s.sqlExecutor = sql.NewExecutor(execCfg, s.stopper)
-	s.registry.AddMetricStruct(s.sqlExecutor)
+	if !s.cfg.UseFrontendV2 {
+		s.registry.AddMetricStruct(s.sqlExecutor)
+	}
 
 	s.pgServer = pgwire.MakeServer(
 		s.cfg.AmbientCtx,
@@ -531,8 +533,13 @@ func NewServer(cfg Config, stopper *stop.Stopper) (*Server, error) {
 		&s.internalMemMetrics,
 		&rootSQLMemoryMonitor,
 		s.cfg.HistogramWindowInterval(),
+		&execCfg,
 	)
 	s.registry.AddMetricStruct(s.pgServer.Metrics())
+	if s.cfg.UseFrontendV2 {
+		s.registry.AddMetricStruct(s.pgServer.StatementCounters())
+		s.registry.AddMetricStruct(s.pgServer.EngineMetrics())
+	}
 
 	sqlExecutor.ExecCfg = &execCfg
 	s.execCfg = &execCfg
@@ -1166,6 +1173,7 @@ If problems persist, please see ` + base.DocsURL("cluster-setup-troubleshooting.
 
 	s.sqlExecutor.Start(ctx, s.execCfg.DistSQLPlanner)
 	s.distSQLServer.Start()
+	s.pgServer.Start(ctx, s.stopper)
 
 	s.serveMode.set(modeOperational)
 
@@ -1268,7 +1276,13 @@ If problems persist, please see ` + base.DocsURL("cluster-setup-troubleshooting.
 			connCtx := log.WithLogTagStr(pgCtx, "client", conn.RemoteAddr().String())
 			setTCPKeepAlive(connCtx, conn)
 
-			if err := s.pgServer.ServeConn(connCtx, conn); err != nil && !netutil.IsClosedConnection(err) {
+			var serveFn func(ctx context.Context, conn net.Conn) error
+			if s.cfg.UseFrontendV2 {
+				serveFn = s.pgServer.ServeConn2
+			} else {
+				serveFn = s.pgServer.ServeConn
+			}
+			if err := serveFn(connCtx, conn); err != nil && !netutil.IsClosedConnection(err) {
 				// Report the error on this connection's context, so that we
 				// know which remote client caused the error when looking at
 				// the logs.
@@ -1340,6 +1354,7 @@ If problems persist, please see ` + base.DocsURL("cluster-setup-troubleshooting.
 func (s *Server) doDrain(
 	ctx context.Context, modes []serverpb.DrainMode, setTo bool,
 ) ([]serverpb.DrainMode, error) {
+	log.Infof(ctx, "!!! Server.doDrain")
 	for _, mode := range modes {
 		switch mode {
 		case serverpb.DrainMode_CLIENT:

pkg/server/updates.go

@@ -242,8 +242,13 @@ func (s *Server) maybeReportDiagnostics(
 	if log.DiagnosticsReportingEnabled.Get(&s.st.SV) && diagnosticsMetricsEnabled.Get(&s.st.SV) {
 		s.reportDiagnostics(running)
 	}
-	s.sqlExecutor.ResetStatementStats(ctx)
-	s.sqlExecutor.ResetUnimplementedCounts()
+	if s.cfg.UseFrontendV2 {
+		s.pgServer.SQLServer.ResetStatementStats(ctx)
+		s.pgServer.SQLServer.ResetUnimplementedCounts()
+	} else {
+		s.sqlExecutor.ResetStatementStats(ctx)
+		s.sqlExecutor.ResetUnimplementedCounts()
+	}
 
 	return scheduled.Add(diagnosticReportFrequency.Get(&s.st.SV))
 }
@@ -280,7 +285,6 @@ func (s *Server) getReportingInfo(ctx context.Context) *diagnosticspb.Diagnostic
 		schema = nil
 	}
 	info.Schema = schema
-	info.SqlStats = s.sqlExecutor.GetScrubbedStmtStats()
 	info.UnimplementedErrors = make(map[string]int64)
 
 	// Read the system.settings table to determine the settings for which we have
@@ -318,7 +322,15 @@ func (s *Server) getReportingInfo(ctx context.Context) *diagnosticspb.Diagnostic
 		}
 	}
 
-	s.sqlExecutor.FillUnimplementedErrorCounts(info.UnimplementedErrors)
+	if s.cfg.UseFrontendV2 {
+		log.Infof(context.TODO(), "!!! server asking v2 for stats")
+		info.SqlStats = s.pgServer.SQLServer.GetScrubbedStmtStats()
+		s.pgServer.SQLServer.FillUnimplementedErrorCounts(info.UnimplementedErrors)
+	} else {
+		log.Infof(context.TODO(), "!!! server asking v1 for stats")
+		info.SqlStats = s.sqlExecutor.GetScrubbedStmtStats()
+		s.sqlExecutor.FillUnimplementedErrorCounts(info.UnimplementedErrors)
+	}
 	return &info
 }
 

pkg/sql/app_stats.go

@@ -165,7 +165,7 @@ func anonymizeStmt(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
@@ -281,14 +281,20 @@ func scrubStmtStatKey(vt VirtualTabler, key string) (string, bool) {
 // queries scrubbed of their identifiers. Any statements which cannot be
 // scrubbed will be omitted from the returned map.
 func (e *Executor) GetScrubbedStmtStats() []roachpb.CollectedStatementStatistics {
+	return e.sqlStats.getScrubbedStmtStats(e.cfg.VirtualSchemas)
+}
+
+func (s *sqlStats) getScrubbedStmtStats(
+	vt *VirtualSchemaHolder,
+) []roachpb.CollectedStatementStatistics {
+	s.Lock()
+	defer s.Unlock()
 	var ret []roachpb.CollectedStatementStatistics
-	vt := e.cfg.VirtualSchemas
-	e.sqlStats.Lock()
-	salt := ClusterSecret.Get(&e.cfg.Settings.SV)
-	for appName, a := range e.sqlStats.apps {
+	salt := ClusterSecret.Get(&s.st.SV)
+	for appName, a := range s.apps {
 		if cap(ret) == 0 {
 			// guesstimate that we'll need apps*(queries-per-app).
-			ret = make([]roachpb.CollectedStatementStatistics, 0, len(a.stmts)*len(e.sqlStats.apps))
+			ret = make([]roachpb.CollectedStatementStatistics, 0, len(a.stmts)*len(s.apps))
 		}
 		hashedAppName := HashForReporting(salt, appName)
 		a.Lock()
@@ -315,7 +321,6 @@ func (e *Executor) GetScrubbedStmtStats() []roachpb.CollectedStatementStatistics
 		}
 		a.Unlock()
 	}
-	e.sqlStats.Unlock()
 	return ret
 }
 

pkg/sql/builtin_mem_usage_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
 )
 
 // lowMemoryBudget is the memory budget used to test builtins are recording
@@ -59,6 +60,7 @@ CREATE TABLE d.t (a STRING)
 	}
 
 	for i := 0; i < numRows; i++ {
+		log.Infof(context.TODO(), "!!! test sending insert: %d", i)
 		if _, err := sqlDB.Exec(`INSERT INTO d.t VALUES (REPEAT('a', $1))`, rowSize); err != nil {
 			return err
 		}

pkg/sql/cancel_query.go

@@ -23,6 +23,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/server/serverpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/types"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/uint128"
 )
 
@@ -56,6 +57,7 @@ func (n *cancelQueryNode) startExec(params runParams) error {
 	if err != nil {
 		return err
 	}
+	log.Infof(params.ctx, "!!! cancel queryID: %s", queryIDDatum)
 
 	queryIDString := tree.AsStringWithFlags(queryIDDatum, tree.FmtBareStrings)
 	queryID, err := uint128.FromString(queryIDString)