roachtest: add read committed variants of ycsb

pkg/cmd/roachtest/tests/ycsb.go

@@ -29,6 +29,7 @@ const envYCSBFlags = "ROACHTEST_YCSB_FLAGS"
 func registerYCSB(r registry.Registry) {
 	workloads := []string{"A", "B", "C", "D", "E", "F"}
 	cpusConfigs := []int{8, 32}
+	cpusWithReadCommitted := 32
 	cpusWithGlobalMVCCRangeTombstone := 32
 
 	// concurrencyConfigs contains near-optimal concurrency levels for each
@@ -45,7 +46,7 @@ func registerYCSB(r registry.Registry) {
 	}
 
 	runYCSB := func(
-		ctx context.Context, t test.Test, c cluster.Cluster, wl string, cpus int, rangeTombstone bool,
+		ctx context.Context, t test.Test, c cluster.Cluster, wl string, cpus int, readCommitted, rangeTombstone bool,
 	) {
 		// For now, we only want to run the zfs tests on GCE, since only GCE supports
 		// starting roachprod instances on zfs.
@@ -75,9 +76,11 @@ func registerYCSB(r registry.Registry) {
 		m := c.NewMonitor(ctx, c.Range(1, nodes))
 		m.Go(func(ctx context.Context) error {
 			var args string
-			args += fmt.Sprintf(" --select-for-update=%t", t.IsBuildVersion("v19.2.0"))
 			args += " --ramp=" + ifLocal(c, "0s", "2m")
 			args += " --duration=" + ifLocal(c, "10s", "30m")
+			if readCommitted {
+				args += " --isolation-level=read_committed"
+			}
 			if envFlags := os.Getenv(envYCSBFlags); envFlags != "" {
 				args += " " + envFlags
 			}
@@ -107,7 +110,7 @@ func registerYCSB(r registry.Registry) {
 				Benchmark: true,
 				Cluster:   r.MakeClusterSpec(4, spec.CPU(cpus)),
 				Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
-					runYCSB(ctx, t, c, wl, cpus, false /* rangeTombstone */)
+					runYCSB(ctx, t, c, wl, cpus, false /* readCommitted */, false /* rangeTombstone */)
 				},
 				Tags: registry.Tags(`aws`),
 			})
@@ -119,19 +122,32 @@ func registerYCSB(r registry.Registry) {
 					Benchmark: true,
 					Cluster:   r.MakeClusterSpec(4, spec.CPU(cpus), spec.SetFileSystem(spec.Zfs)),
 					Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
-						runYCSB(ctx, t, c, wl, cpus, false /* rangeTombstone */)
+						runYCSB(ctx, t, c, wl, cpus, false /* readCommitted */, false /* rangeTombstone */)
 					},
 				})
 			}
 
+			if cpus == cpusWithReadCommitted {
+				r.Add(registry.TestSpec{
+					Name:      fmt.Sprintf("%s/isolation-level=read-committed", name),
+					Owner:     registry.OwnerTestEng,
+					Benchmark: true,
+					Cluster:   r.MakeClusterSpec(4, spec.CPU(cpus)),
+					Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
+						runYCSB(ctx, t, c, wl, cpus, true /* readCommitted */, false /* rangeTombstone */)
+					},
+					Tags: registry.Tags(`aws`),
+				})
+			}
+
 			if cpus == cpusWithGlobalMVCCRangeTombstone {
 				r.Add(registry.TestSpec{
 					Name:      fmt.Sprintf("%s/mvcc-range-keys=global", name),
 					Owner:     registry.OwnerTestEng,
 					Benchmark: true,
 					Cluster:   r.MakeClusterSpec(4, spec.CPU(cpus)),
 					Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
-						runYCSB(ctx, t, c, wl, cpus, true /* rangeTombstone */)
+						runYCSB(ctx, t, c, wl, cpus, false /* readCommitted */, true /* rangeTombstone */)
 					},
 					Tags: registry.Tags(`aws`),
 				})

pkg/workload/connection.go

@@ -107,3 +107,33 @@ func SanitizeUrls(gen Generator, dbOverride string, urls []string) (string, erro
 	}
 	return dbName, nil
 }
+
+// SetDefaultIsolationLevel configures the provided URLs with the specified
+// default transaction isolation level, if any.
+func SetDefaultIsolationLevel(urls []string, isoLevel string) error {
+	if isoLevel == "" {
+		return nil
+	}
+	// As a convenience, replace underscores with spaces. This allows users of the
+	// workload tool to pass --isolation-level=read_committed instead of needing
+	// to pass --isolation-level="read committed".
+	isoLevel = strings.ReplaceAll(isoLevel, "_", " ")
+	// NOTE: validation of the isolation level value is done by the server during
+	// connection establishment.
+	return setUrlParam(urls, "default_transaction_isolation", isoLevel)
+}
+
+// setUrlParam sets the given parameter to the given value in the provided URLs.
+func setUrlParam(urls []string, param, value string) error {
+	for i := range urls {
+		parsed, err := url.Parse(urls[i])
+		if err != nil {
+			return err
+		}
+		q := parsed.Query()
+		q.Set(param, value)
+		parsed.RawQuery = q.Encode()
+		urls[i] = parsed.String()
+	}
+	return nil
+}

pkg/workload/ycsb/ycsb.go

@@ -93,6 +93,7 @@ type ycsb struct {
 	flags     workload.Flags
 	connFlags *workload.ConnFlags
 
+	isoLevel    string
 	timeString  bool
 	insertHash  bool
 	zeroPadding int
@@ -125,8 +126,10 @@ var ycsbMeta = workload.Meta{
 		g := &ycsb{}
 		g.flags.FlagSet = pflag.NewFlagSet(`ycsb`, pflag.ContinueOnError)
 		g.flags.Meta = map[string]workload.FlagMeta{
-			`workload`: {RuntimeOnly: true},
+			`isolation-level`: {RuntimeOnly: true},
+			`workload`:        {RuntimeOnly: true},
 		}
+		g.flags.StringVar(&g.isoLevel, `isolation-level`, ``, `Isolation level to run workload transactions under [serializable, snapshot, read_committed]. If unset, the workload will run with the default isolation level of the database.`)
 		g.flags.BoolVar(&g.timeString, `time-string`, false, `Prepend field[0-9] data with current time in microsecond precision.`)
 		g.flags.BoolVar(&g.insertHash, `insert-hash`, true, `Key to be hashed or ordered.`)
 		g.flags.IntVar(&g.zeroPadding, `zero-padding`, 1, `Key using "insert-hash=false" has zeros padded to left to make this length of digits.`)
@@ -378,6 +381,9 @@ func (g *ycsb) Ops(
 	if err != nil {
 		return workload.QueryLoad{}, err
 	}
+	if err := workload.SetDefaultIsolationLevel(urls, g.isoLevel); err != nil {
+		return workload.QueryLoad{}, err
+	}
 
 	const readStmtStr = `SELECT * FROM usertable WHERE ycsb_key = $1`