func createTestServerContext() (*server.Context, *CommandFilters) {
	ctx := server.NewTestContext()
	var cmdFilters CommandFilters
	cmdFilters.AppendFilter(checkEndTransactionTrigger, true)
	ctx.TestingKnobs.StoreTestingKnobs.TestingCommandFilter = cmdFilters.runFilters
	return ctx, &cmdFilters
}

// benchmarkSelect1 is a benchmark of the simplest SQL query: SELECT 1. This
// query requires no tables, expression analysis, etc. As such, it is measuring
// the overhead of parsing and other non-table processing (e.g. reading
// requests, writing responses).
func benchmarkSelect1(b *testing.B, scheme string) {
	s := &server.TestServer{}
	s.Ctx = server.NewTestContext()
	s.Ctx.Insecure = (scheme == "http" || scheme == "rpc")
	if err := s.Start(); err != nil {
		b.Fatal(err)
	}
	defer s.Stop()

	db, err := sql.Open("cockroach",
		scheme+"://[email protected]"+s.ServingAddr()+"?certs="+s.Ctx.Certs)
	if err != nil {
		b.Fatal(err)
	}
	defer db.Close()

	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		rows, err := db.Query(`SELECT 1`)
		if err != nil {
			b.Fatal(err)
		}
		rows.Close()
	}
	b.StopTimer()
}

func TestProtocols(t *testing.T) {
	defer leaktest.AfterTest(t)

	// Test that all of the network protocols work.
	for _, scheme := range []string{"http", "https", "rpc", "rpcs"} {
		func() {
			// Start test server in insecure mode.
			s := &server.TestServer{}
			s.Ctx = server.NewTestContext()
			s.Ctx.Insecure = (scheme == "http" || scheme == "rpc")
			if err := s.Start(); err != nil {
				t.Fatalf("Could not start server: %v", err)
			}
			defer s.Stop()

			db, err := sql.Open("cockroach",
				scheme+"://[email protected]"+s.ServingAddr()+"?certs="+s.Ctx.Certs)
			if err != nil {
				t.Fatal(err)
			}
			defer db.Close()

			if _, err := db.Exec(`SELECT 1`); err != nil {
				t.Fatal(err)
			}
		}()
	}
}

func ExampleDB_Put_insecure() {
	s := &server.TestServer{}
	s.Ctx = server.NewTestContext()
	s.Ctx.Insecure = true
	if pErr := s.Start(); pErr != nil {
		log.Fatalf("Could not start server: %v", pErr)
	}
	defer s.Stop()

	db, err := client.Open(s.Stopper(), "rpc://[email protected]"+s.ServingAddr())
	if err != nil {
		log.Fatal(err)
	}

	if pErr := db.Put("aa", "1"); pErr != nil {
		panic(pErr)
	}
	result, pErr := db.Get("aa")
	if pErr != nil {
		panic(pErr)
	}
	fmt.Printf("aa=%s\n", result.ValueBytes())

	// Output:
	// aa=1
}

func ExampleDB_Insecure() {
	s := &server.TestServer{}
	s.Ctx = server.NewTestContext()
	s.Ctx.Insecure = true
	if err := s.Start(); err != nil {
		log.Fatalf("Could not start server: %v", err)
	}
	log.Printf("Test server listening on %s: %s", s.Ctx.RequestScheme(), s.ServingAddr())
	defer s.Stop()

	db, err := client.Open("http://[email protected]" + s.ServingAddr())
	if err != nil {
		log.Fatal(err)
	}

	if err := db.Put("aa", "1"); err != nil {
		panic(err)
	}
	result, err := db.Get("aa")
	if err != nil {
		panic(err)
	}
	fmt.Printf("aa=%s\n", result.ValueBytes())

	// Output:
	// aa=1
}

func (t *parallelTest) setup() {
	ctx := server.NewTestContext()
	ctx.MaxOffset = logicMaxOffset
	ctx.TestingKnobs.ExecutorTestingKnobs.WaitForGossipUpdate = true
	ctx.TestingKnobs.ExecutorTestingKnobs.CheckStmtStringChange = true
	t.srv = setupTestServerWithContext(t.T, ctx)
}

func TestInsecure(t *testing.T) {
	defer leaktest.AfterTest(t)
	// Start test server in insecure mode.
	s := &server.TestServer{}
	s.Ctx = server.NewTestContext()
	s.Ctx.Insecure = true
	if err := s.Start(); err != nil {
		t.Fatalf("Could not start server: %v", err)
	}
	t.Logf("Test server listening on %s: %s", s.Ctx.RequestScheme(), s.ServingAddr())
	defer s.Stop()

	// We can't attempt a connection through HTTPS since the client just retries forever.
	// DB connection using plain HTTP.
	db, err := sql.Open("cockroach", "http://[email protected]"+s.ServingAddr())
	if err != nil {
		t.Fatal(err)
	}
	defer func() {
		_ = db.Close()
	}()
	if _, err := db.Exec(`SELECT 1`); err != nil {
		t.Fatal(err)
	}
}

func createTestServerContext() (*server.Context, *CommandFilters) {
	ctx := server.NewTestContext()
	var cmdFilters CommandFilters
	cmdFilters.AppendFilter(checkEndTransactionTrigger, true)
	// Disable one phase commits as they otherwise confuse the
	// various bits of machinery in sql tests which inject via
	// the testing command filter and inspect the transaction.
	ctx.TestingKnobs.StoreTestingKnobs.DisableOnePhaseCommits = true
	ctx.TestingKnobs.StoreTestingKnobs.TestingCommandFilter = cmdFilters.runFilters
	return ctx, &cmdFilters
}

func (t *parallelTest) run(dir string) {
	fmt.Printf("Running test %s\n", dir)
	// Set up database
	defer t.close()
	ctx := server.NewTestContext()
	ctx.MaxOffset = logicMaxOffset
	t.srv = setupTestServer(t.T)

	// Add the main client and set up the database.
	t.addClient(true)

	// Open the main faile.
	mainFile := filepath.Join(dir, "main")
	file, err := os.Open(mainFile)
	if err != nil {
		t.Fatal(err)
	}
	defer file.Close()
	s := newLineScanner(file)
	for s.Scan() {
		fields := strings.Fields(s.Text())
		if len(fields) == 0 {
			continue
		}
		cmd := fields[0]
		if strings.HasPrefix(cmd, "#") {
			// Skip comment lines.
			continue
		}
		switch cmd {
		case "run":
			testFiles := fields[1:]
			for len(t.clients) < len(testFiles) {
				t.addClient(false)
			}
			if testing.Verbose() || log.V(1) {
				fmt.Printf("%s:%d: running %s\n", mainFile, s.line, strings.Join(testFiles, ","))
			}
			ch := make(chan bool)
			for i, f := range testFiles {
				go t.processTestFile(filepath.Join(dir, f), t.clients[i].db, ch)
			}
			// Wait for all clients to complete.
			for range testFiles {
				<-ch
			}
		default:
			t.Fatalf("%s:%d: unknown command: %s", mainFile, s.line, cmd)
		}
	}
}

func TestLeaseManagerReacquire(testingT *testing.T) {
	defer leaktest.AfterTest(testingT)()
	t := newLeaseTest(testingT, server.NewTestContext())
	defer t.cleanup()

	const descID = keys.LeaseTableID

	// Acquire 2 leases from the same node. They should point to the same lease
	// structure.
	l1 := t.mustAcquire(1, descID, 0)
	l2 := t.mustAcquire(1, descID, 0)
	if l1 != l2 {
		t.Fatalf("expected same lease, but found %p != %p", l1, l2)
	}
	if l1.Refcount() != 2 {
		t.Fatalf("expected refcount of 2, but found %d", l1.Refcount())
	}
	t.expectLeases(descID, "/1/1")

	// Set the minimum lease duration such that the next lease acquisition will
	// require the lease to be reacquired.
	savedLeaseDuration, savedMinLeaseDuration := csql.LeaseDuration, csql.MinLeaseDuration
	defer func() {
		csql.LeaseDuration, csql.MinLeaseDuration = savedLeaseDuration, savedMinLeaseDuration
	}()
	csql.MinLeaseDuration = l1.Expiration().Sub(timeutil.Now())
	csql.LeaseDuration = 2 * csql.MinLeaseDuration

	// Another lease acquisition from the same node will result in a new lease.
	l3 := t.mustAcquire(1, descID, 0)
	if l1 == l3 {
		t.Fatalf("expected different leases, but found %p", l1)
	}
	if l3.Refcount() != 1 {
		t.Fatalf("expected refcount of 1, but found %d", l3.Refcount())
	}
	if l3.Expiration().Before(l1.Expiration()) {
		t.Fatalf("expected new lease expiration (%s) to be after old lease expiration (%s)",
			l3.Expiration(), l1.Expiration())
	}
	t.expectLeases(descID, "/1/1 /1/1")

	t.mustRelease(1, l1)
	t.mustRelease(1, l2)
	t.mustRelease(1, l3)
}

// benchmarkSelect1 is a benchmark of the simplest SQL query: SELECT 1. This
// query requires no tables, expression analysis, etc. As such, it is measuring
// the overhead of parsing and other non-table processing (e.g. reading
// requests, writing responses).
func benchmarkSelect1(b *testing.B, scheme string) {
	s := &server.TestServer{}
	s.Ctx = server.NewTestContext()
	s.Ctx.Insecure = (scheme == "http" || scheme == "rpc")
	if err := s.Start(); err != nil {
		b.Fatal(err)
	}
	defer s.Stop()

	db, err := sql.Open("cockroach",
		scheme+"://[email protected]"+s.ServingAddr()+"?certs="+s.Ctx.Certs)
	if err != nil {
		b.Fatal(err)
	}
	defer db.Close()

	runBenchmarkSelect1(b, db)
}

func (t *logicTest) setup() {
	// TODO(pmattis): Add a flag to make it easy to run the tests against a local
	// MySQL or Postgres instance.
	ctx := server.NewTestContext()
	ctx.MaxOffset = logicMaxOffset
	t.srv = setupTestServerWithContext(t.T, ctx)

	// db may change over the lifetime of this function, with intermediate
	// values cached in t.clients and finally closed in t.close().
	t.cleanupRootUser = t.setUser(security.RootUser)

	if _, err := t.db.Exec(`
CREATE DATABASE test;
SET DATABASE = test;
`); err != nil {
		t.Fatal(err)
	}
}

// Test that if there's an error on COMMIT that needs to be reported to the user
// the txn will be rolled back. As opposed to an error on a COMMIT in an auto-retry
// txn, where we retry the txn (not tested here).
func TestErrorOnCommit(t *testing.T) {
	defer leaktest.AfterTest(t)()

	ctx := server.NewTestContext()
	ctx.TestingKnobs.SQLExecutor = &sql.ExecutorTestingKnobs{FixTxnPriority: true}
	server, sqlDB, _ := setupWithContext(t, ctx)
	defer cleanup(server, sqlDB)
	if _, err := sqlDB.Exec(`
CREATE DATABASE t; CREATE TABLE t.test (k INT PRIMARY KEY, v TEXT);
`); err != nil {
		t.Fatal(err)
	}

	tx, err := sqlDB.Begin()
	if err != nil {
		t.Fatal(err)
	}
	if _, err := tx.Exec("SAVEPOINT cockroach_restart; SET TRANSACTION PRIORITY LOW;"); err != nil {
		t.Fatal(err)
	}
	if _, err = tx.Exec("INSERT INTO t.test (k, v) VALUES (0, 'sentinel');"); err != nil {
		t.Fatal(err)
	}

	abortTxn(t, sqlDB, 0)

	if err = tx.Commit(); err == nil {
		t.Fatal("expected commit to fail")
	}

	// Check that there's no error reading and we don't see any rows.
	var rows *gosql.Rows
	if rows, err = sqlDB.Query("SELECT * FROM t.test"); err != nil {
		t.Fatal(err)
	}
	if rows.Next() {
		var k int
		var v string
		_ = rows.Scan(&k, &v)
		t.Fatalf("found unexpected row: %d %s", k, v)
	}
	rows.Close()
}

func Example_insecure() {
	c := cliTest{}
	c.TestServer = &server.TestServer{}
	c.Ctx = server.NewTestContext()
	c.Ctx.Insecure = true
	if err := c.Start(); err != nil {
		log.Fatalf("Could not start server: %v", err)
	}
	defer c.Stop()

	c.Run("kv --insecure put a 1 b 2")
	c.Run("kv --insecure scan")

	// Output:
	// kv --insecure put a 1 b 2
	// kv --insecure scan
	// "a"	"1"
	// "b"	"2"
	// 2 result(s)
}

func Example_insecure() {
	c := cliTest{cleanupFunc: func() {}}
	c.TestServer = &server.TestServer{}
	c.Ctx = server.NewTestContext()
	c.Ctx.Insecure = true
	if err := c.Start(); err != nil {
		log.Fatalf("Could not start server: %v", err)
	}
	defer c.stop()

	c.Run("debug kv put --insecure a 1 b 2")
	c.Run("debug kv scan --insecure")

	// Output:
	// debug kv put --insecure a 1 b 2
	// debug kv scan --insecure
	// "a"	"1"
	// "b"	"2"
	// 2 result(s)
}

func Example_insecure() {
	c := cliTest{}
	c.TestServer = &server.TestServer{}
	c.Ctx = server.NewTestContext()
	c.Ctx.Insecure = true
	if err := c.Start(); err != nil {
		log.Fatalf("Could not start server: %v", err)
	}

	c.Run("kv --insecure put a 1 b 2")
	c.Run("kv --insecure scan")
	c.Run("quit --insecure")

	// Output:
	// kv --insecure put a 1 b 2
	// kv --insecure scan
	// "a"	"1"
	// "b"	"2"
	// quit --insecure
	// node drained and shutdown: ok
}

func ExampleDB_Put_insecure() {
	s := &server.TestServer{}
	s.Ctx = server.NewTestContext()
	s.Ctx.Insecure = true
	if pErr := s.Start(); pErr != nil {
		log.Fatalf("Could not start server: %v", pErr)
	}
	defer s.Stop()

	db := s.DB()
	if pErr := db.Put("aa", "1"); pErr != nil {
		panic(pErr)
	}
	result, pErr := db.Get("aa")
	if pErr != nil {
		panic(pErr)
	}
	fmt.Printf("aa=%s\n", result.ValueBytes())

	// Output:
	// aa=1
}

func TestLeaseManagerPublishVersionChanged(testingT *testing.T) {
	defer leaktest.AfterTest(testingT)()
	t := newLeaseTest(testingT, server.NewTestContext())
	defer t.cleanup()

	const descID = keys.LeaseTableID

	// Start two goroutines that are concurrently trying to publish a new version
	// of the descriptor. The first goroutine progresses to the update function
	// and then signals the second goroutine to start which is allowed to proceed
	// through completion. The first goroutine is then signaled and when it
	// attempts to publish the new version it will encounter an update error and
	// retry the transaction. Upon retry it will see that the descriptor version
	// has changed and have to proceed to its outer retry loop and wait for the
	// number of leases on the previous version to drop to 0.

	n1 := t.node(1)
	n2 := t.node(2)

	n1update := make(chan struct{})
	n2start := make(chan struct{})
	var wg sync.WaitGroup
	wg.Add(2)

	go func(n1update, n2start chan struct{}) {
		_, err := n1.Publish(descID, func(*sqlbase.TableDescriptor) error {
			if n2start != nil {
				// Signal node 2 to start.
				close(n2start)
				n2start = nil
			}
			// Wait for node 2 signal indicating that node 2 finished publication of
			// a new version.
			<-n1update
			return nil
		})
		if err != nil {
			panic(err)
		}
		wg.Done()
	}(n1update, n2start)

	go func(n1update, n2start chan struct{}) {
		// Wait for node 1 signal indicating that node 1 is in its update()
		// function.
		<-n2start
		_, err := n2.Publish(descID, func(*sqlbase.TableDescriptor) error {
			return nil
		})
		if err != nil {
			panic(err)
		}
		close(n1update)
		wg.Done()
	}(n1update, n2start)

	wg.Wait()

	t.mustAcquire(1, descID, 0)
	t.expectLeases(descID, "/3/1")
}

func setupClientBenchData(useSSL bool, numVersions, numKeys int, b *testing.B) (
	*server.TestServer, *client.DB) {
	const cacheSize = 8 << 30        // 8 GB
	const memtableBudget = 512 << 20 // 512 MB
	loc := fmt.Sprintf("client_bench_%d_%d", numVersions, numKeys)

	exists := true
	if _, err := os.Stat(loc); os.IsNotExist(err) {
		exists = false
	}

	s := &server.TestServer{}
	s.Ctx = server.NewTestContext()
	s.SkipBootstrap = exists
	if !useSSL {
		s.Ctx.Insecure = true
	}
	stopper := stop.NewStopper()
	s.Ctx.Engines = []engine.Engine{
		engine.NewRocksDB(roachpb.Attributes{Attrs: []string{"ssd"}}, loc,
			cacheSize, memtableBudget, stopper),
	}
	if err := s.StartWithStopper(stopper); err != nil {
		b.Fatal(err)
	}

	db, err := client.Open(s.Stopper(), fmt.Sprintf("%s://%[email protected]%s?certs=%s",
		s.Ctx.RPCRequestScheme(), security.NodeUser, s.ServingAddr(), s.Ctx.Certs))
	if err != nil {
		b.Fatal(err)
	}

	if exists {
		return s, db
	}

	rng, _ := randutil.NewPseudoRand()
	keys := make([]roachpb.Key, numKeys)
	nvs := make([]int, numKeys)
	for t := 1; t <= numVersions; t++ {
		batch := &client.Batch{}
		for i := 0; i < numKeys; i++ {
			if t == 1 {
				keys[i] = roachpb.Key(encoding.EncodeUvarintAscending([]byte("key-"), uint64(i)))
				nvs[i] = int(rand.Int31n(int32(numVersions)) + 1)
			}
			// Only write values if this iteration is less than the random
			// number of versions chosen for this key.
			if t <= nvs[i] {
				batch.Put(roachpb.Key(keys[i]), randutil.RandBytes(rng, valueSize))
			}
			if (i+1)%1000 == 0 {
				if pErr := db.Run(batch); pErr != nil {
					b.Fatal(pErr)
				}
				batch = &client.Batch{}
			}
		}
		if len(batch.Results) != 0 {
			if pErr := db.Run(batch); pErr != nil {
				b.Fatal(pErr)
			}
		}
	}

	if r, ok := s.Ctx.Engines[0].(*engine.RocksDB); ok {
		r.CompactRange(engine.NilKey, engine.NilKey)
	}

	return s, db
}

func setupClientBenchData(useRPC, useSSL bool, numVersions, numKeys int, b *testing.B) (
	*server.TestServer, *client.DB) {
	const cacheSize = 8 << 30 // 8 GB
	loc := fmt.Sprintf("client_bench_%d_%d", numVersions, numKeys)

	exists := true
	if _, err := os.Stat(loc); os.IsNotExist(err) {
		exists = false
	}

	s := &server.TestServer{}
	s.Ctx = server.NewTestContext()
	s.Ctx.ExperimentalRPCServer = true
	s.SkipBootstrap = exists
	if !useSSL {
		s.Ctx.Insecure = true
	}
	s.Engines = []engine.Engine{engine.NewRocksDB(proto.Attributes{Attrs: []string{"ssd"}}, loc, cacheSize)}
	if err := s.Start(); err != nil {
		b.Fatal(err)
	}

	var scheme string
	if useRPC {
		scheme = "rpcs"
	} else {
		scheme = "https"
	}

	db, err := client.Open(scheme + "://[email protected]" + s.ServingAddr() + "?certs=" + s.Ctx.Certs)
	if err != nil {
		b.Fatal(err)
	}

	if exists {
		return s, db
	}

	rng, _ := util.NewPseudoRand()
	keys := make([]proto.Key, numKeys)
	nvs := make([]int, numKeys)
	for t := 1; t <= numVersions; t++ {
		batch := &client.Batch{}
		for i := 0; i < numKeys; i++ {
			if t == 1 {
				keys[i] = proto.Key(encoding.EncodeUvarint([]byte("key-"), uint64(i)))
				nvs[i] = int(rand.Int31n(int32(numVersions)) + 1)
			}
			// Only write values if this iteration is less than the random
			// number of versions chosen for this key.
			if t <= nvs[i] {
				batch.Put(proto.Key(keys[i]), util.RandBytes(rng, valueSize))
			}
			if (i+1)%1000 == 0 {
				if err := db.Run(batch); err != nil {
					b.Fatal(err)
				}
				batch = &client.Batch{}
			}
		}
		if len(batch.Results) != 0 {
			if err := db.Run(batch); err != nil {
				b.Fatal(err)
			}
		}
	}

	if r, ok := s.Engines[0].(*engine.RocksDB); ok {
		r.CompactRange(nil, nil)
	}

	return s, db
}