change interface of resultObserver

Signed-off-by: Andres Taylor <[email protected]>

go/vt/vtgate/executor.go

@@ -672,9 +672,8 @@ func (e *Executor) executeSPInAllSessions(ctx context.Context, safeSession *Safe
 			})
 			queries = append(queries, &querypb.BoundQuery{Sql: sql})
 		}
-		observer := resultObserverFactory(sql)
-		qr, errs = e.ExecuteMultiShard(ctx, nil, rss, queries, safeSession, false /*autocommit*/, ignoreMaxMemoryRows, observer)
-		observer.close()
+		qr, errs = e.ExecuteMultiShard(ctx, nil, rss, queries, safeSession, false /*autocommit*/, ignoreMaxMemoryRows, nullResultsObserver{})
+
 		err := vterrors.Aggregate(errs)
 		if err != nil {
 			return nil, err
@@ -1408,7 +1407,6 @@ func (e *Executor) prepare(ctx context.Context, safeSession *SafeSession, sql st
 func (e *Executor) handlePrepare(ctx context.Context, safeSession *SafeSession, sql string, bindVars map[string]*querypb.BindVariable, logStats *logstats.LogStats) ([]*querypb.Field, error) {
 	query, comments := sqlparser.SplitMarginComments(sql)
 	vcursor, _ := newVCursorImpl(safeSession, comments, e, logStats, e.vm, e.VSchema(), e.resolver.resolver, e.serv, e.warnShardedOnly, e.pv, sql)
-	defer vcursor.Close()
 	stmt, reservedVars, err := parseAndValidateQuery(query, e.env.Parser())
 	if err != nil {
 		return nil, err

go/vt/vtgate/legacy_scatter_conn_test.go

@@ -380,7 +380,9 @@ func (o *recordingResultsObserver) observe(result *sqltypes.Result) {
 	mu.Unlock()
 }
 
-func (o *recordingResultsObserver) close() {}
+func (o *recordingResultsObserver) clone() resultsObserver {
+	return o
+}
 
 func TestMultiExecs(t *testing.T) {
 	ctx := utils.LeakCheckContext(t)

go/vt/vtgate/plan_execute.go

@@ -115,7 +115,6 @@ func (e *Executor) newExecute(
 		if err != nil {
 			return err
 		}
-		defer vcursor.Close() // we know this happens in a loop, but MaxBufferingRetries is small, so it's fine
 
 		// 3: Create a plan for the query.
 		// If we are retrying, it is likely that the routing rules have changed and hence we need to

go/vt/vtgate/scatter_conn.go

@@ -24,6 +24,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"vitess.io/vitess/go/vt/vtgate/logstats"
+
 	"vitess.io/vitess/go/mysql/sqlerror"
 	"vitess.io/vitess/go/vt/sqlparser"
 
@@ -76,20 +78,22 @@ type (
 	// resultsObserver will be given the chance to observe the results coming in for a specific query
 	resultsObserver interface {
 		observe(*sqltypes.Result)
-		close()
+		clone() resultsObserver
 	}
 
-	nullResultsObserver       struct{}
-	nullResultObserverFactory func(sql string) resultsObserver
+	nullResultsObserver struct{}
 )
 
 // resultObserverFactory is a factory function that creates a resultsObserver, associating them with the query
-var resultObserverFactory func(sql string) resultsObserver = func(sql string) resultsObserver {
+var resultObserverFactory = func(*logstats.LogStats) resultsObserver {
 	return nullResultsObserver{}
 }
 
 func (nullResultsObserver) observe(*sqltypes.Result) {}
-func (nullResultsObserver) close()                   {}
+
+func (nullResultsObserver) clone() resultsObserver {
+	return nullResultsObserver{}
+}
 
 // NewScatterConn creates a new ScatterConn.
 func NewScatterConn(statsName string, txConn *TxConn, gw *TabletGateway) *ScatterConn {

go/vt/vtgate/vcursor_impl.go

@@ -198,7 +198,7 @@ func newVCursorImpl(
 		pv:                  pv,
 		warmingReadsPercent: warmingReadsPct,
 		warmingReadsChannel: warmingReadsChan,
-		resultsObserver:     resultObserverFactory(sql),
+		resultsObserver:     resultObserverFactory(logStats),
 	}, nil
 }
 
@@ -1329,7 +1329,7 @@ func (vc *vcursorImpl) cloneWithAutocommitSession() *vcursorImpl {
 		topoServer:      vc.topoServer,
 		warnShardedOnly: vc.warnShardedOnly,
 		pv:              vc.pv,
-		resultsObserver: vc.resultsObserver,
+		resultsObserver: vc.resultsObserver.clone(),
 	}
 }
 
@@ -1402,7 +1402,7 @@ func (vc *vcursorImpl) CloneForReplicaWarming(ctx context.Context) engine.VCurso
 		warnShardedOnly:     vc.warnShardedOnly,
 		warnings:            vc.warnings,
 		pv:                  vc.pv,
-		resultsObserver:     vc.resultsObserver,
+		resultsObserver:     vc.resultsObserver.clone(),
 	}
 
 	v.marginComments.Trailing += "/* warming read */"
@@ -1434,7 +1434,7 @@ func (vc *vcursorImpl) CloneForMirroring(ctx context.Context) engine.VCursor {
 		warnShardedOnly:     vc.warnShardedOnly,
 		warnings:            vc.warnings,
 		pv:                  vc.pv,
-		resultsObserver:     vc.resultsObserver,
+		resultsObserver:     vc.resultsObserver.clone(),
 	}
 
 	v.marginComments.Trailing += "/* mirror query */"
@@ -1460,7 +1460,3 @@ func (vc *vcursorImpl) UpdateForeignKeyChecksState(fkStateFromQuery *bool) {
 func (vc *vcursorImpl) GetForeignKeyChecksState() *bool {
 	return vc.fkChecksState
 }
-
-func (vc *vcursorImpl) Close() {
-	vc.resultsObserver.close()
-}