flowinfra: improve BenchmarkFlowSetup

This commit improves `BenchmarkFlowSetup` by pulling out parsing of the
statement from the hot path as well as more precise usage of the timer
(to exclude the construction of the internal planner). This way the
benchmark is better focused on its goal - the performance of the
planning and execution.

Release justification: test-only change.

Release note: None

pkg/sql/flowinfra/BUILD.bazel

@@ -92,6 +92,7 @@ go_test(
         "//pkg/sql/execinfra",
         "//pkg/sql/execinfra/execopnode",
         "//pkg/sql/execinfrapb",
+        "//pkg/sql/parser",
         "//pkg/sql/pgwire/pgerror",
         "//pkg/sql/randgen",
         "//pkg/sql/rowenc",

pkg/sql/flowinfra/flow_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/sql/parser"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondatapb"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
@@ -44,13 +45,18 @@ func BenchmarkFlowSetup(b *testing.B) {
 
 	execCfg := s.ExecutorConfig().(sql.ExecutorConfig)
 	dsp := execCfg.DistSQLPlanner
+	stmt, err := parser.ParseOne("SELECT k FROM b.test WHERE k=1")
+	if err != nil {
+		b.Fatal(err)
+	}
 	for _, vectorize := range []bool{true, false} {
 		for _, distribute := range []bool{true, false} {
 			b.Run(fmt.Sprintf("vectorize=%t/distribute=%t", vectorize, distribute), func(b *testing.B) {
 				vectorizeMode := sessiondatapb.VectorizeOff
 				if vectorize {
 					vectorizeMode = sessiondatapb.VectorizeOn
 				}
+				b.ResetTimer()
 				for i := 0; i < b.N; i++ {
 					// NB: planner cannot be reset and can only be used for
 					// a single statement, so we create a new one on every
@@ -63,12 +69,15 @@ func BenchmarkFlowSetup(b *testing.B) {
 						&execCfg,
 						sessiondatapb.SessionData{VectorizeMode: vectorizeMode},
 					)
-					if err := dsp.Exec(
+					b.StartTimer()
+					err := dsp.Exec(
 						ctx,
 						planner,
-						"SELECT k FROM b.test WHERE k=1",
+						stmt,
 						distribute,
-					); err != nil {
+					)
+					b.StopTimer()
+					if err != nil {
 						b.Fatal(err)
 					}
 					cleanup()

pkg/sql/testutils.go

@@ -124,12 +124,8 @@ func (r *StmtBufReader) AdvanceOne() {
 // interface{} so that external packages can call NewInternalPlanner and pass
 // the result) and executes a sql statement through the DistSQLPlanner.
 func (dsp *DistSQLPlanner) Exec(
-	ctx context.Context, localPlanner interface{}, sql string, distribute bool,
+	ctx context.Context, localPlanner interface{}, stmt parser.Statement, distribute bool,
 ) error {
-	stmt, err := parser.ParseOne(sql)
-	if err != nil {
-		return err
-	}
 	p := localPlanner.(*planner)
 	p.stmt = makeStatement(stmt, clusterunique.ID{} /* queryID */)
 	if err := p.makeOptimizerPlan(ctx); err != nil {