func checkCounterGE(s serverutils.TestServerInterface, meta metric.Metadata, e int64) error {
	if a := s.MustGetSQLCounter(meta.Name); a < e {
		return errors.Errorf("stat %s: expected: actual %d >= %d",
			meta.Name, a, e)
	}
	return nil
}

// runTestFlow runs a flow with the given processors and returns the results.
// Any errors stop the current test.
func runTestFlow(
	t *testing.T, srv serverutils.TestServerInterface, procs ...distsqlrun.ProcessorSpec,
) sqlbase.EncDatumRows {
	kvDB := srv.KVClient().(*client.DB)
	distSQLSrv := srv.DistSQLServer().(*distsqlrun.ServerImpl)

	req := distsqlrun.SetupFlowRequest{
		Txn: client.NewTxn(context.TODO(), *kvDB).Proto,
		Flow: distsqlrun.FlowSpec{
			FlowID:     distsqlrun.FlowID{UUID: uuid.MakeV4()},
			Processors: procs,
		},
	}

	var rowBuf distsqlrun.RowBuffer

	flow, err := distSQLSrv.SetupSyncFlow(context.TODO(), &req, &rowBuf)
	if err != nil {
		t.Fatal(err)
	}
	flow.Start(func() {})
	flow.Wait()
	flow.Cleanup()

	if rowBuf.Err != nil {
		t.Fatal(rowBuf.Err)
	}
	if !rowBuf.Closed {
		t.Errorf("output not closed")
	}
	return rowBuf.Rows
}

func checkCounterDelta(
	s serverutils.TestServerInterface, meta metric.Metadata, init, delta int64,
) (int64, error) {
	actual := s.MustGetSQLCounter(meta.Name)
	if actual != (init + delta) {
		return actual, errors.Errorf("query %s: actual %d != (init %d + delta %d)",
			meta.Name, actual, init, delta)
	}
	return actual, nil
}

// getText fetches the HTTP response body as text in the form of a
// byte slice from the specified URL.
func getText(ts serverutils.TestServerInterface, url string) ([]byte, error) {
	httpClient, err := ts.GetHTTPClient()
	if err != nil {
		return nil, err
	}
	resp, err := httpClient.Get(url)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	return ioutil.ReadAll(resp.Body)
}

// setupMultipleRanges creates a test server and splits the
// key range at the given keys. Returns the test server and client.
// The caller is responsible for stopping the server and
// closing the client.
func setupMultipleRanges(
	t *testing.T, ts serverutils.TestServerInterface, splitAt ...string,
) *client.DB {
	db := createTestClient(t, ts.Stopper(), ts.ServingAddr())

	// Split the keyspace at the given keys.
	for _, key := range splitAt {
		if err := db.AdminSplit(context.TODO(), key); err != nil {
			// Don't leak server goroutines.
			t.Fatal(err)
		}
	}

	return db
}

func createTestClientForUser(
	t *testing.T, s serverutils.TestServerInterface, user string,
) *client.DB {
	var ctx base.Config
	ctx.InitDefaults()
	ctx.User = user
	ctx.SSLCA = filepath.Join(security.EmbeddedCertsDir, security.EmbeddedCACert)
	ctx.SSLCert = filepath.Join(security.EmbeddedCertsDir, fmt.Sprintf("%s.crt", user))
	ctx.SSLCertKey = filepath.Join(security.EmbeddedCertsDir, fmt.Sprintf("%s.key", user))
	conn, err := rpc.NewContext(log.AmbientContext{}, &ctx, s.Clock(), s.Stopper()).GRPCDial(s.ServingAddr())
	if err != nil {
		t.Fatal(err)
	}
	return client.NewDB(client.NewSender(conn))
}

func initReverseScanTestEnv(s serverutils.TestServerInterface, t *testing.T) *client.DB {
	db := createTestClient(t, s.Stopper(), s.ServingAddr())

	// Set up multiple ranges:
	// ["", "b"),["b", "e") ,["e", "g") and ["g", "\xff\xff").
	for _, key := range []string{"b", "e", "g"} {
		// Split the keyspace at the given key.
		if err := db.AdminSplit(context.TODO(), key); err != nil {
			t.Fatal(err)
		}
	}
	// Write keys before, at, and after the split key.
	for _, key := range []string{"a", "b", "c", "d", "e", "f", "g", "h"} {
		if err := db.Put(context.TODO(), key, "value"); err != nil {
			t.Fatal(err)
		}
	}
	return db
}

func createTestClientForUser(
	t *testing.T, s serverutils.TestServerInterface, user string, dbCtx client.DBContext,
) *client.DB {
	rpcContext := rpc.NewContext(log.AmbientContext{}, &base.Config{
		User:       user,
		SSLCA:      filepath.Join(security.EmbeddedCertsDir, security.EmbeddedCACert),
		SSLCert:    filepath.Join(security.EmbeddedCertsDir, fmt.Sprintf("%s.crt", user)),
		SSLCertKey: filepath.Join(security.EmbeddedCertsDir, fmt.Sprintf("%s.key", user)),
	}, s.Clock(), s.Stopper())
	conn, err := rpcContext.GRPCDial(s.ServingAddr())
	if err != nil {
		t.Fatal(err)
	}
	return client.NewDBWithContext(client.NewSender(conn), dbCtx)
}

// checkSQLNetworkMetrics returns the server's pgwire bytesIn/bytesOut and an
// error if the bytesIn/bytesOut don't satisfy the given minimums and maximums.
func checkSQLNetworkMetrics(
	s serverutils.TestServerInterface, minBytesIn, minBytesOut, maxBytesIn, maxBytesOut int64,
) (int64, int64, error) {
	if err := s.WriteSummaries(); err != nil {
		return -1, -1, err
	}

	bytesIn := s.MustGetSQLNetworkCounter(pgwire.MetaBytesIn.Name)
	bytesOut := s.MustGetSQLNetworkCounter(pgwire.MetaBytesOut.Name)
	if a, min := bytesIn, minBytesIn; a < min {
		return bytesIn, bytesOut, errors.Errorf("bytesin %d < expected min %d", a, min)
	}
	if a, min := bytesOut, minBytesOut; a < min {
		return bytesIn, bytesOut, errors.Errorf("bytesout %d < expected min %d", a, min)
	}
	if a, max := bytesIn, maxBytesIn; a > max {
		return bytesIn, bytesOut, errors.Errorf("bytesin %d > expected max %d", a, max)
	}
	if a, max := bytesOut, maxBytesOut; a > max {
		return bytesIn, bytesOut, errors.Errorf("bytesout %d > expected max %d", a, max)
	}
	return bytesIn, bytesOut, nil
}

// debugURL returns the root debug URL.
func debugURL(s serverutils.TestServerInterface) string {
	return s.AdminURL() + debugEndpoint
}

func checkCounterGE(t *testing.T, s serverutils.TestServerInterface, meta metric.Metadata, e int64) {
	if a := s.MustGetSQLCounter(meta.Name); a < e {
		t.Error(errors.Errorf("stat %s: expected: actual %d >= %d", meta.Name, a, e))
	}
}