Repository: petermattis/pebble
Pull request #150 (2019-05-28 22:35:25)

Derive multiplier from bottom level and smooth it
At the end of insertion bursts, typically the second-to-bottommost level was way larger than the bottommost level, implying there's a lot of compaction debt. That's because we were (like RocksDB) deriving target sizes based on the largest level rather than based on the bottom level. I didn't really understand that heuristic so removed it such that target sizes are based on the bottom level size.

While doing so, I also added back the logic for maintaining a fixed ratio for consecutive levels' target sizes. This was proved to be optimal for write-amp although sacrifices space-amp - more so when the number of levels is smaller.

The below results are averaged over three trials. Throughput improved insignificantly, overall write-amp reduced 3%, and compaction debt at the end of the write burst reduced ~14%.

concurrency | value bytes | instance type | num ops | branch | ops/sec | CPU seconds | Peak RSS (MB) | Burst w-amp | Final w-amp | Burst s-amp | Final s-amp
-- | -- | -- | -- | -- | -- | -- | -- | -- | -- | -- | --
256 | 256 | c5d.2xlarge | 200000000 | master | 55313 | 4149 | 2552 | 10.3 | 13.0 | 1.03 | 1.03
256 | 256 | c5d.2xlarge | 200000000 | level-smoothing | 55378 | 4022 | 2600 | 10.3 | 12.6 | 1.03 | 1.03


petermattis: 

<!-- Reviewable:start -->
This change is [<img src="https://reviewable.io/review_button.png" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/petermattis/pebble/150)
<!-- Reviewable:end -->

ajkr: Hm, these benchmarks aren't comprehensive because space-amp is fixed for insertion only workloads (i.e., even if all files were in L0, space amp would still be 1.03). Ideally we should get overwrite benchmarks, but I won't have time to script those for a little while.
ajkr: Will fix the tests tomorrow. One of the cases is generating non-obvious target sizes.

Patch:
diff --git a/batch.go b/batch.go
index 0988639..e38eb3a 100644
--- a/batch.go
+++ b/batch.go
@@ -937,6 +937,8 @@ func (b *flushableBatch) newIter(o *IterOptions) internalIterator {
 		offsets: b.offsets,
 		cmp:     b.cmp,
 		index:   -1,
+		lower:   o.GetLowerBound(),
+		upper:   o.GetUpperBound(),
 	}
 }
 
@@ -997,6 +999,8 @@ type flushableBatchIter struct {
 	index   int
 	key     InternalKey
 	err     error
+	lower   []byte
+	upper   []byte
 }
 
 // flushableBatchIter implements the internalIterator interface.
@@ -1011,6 +1015,10 @@ func (i *flushableBatchIter) SeekGE(key []byte) (*InternalKey, []byte) {
 		return nil, nil
 	}
 	i.key = i.getKey(i.index)
+	if i.upper != nil && i.cmp(i.key.UserKey, i.upper) >= 0 {
+		i.index = len(i.offsets)
+		return nil, nil
+	}
 	return &i.key, i.Value()
 }
 
@@ -1028,6 +1036,10 @@ func (i *flushableBatchIter) SeekLT(key []byte) (*InternalKey, []byte) {
 		return nil, nil
 	}
 	i.key = i.getKey(i.index)
+	if i.lower != nil && i.cmp(i.key.UserKey, i.lower) < 0 {
+		i.index = -1
+		return nil, nil
+	}
 	return &i.key, i.Value()
 }
 
@@ -1037,6 +1049,10 @@ func (i *flushableBatchIter) First() (*InternalKey, []byte) {
 	}
 	i.index = 0
 	i.key = i.getKey(i.index)
+	if i.upper != nil && i.cmp(i.key.UserKey, i.upper) >= 0 {
+		i.index = len(i.offsets)
+		return nil, nil
+	}
 	return &i.key, i.Value()
 }
 
@@ -1046,6 +1062,10 @@ func (i *flushableBatchIter) Last() (*InternalKey, []byte) {
 	}
 	i.index = len(i.offsets) - 1
 	i.key = i.getKey(i.index)
+	if i.lower != nil && i.cmp(i.key.UserKey, i.lower) < 0 {
+		i.index = -1
+		return nil, nil
+	}
 	return &i.key, i.Value()
 }
 
@@ -1058,6 +1078,10 @@ func (i *flushableBatchIter) Next() (*InternalKey, []byte) {
 		return nil, nil
 	}
 	i.key = i.getKey(i.index)
+	if i.upper != nil && i.cmp(i.key.UserKey, i.upper) >= 0 {
+		i.index = len(i.offsets)
+		return nil, nil
+	}
 	return &i.key, i.Value()
 }
 
@@ -1070,6 +1094,10 @@ func (i *flushableBatchIter) Prev() (*InternalKey, []byte) {
 		return nil, nil
 	}
 	i.key = i.getKey(i.index)
+	if i.lower != nil && i.cmp(i.key.UserKey, i.lower) < 0 {
+		i.index = -1
+		return nil, nil
+	}
 	return &i.key, i.Value()
 }
 
@@ -1106,6 +1134,6 @@ func (i *flushableBatchIter) Close() error {
 }
 
 func (i *flushableBatchIter) SetBounds(lower, upper []byte) {
-	// This should not be called as bounds are not used for this iterator.
-	panic("TODO(ryan): pebble: SetBounds unimplemented")
+	i.lower = lower
+	i.upper = upper
 }
diff --git a/cmd/pebble/main.go b/cmd/pebble/main.go
index d6832d3..ed56b0c 100644
--- a/cmd/pebble/main.go
+++ b/cmd/pebble/main.go
@@ -6,7 +6,6 @@ package main
 
 import (
 	"log"
-	"math"
 	"os"
 	"time"
 
@@ -53,48 +52,8 @@ func main() {
 			"wait for background compactions to complete after load stops")
 		cmd.Flags().BoolVarP(
 			&wipe, "wipe", "w", false, "wipe the database before starting")
-		cmd.Flags().BoolVar(
-			&walOnly, "wal-only", false, "write data only to the WAL")
 	}
 
-	scanCmd.Flags().BoolVarP(
-		&scanReverse, "reverse", "r", false, "reverse scan")
-	scanCmd.Flags().IntVar(
-		&scanRows, "rows", scanRows, "number of rows to scan in each operation")
-	scanCmd.Flags().IntVar(
-		&scanValueSize, "value", scanValueSize, "size of values to scan")
-
-	ycsbCmd.Flags().IntVar(
-		&ycsbConfig.batch, "batch", 1,
-		"Number of keys to read/insert in each operation")
-	ycsbCmd.Flags().Int64Var(
-		&ycsbConfig.cycleLength, "cycle-length", math.MaxInt64,
-		"Number of keys repeatedly accessed by each writer")
-	ycsbCmd.Flags().IntVar(
-		&ycsbConfig.minBlockBytes, "min-block-bytes", 1,
-		"Minimum amount of raw data written with each insertion")
-	ycsbCmd.Flags().IntVar(
-		&ycsbConfig.maxBlockBytes, "max-block-bytes", 1,
-		"Maximum amount of raw data written with each insertion")
-	ycsbCmd.Flags().Uint64VarP(
-		&ycsbConfig.numOps, "num-ops", "n", 0, "maximum number of operations (0 means unlimited)")
-	ycsbCmd.Flags().IntVar(
-		&ycsbConfig.readPercent, "read-percent", 0,
-		"Percent (0-100) of operations that are reads of existing keys")
-	ycsbCmd.Flags().Int64Var(
-		&ycsbConfig.seed, "seed", 1, "Key hash seed")
-	ycsbCmd.Flags().BoolVar(
-		&ycsbConfig.sequential, "sequential", false,
-		"Pick keys sequentially instead of uniformly at random")
-	ycsbCmd.Flags().StringVar(
-		&ycsbConfig.writeSeq, "write-seq", "",
-		"Initial write sequence value. Can be used to use the data produced by a previous run. "+
-			"It has to be of the form (R|S)<number>, where S implies that it was taken from a "+
-			"previous --sequential run and R implies a previous random run.")
-	ycsbCmd.Flags().Float64Var(
-		&ycsbConfig.targetCompressionRatio, "target-compression-ratio", 1.0,
-		"Target compression ratio for data blocks. Must be >= 1.0")
-
 	if err := rootCmd.Execute(); err != nil {
 		// Cobra has already printed the error message.
 		os.Exit(1)
diff --git a/cmd/pebble/mvcc.go b/cmd/pebble/mvcc.go
index cc5a7b6..d9ddbe5 100644
--- a/cmd/pebble/mvcc.go
+++ b/cmd/pebble/mvcc.go
@@ -90,7 +90,7 @@ func mvccEncode(dst, key []byte, walltime uint64, logical uint32) []byte {
 	return dst
 }
 
-func mvccForwardScan(d *pebble.DB, start, end, ts []byte) int {
+func mvccForwardScan(d *pebble.DB, start, end, ts []byte) (int, int64) {
 	it := d.NewIter(&pebble.IterOptions{
 		LowerBound: mvccEncode(nil, start, 0, 0),
 		UpperBound: mvccEncode(nil, end, 0, 0),
@@ -99,6 +99,7 @@ func mvccForwardScan(d *pebble.DB, start, end, ts []byte) int {
 
 	var data bytealloc.A
 	var count int
+	var nbytes int64
 
 	for valid := it.First(); valid; valid = it.Next() {
 		key, keyTS, _ := mvccSplitKey(it.Key())
@@ -107,11 +108,12 @@ func mvccForwardScan(d *pebble.DB, start, end, ts []byte) int {
 			data, _ = data.Copy(it.Value())
 		}
 		count++
+		nbytes += int64(len(it.Key()) + len(it.Value()))
 	}
-	return count
+	return count, nbytes
 }
 
-func mvccReverseScan(d *pebble.DB, start, end, ts []byte) int {
+func mvccReverseScan(d *pebble.DB, start, end, ts []byte) (int, int64) {
 	it := d.NewIter(&pebble.IterOptions{
 		LowerBound: mvccEncode(nil, start, 0, 0),
 		UpperBound: mvccEncode(nil, end, 0, 0),
@@ -120,6 +122,7 @@ func mvccReverseScan(d *pebble.DB, start, end, ts []byte) int {
 
 	var data bytealloc.A
 	var count int
+	var nbytes int64
 
 	for valid := it.Last(); valid; valid = it.Prev() {
 		key, keyTS, _ := mvccSplitKey(it.Key())
@@ -128,6 +131,7 @@ func mvccReverseScan(d *pebble.DB, start, end, ts []byte) int {
 			data, _ = data.Copy(it.Value())
 		}
 		count++
+		nbytes += int64(len(it.Key()) + len(it.Value()))
 	}
-	return count
+	return count, nbytes
 }
diff --git a/cmd/pebble/random.go b/cmd/pebble/random.go
index d8b8c14..5afbed7 100644
--- a/cmd/pebble/random.go
+++ b/cmd/pebble/random.go
@@ -7,17 +7,70 @@ package main
 import (
 	"crypto/sha1"
 	"encoding/binary"
+	"fmt"
 	"hash"
+	"regexp"
+	"strconv"
+	"strings"
 	"sync/atomic"
 	"time"
 
+	"github.com/petermattis/pebble/internal/randvar"
 	"golang.org/x/exp/rand"
 )
 
+var randVarRE = regexp.MustCompile(`^(?:(uniform|zipf):)?(\d+)(?:-(\d+))?$`)
+
+func parseRandVarSpec(d string) (randvar.Static, error) {
+	m := randVarRE.FindStringSubmatch(d)
+	if m == nil {
+		return nil, fmt.Errorf("invalid random var spec: %s", d)
+	}
+
+	min, err := strconv.Atoi(m[2])
+	if err != nil {
+		return nil, err
+	}
+	max := min
+	if m[3] != "" {
+		max, err = strconv.Atoi(m[3])
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	switch strings.ToLower(m[1]) {
+	case "", "uniform":
+		return randvar.NewUniform(nil, uint64(min), uint64(max)), nil
+	case "zipf":
+		return randvar.NewZipf(nil, uint64(min), uint64(max), 0.99)
+	default:
+		return nil, fmt.Errorf("unknown distribution: %s", m[1])
+	}
+}
+
+func parseValuesSpec(v string) (randvar.Static, float64, error) {
+	parts := strings.Split(v, "/")
+	if len(parts) == 0 || len(parts) > 2 {
+		return nil, 0, fmt.Errorf("invalid values spec: %s", v)
+	}
+	r, err := parseRandVarSpec(parts[0])
+	if err != nil {
+		return nil, 0, err
+	}
+	targetCompression := 1.0
+	if len(parts) == 2 {
+		targetCompression, err = strconv.ParseFloat(parts[1], 64)
+		if err != nil {
+			return nil, 0, err
+		}
+	}
+	return r, targetCompression, nil
+}
+
 func randomBlock(
-	r *rand.Rand, minBlockBytes, maxBlockBytes int, targetCompressionRatio float64,
+	r *rand.Rand, size int, targetCompressionRatio float64,
 ) []byte {
-	size := r.Intn(maxBlockBytes-minBlockBytes+1) + minBlockBytes
 	uniqueSize := int(float64(size) / targetCompressionRatio)
 	if uniqueSize < 1 {
 		uniqueSize = 1
diff --git a/cmd/pebble/scan.go b/cmd/pebble/scan.go
index 11dca1f..7681eab 100644
--- a/cmd/pebble/scan.go
+++ b/cmd/pebble/scan.go
@@ -17,11 +17,11 @@ import (
 	"golang.org/x/exp/rand"
 )
 
-var (
-	scanRows      = 100
-	scanValueSize = 8
-	scanReverse   = false
-)
+var scanConfig struct {
+	reverse bool
+	rows    string
+	values  string
+}
 
 var scanCmd = &cobra.Command{
 	Use:   "scan <dir>",
@@ -31,9 +31,21 @@ var scanCmd = &cobra.Command{
 	Run:   runScan,
 }
 
+func init() {
+	scanCmd.Flags().BoolVarP(
+		&scanConfig.reverse, "reverse", "r", false, "reverse scan")
+	scanCmd.Flags().StringVar(
+		&scanConfig.rows, "rows", "100", "number of rows to scan in each operation")
+	scanCmd.Flags().StringVar(
+		&scanConfig.values, "values", "8",
+		"value size distribution [{zipf,uniform}:]min[-max][/<target-compression>]")
+}
+
 func runScan(cmd *cobra.Command, args []string) {
 	var (
+		bytes       int64
 		scanned     int64
+		lastBytes   int64
 		lastScanned int64
 		lastElapsed time.Duration
 	)
@@ -43,26 +55,32 @@ func runScan(cmd *cobra.Command, args []string) {
 		opts = pebble.NoSync
 	}
 
+	rowDist, err := parseRandVarSpec(scanConfig.rows)
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
+
+	valueDist, targetCompression, err := parseValuesSpec(scanConfig.values)
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
+
 	runTest(args[0], test{
 		init: func(d *pebble.DB, wg *sync.WaitGroup) {
 			const count = 100000
 			const batch = 1000
 
 			rng := rand.New(rand.NewSource(1449168817))
-			randBytes := func(size int) []byte {
-				data := make([]byte, size)
-				for i := range data {
-					data[i] = byte(rng.Int() & 0xff)
-				}
-				return data
-			}
 			keys := make([][]byte, count)
 
 			for i := 0; i < count; {
 				b := d.NewBatch()
 				for end := i + batch; i < end; i++ {
 					keys[i] = mvccEncode(nil, encodeUint32Ascending([]byte("key-"), uint32(i)), uint64(i+1), 0)
-					value := randBytes(scanValueSize)
+					length := int(valueDist.Uint64())
+					value := randomBlock(rng, length, targetCompression)
 					if err := b.Set(keys[i], value, nil); err != nil {
 						log.Fatal(err)
 					}
@@ -87,21 +105,24 @@ func runScan(cmd *cobra.Command, args []string) {
 					minTS := encodeUint64Ascending(nil, math.MaxUint64)
 
 					for {
-						startIdx := rng.Int31n(int32(len(keys) - scanRows))
+						rows := int(rowDist.Uint64())
+						startIdx := rng.Int31n(int32(len(keys) - rows))
 						startKey := encodeUint32Ascending(startKeyBuf[:4], uint32(startIdx))
-						endKey := encodeUint32Ascending(endKeyBuf[:4], uint32(startIdx+int32(scanRows)))
+						endKey := encodeUint32Ascending(endKeyBuf[:4], uint32(startIdx+int32(rows)))
 
 						var count int
-						if scanReverse {
-							count = mvccReverseScan(d, startKey, endKey, minTS)
+						var nbytes int64
+						if scanConfig.reverse {
+							count, nbytes = mvccReverseScan(d, startKey, endKey, minTS)
 						} else {
-							count = mvccForwardScan(d, startKey, endKey, minTS)
+							count, nbytes = mvccForwardScan(d, startKey, endKey, minTS)
 						}
 
-						if count != scanRows {
-							log.Fatalf("scanned %d, expected %d\n", count, scanRows)
+						if count != rows {
+							log.Fatalf("scanned %d, expected %d\n", count, rows)
 						}
 
+						atomic.AddInt64(&bytes, nbytes)
 						atomic.AddInt64(&scanned, int64(count))
 					}
 				}(i)
@@ -113,26 +134,29 @@ func runScan(cmd *cobra.Command, args []string) {
 				fmt.Println("_elapsed_______rows/sec_______MB/sec_______ns/row")
 			}
 
-			cur := atomic.LoadInt64(&scanned)
+			curBytes := atomic.LoadInt64(&bytes)
+			curScanned := atomic.LoadInt64(&scanned)
 			dur := elapsed - lastElapsed
 			fmt.Printf("%8s %14.1f %12.1f %12.1f\n",
 				time.Duration(elapsed.Seconds()+0.5)*time.Second,
-				float64(cur-lastScanned)/dur.Seconds(),
-				float64(int64(scanValueSize)*(cur-lastScanned))/(dur.Seconds()*(1<<20)),
-				float64(dur)/float64(cur-lastScanned),
+				float64(curScanned-lastScanned)/dur.Seconds(),
+				float64(curBytes-lastBytes)/(dur.Seconds()*(1<<20)),
+				float64(dur)/float64(curScanned-lastScanned),
 			)
-			lastScanned = cur
+			lastBytes = curBytes
+			lastScanned = curScanned
 			lastElapsed = elapsed
 		},
 
 		done: func(elapsed time.Duration) {
-			cur := atomic.LoadInt64(&scanned)
+			curBytes := atomic.LoadInt64(&bytes)
+			curScanned := atomic.LoadInt64(&scanned)
 			fmt.Println("\n_elapsed___ops/sec(cum)__MB/sec(cum)__ns/row(avg)")
 			fmt.Printf("%7.1fs %14.1f %12.1f %12.1f\n\n",
 				elapsed.Seconds(),
-				float64(cur)/elapsed.Seconds(),
-				float64(int64(scanValueSize)*cur)/(elapsed.Seconds()*(1<<20)),
-				float64(elapsed)/float64(cur),
+				float64(curScanned)/elapsed.Seconds(),
+				float64(curBytes)/(elapsed.Seconds()*(1<<20)),
+				float64(elapsed)/float64(curScanned),
 			)
 		},
 	})
diff --git a/cmd/pebble/sync.go b/cmd/pebble/sync.go
index 1a26c25..2abecbd 100644
--- a/cmd/pebble/sync.go
+++ b/cmd/pebble/sync.go
@@ -16,6 +16,12 @@ import (
 	"golang.org/x/exp/rand"
 )
 
+var syncConfig struct {
+	batch   string
+	walOnly bool
+	values  string
+}
+
 var syncCmd = &cobra.Command{
 	Use:   "sync <dir>",
 	Short: "run the sync benchmark",
@@ -24,6 +30,17 @@ var syncCmd = &cobra.Command{
 	Run:   runSync,
 }
 
+func init() {
+	syncCmd.Flags().StringVar(
+		&syncConfig.batch, "batch", "5",
+		"batch size distribution [{zipf,uniform}:]min[-max]")
+	syncCmd.Flags().BoolVar(
+		&syncConfig.walOnly, "wal-only", false, "write data only to the WAL")
+	syncCmd.Flags().StringVar(
+		&syncConfig.values, "values", "uniform:60-80/1.0",
+		"value size distribution [{zipf,uniform}:]min[-max][/<target-compression>]")
+}
+
 func runSync(cmd *cobra.Command, args []string) {
 	reg := newHistogramRegistry()
 	var bytes, lastBytes uint64
@@ -33,6 +50,18 @@ func runSync(cmd *cobra.Command, args []string) {
 		opts = pebble.NoSync
 	}
 
+	batchDist, err := parseRandVarSpec(syncConfig.batch)
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
+
+	valueDist, targetCompression, err := parseValuesSpec(syncConfig.values)
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
+
 	runTest(args[0], test{
 		init: func(d *pebble.DB, wg *sync.WaitGroup) {
 			wg.Add(concurrency)
@@ -48,10 +77,12 @@ func runSync(cmd *cobra.Command, args []string) {
 						start := time.Now()
 						b := d.NewBatch()
 						var n uint64
-						for j := 0; j < 5; j++ {
-							block := randomBlock(rand, 60, 80, 1.0)
+						count := int(batchDist.Uint64())
+						for j := 0; j < count; j++ {
+							length := int(valueDist.Uint64())
+							block := randomBlock(rand, length, targetCompression)
 
-							if walOnly {
+							if syncConfig.walOnly {
 								if err := b.LogData(block, nil); err != nil {
 									log.Fatal(err)
 								}
diff --git a/cmd/pebble/ycsb.go b/cmd/pebble/ycsb.go
index 2b09629..71d8761 100644
--- a/cmd/pebble/ycsb.go
+++ b/cmd/pebble/ycsb.go
@@ -7,190 +7,477 @@ package main
 import (
 	"fmt"
 	"log"
-	"math"
 	"strconv"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 
 	"github.com/petermattis/pebble"
+	"github.com/petermattis/pebble/internal/ackseq"
+	"github.com/petermattis/pebble/internal/randvar"
 	"github.com/spf13/cobra"
+	"golang.org/x/exp/rand"
+)
+
+const (
+	ycsbInsert = iota
+	ycsbRead
+	ycsbScan
+	ycsbUpdate
+	ycsbNumOps
 )
 
 var ycsbConfig struct {
-	batch                  int
-	cycleLength            int64
-	minBlockBytes          int
-	maxBlockBytes          int
-	numOps                 uint64
-	readPercent            int
-	seed                   int64
-	sequential             bool
-	writeSeq               string
-	targetCompressionRatio float64
+	batch       string
+	keys        string
+	initialKeys int
+	numOps      uint64
+	scans       string
+	workload    string
+	values      string
 }
 
 var ycsbCmd = &cobra.Command{
 	Use:   "ycsb <dir>",
-	Short: "Run customizable YCSB-like workload. Does not yet offer the standard workloads.",
-	Long:  "",
-	Args:  cobra.ExactArgs(1),
-	RunE:  runYcsb,
+	Short: "run customizable YCSB workload",
+	Long: `
+Run a customizable YCSB workload. The workload is specified by the --workload
+flag which can take either one of the standard workload mixes (A-F), or
+customizable workload fixes specified as a command separated list of op=weight
+pairs. For example, --workload=read=50,update=50 performs a workload composed
+of 50% reads and 50% updates. This is identical to the standard workload A.
+
+The --batch, --scans, and --values flags take the specification for a random
+variable: [<type>:]<min>[-<max>]. The <type> parameter must be one of "uniform"
+or "zipf". If <type> is omitted, a uniform distribution is used. If <max> is
+omitted it is set to the same value as <min>. The specification "1000" results
+in a constant 1000. The specification "10-100" results in a uniformly random
+variable in the range [10,100). The specification "zipf(10,100)" results in a
+zipf distribution with a minimum value of 10 and a maximum value of 100.
+
+The --batch flag controls the size of batches used for insert and update
+operations. The --scans flag controls the number of iterations performed by a
+scan operation. Read operations always read a single key.
+
+The --values flag provides for an optional "/<target-compression-ratio>"
+suffix. The default target compression ratio is 1.0 (i.e. incompressible random
+data). A value of 2 will cause random data to be generated that should compress
+to 50% of its uncompressed size.
+
+Standard workloads:
+
+  A:  50% reads   /  50% updates
+  B:  95% reads   /   5% updates
+  C: 100% reads
+  D:  95% reads   /   5% inserts
+  E:  95% scans   /   5% inserts
+  F: 100% inserts
+`,
+	Args: cobra.ExactArgs(1),
+	RunE: runYcsb,
 }
 
-func runYcsb(cmd *cobra.Command, args []string) error {
-	// benchmark-wide state
-	var (
-		reg             *histogramRegistry
-		seq             sequence
-		numSuccess      uint64
-		numEmptyResults int64
-	)
-
-	if ycsbConfig.maxBlockBytes < ycsbConfig.minBlockBytes {
-		return fmt.Errorf("Value of 'max-block-bytes' (%d) must be greater than or equal to value of 'min-block-bytes' (%d)",
-			ycsbConfig.maxBlockBytes, ycsbConfig.minBlockBytes)
-	}
-	if ycsbConfig.readPercent > 100 || ycsbConfig.readPercent < 0 {
-		return fmt.Errorf("'read-percent' must be in range [0, 100]")
-	}
-	if ycsbConfig.targetCompressionRatio < 1.0 || math.IsNaN(ycsbConfig.targetCompressionRatio) {
-		return fmt.Errorf("'target-compression-ratio' must be a number >= 1.0")
-	}
-	if ycsbConfig.writeSeq != "" {
-		first := ycsbConfig.writeSeq[0]
-		if len(ycsbConfig.writeSeq) < 2 || (first != 'R' && first != 'S') {
-			return fmt.Errorf("--write-seq has to be of the form '(R|S)<num>'")
+func init() {
+	ycsbCmd.Flags().StringVar(
+		&ycsbConfig.batch, "batch", "1",
+		"batch size distribution [{zipf,uniform}:]min[-max]")
+	ycsbCmd.Flags().StringVar(
+		&ycsbConfig.keys, "keys", "zipf", "latest, uniform, or zipf")
+	ycsbCmd.Flags().IntVar(
+		&ycsbConfig.initialKeys, "initial-keys", 10000,
+		"initial number of keys to insert before beginning workload")
+	ycsbCmd.Flags().Uint64VarP(
+		&ycsbConfig.numOps, "num-ops", "n", 0,
+		"maximum number of operations (0 means unlimited)")
+	ycsbCmd.Flags().StringVar(
+		&ycsbConfig.scans, "scans", "zipf:1-1000",
+		"scan length distribution [{zipf,uniform}:]min[-max]")
+	ycsbCmd.Flags().StringVar(
+		&ycsbConfig.workload, "workload", "B",
+		"workload type (A-F) or spec (read=X,update=Y,...)")
+	ycsbCmd.Flags().StringVar(
+		&ycsbConfig.values, "values", "1000",
+		"value size distribution [{zipf,uniform}:]min[-max][/<target-compression>]")
+}
+
+type ycsbWeights []float64
+
+func (w ycsbWeights) get(i int) float64 {
+	if i >= len(w) {
+		return 0
+	}
+	return w[i]
+}
+
+var ycsbWorkloads = map[string]ycsbWeights{
+	"A": ycsbWeights{
+		ycsbRead:   0.5,
+		ycsbUpdate: 0.5,
+	},
+	"B": ycsbWeights{
+		ycsbRead:   0.95,
+		ycsbUpdate: 0.05,
+	},
+	"C": ycsbWeights{
+		ycsbRead: 1.0,
+	},
+	"D": ycsbWeights{
+		ycsbInsert: 0.05,
+		ycsbRead:   0.95,
+		// TODO(peter): default to skewed-latest distribution.
+	},
+	"E": ycsbWeights{
+		ycsbInsert: 0.05,
+		ycsbScan:   0.95,
+	},
+	"F": ycsbWeights{
+		ycsbInsert: 1.0,
+		// TODO(peter): the real workload is read-modify-write.
+	},
+}
+
+func ycsbParseWorkload(w string) (ycsbWeights, error) {
+	if weights := ycsbWorkloads[w]; weights != nil {
+		return weights, nil
+	}
+	iWeights := make([]int, 4)
+	for _, p := range strings.Split(w, ",") {
+		parts := strings.Split(p, "=")
+		if len(parts) != 2 {
+			return nil, fmt.Errorf("malformed weights: %s", w)
 		}
-		rest := ycsbConfig.writeSeq[1:]
-		var err error
-		seq.val, err = strconv.ParseInt(rest, 10, 64)
+		weight, err := strconv.Atoi(parts[1])
 		if err != nil {
-			return fmt.Errorf("--write-seq has to be of the form '(R|S)<num>'")
+			return nil, err
 		}
-		if first == 'R' && ycsbConfig.sequential {
-			return fmt.Errorf("--sequential incompatible with a Random --write-seq")
-		}
-		if first == 'S' && !ycsbConfig.sequential {
-			return fmt.Errorf("--sequential=false incompatible with a Sequential --write-seq")
+		switch parts[0] {
+		case "insert":
+			iWeights[ycsbInsert] = weight
+		case "read":
+			iWeights[ycsbRead] = weight
+		case "scan":
+			iWeights[ycsbScan] = weight
+		case "update":
+			iWeights[ycsbUpdate] = weight
 		}
 	}
 
-	seq.cycleLength = ycsbConfig.cycleLength
-	seq.seed = ycsbConfig.seed
-	reg = newHistogramRegistry()
+	var sum int
+	for _, w := range iWeights {
+		sum += w
+	}
+	if sum == 0 {
+		return nil, fmt.Errorf("zero weight specified: %s", w)
+	}
+
+	weights := make(ycsbWeights, 4)
+	for i := range weights {
+		weights[i] = float64(iWeights[i]) / float64(sum)
+	}
+	return weights, nil
+}
+
+func ycsbParseKeyDist(d string) (randvar.Dynamic, error) {
+	switch strings.ToLower(d) {
+	case "latest":
+		return randvar.NewDefaultSkewedLatest(nil)
+	case "uniform":
+		return randvar.NewUniform(nil, 1, uint64(ycsbConfig.initialKeys)), nil
+	case "zipf":
+		return randvar.NewZipf(nil, 1, uint64(ycsbConfig.initialKeys), 0.99)
+	default:
+		return nil, fmt.Errorf("unknown distribution: %s", d)
+	}
+}
+
+func runYcsb(cmd *cobra.Command, args []string) error {
+	weights, err := ycsbParseWorkload(ycsbConfig.workload)
+	if err != nil {
+		return err
+	}
+
+	keyDist, err := ycsbParseKeyDist(ycsbConfig.keys)
+	if err != nil {
+		return err
+	}
+
+	batchDist, err := parseRandVarSpec(ycsbConfig.batch)
+	if err != nil {
+		return err
+	}
+
+	scanDist, err := parseRandVarSpec(ycsbConfig.scans)
+	if err != nil {
+		return err
+	}
+
+	valueDist, targetCompression, err := parseValuesSpec(ycsbConfig.values)
+	if err != nil {
+		return err
+	}
+
+	y := newYcsb(weights, keyDist, batchDist, scanDist, valueDist, targetCompression)
 	runTest(args[0], test{
-		init: func(db *pebble.DB, wg *sync.WaitGroup) {
-			wg.Add(concurrency)
-			for i := 0; i < concurrency; i++ {
-				// per-worker goroutine state
-				var gen keyGenerator
-				if ycsbConfig.sequential {
-					gen = newSequentialGenerator(&seq)
-				} else {
-					gen = newHashGenerator(&seq)
+		init: y.init,
+		tick: y.tick,
+		done: y.done,
+	})
+	return nil
+}
+
+type ycsb struct {
+	writeOpts         *pebble.WriteOptions
+	reg               *histogramRegistry
+	weights           ycsbWeights
+	ops               *randvar.Weighted
+	keyDist           randvar.Dynamic
+	batchDist         randvar.Static
+	scanDist          randvar.Static
+	valueDist         randvar.Static
+	targetCompression float64
+	keyNum            *ackseq.S
+	numOps            uint64
+	numKeys           [ycsbNumOps]uint64
+	prevNumKeys       [ycsbNumOps]uint64
+	opsMap            map[string]int
+	latency           [ycsbNumOps]*namedHistogram
+}
+
+func newYcsb(
+	weights ycsbWeights,
+	keyDist randvar.Dynamic,
+	batchDist, scanDist, valueDist randvar.Static,
+	targetCompression float64,
+) *ycsb {
+	y := &ycsb{
+		reg:               newHistogramRegistry(),
+		ops:               randvar.NewWeighted(nil, weights...),
+		keyDist:           keyDist,
+		batchDist:         batchDist,
+		scanDist:          scanDist,
+		valueDist:         valueDist,
+		targetCompression: targetCompression,
+	}
+	y.writeOpts = pebble.Sync
+	if disableWAL {
+		y.writeOpts = pebble.NoSync
+	}
+
+	y.opsMap = make(map[string]int)
+	maybeRegister := func(op int, name string) *namedHistogram {
+		w := weights.get(op)
+		if w == 0 {
+			return nil
+		}
+		wstr := fmt.Sprint(int(100 * w))
+		fill := strings.Repeat("_", 3-len(wstr))
+		if fill == "" {
+			fill = "_"
+		}
+		fullName := fmt.Sprintf("%s%s%s", name, fill, wstr)
+		y.opsMap[fullName] = op
+		return y.reg.Register(fullName)
+	}
+
+	y.latency[ycsbInsert] = maybeRegister(ycsbInsert, "insert")
+	y.latency[ycsbRead] = maybeRegister(ycsbRead, "read")
+	y.latency[ycsbScan] = maybeRegister(ycsbScan, "scan")
+	y.latency[ycsbUpdate] = maybeRegister(ycsbUpdate, "update")
+	return y
+}
+
+func (y *ycsb) init(db *pebble.DB, wg *sync.WaitGroup) {
+	if ycsbConfig.initialKeys > 0 {
+		rng := randvar.NewRand()
+
+		b := db.NewBatch()
+		for i := 1; i <= ycsbConfig.initialKeys; i++ {
+			if len(b.Repr()) >= 1<<20 {
+				if err := b.Commit(y.writeOpts); err != nil {
+					log.Fatal(err)
 				}
-				readLatency := reg.Register("read")
-				writeLatency := reg.Register("write")
-
-				go func() {
-					defer wg.Done()
-					var raw, buf []byte
-					for {
-						if gen.rand().Intn(100) < ycsbConfig.readPercent {
-							num := gen.readKey()
-							raw = encodeUint64Ascending(raw[:0], uint64(num))
-							key := mvccEncode(buf[:0], raw, 0, 0)
-							start := time.Now()
-							iter := db.NewIter(nil)
-							found := 0
-							for iter.SeekGE(key); iter.Valid(); iter.Next() {
-								found++
-								if found == ycsbConfig.batch {
-									break
-								}
-							}
-							if err := iter.Close(); err != nil {
-								log.Fatal(err)
-							}
-							elapsed := time.Since(start)
-							readLatency.Record(elapsed)
-							if found == 0 {
-								atomic.AddInt64(&numEmptyResults, 1)
-							}
-						} else {
-							start := time.Now()
-							b := db.NewBatch()
-							for i := 0; i < ycsbConfig.batch; i++ {
-								num := gen.writeKey()
-								val := randomBlock(
-									gen.rand(), ycsbConfig.minBlockBytes,
-									ycsbConfig.maxBlockBytes,
-									ycsbConfig.targetCompressionRatio,
-								)
-								raw = encodeUint64Ascending(raw[:0], uint64(num))
-								key := mvccEncode(buf[:0], raw, 0, 0)
-								b.Set(key, val, nil)
-							}
-							err := b.Commit(pebble.Sync)
-							if err != nil {
-								log.Fatal(err)
-							}
-							elapsed := time.Since(start)
-							writeLatency.Record(elapsed)
-						}
-						if ycsbConfig.numOps > 0 &&
-							atomic.AddUint64(&numSuccess, 1) >= ycsbConfig.numOps {
-							break
-						}
-					}
-				}()
-			}
-		},
-		tick: func(elapsed time.Duration, i int) {
-			if i%20 == 0 {
-				fmt.Println("optype__elapsed____ops/sec__p50(ms)__p95(ms)__p99(ms)_pMax(ms)")
-			}
-			reg.Tick(func(tick histogramTick) {
-				h := tick.Hist
-				fmt.Printf("%6s %8s %10.1f %8.1f %8.1f %8.1f %8.1f\n",
-					tick.Name,
-					time.Duration(elapsed.Seconds()+0.5)*time.Second,
-					float64(h.TotalCount())/tick.Elapsed.Seconds(),
-					time.Duration(h.ValueAtQuantile(50)).Seconds()*1000,
-					time.Duration(h.ValueAtQuantile(95)).Seconds()*1000,
-					time.Duration(h.ValueAtQuantile(99)).Seconds()*1000,
-					time.Duration(h.ValueAtQuantile(100)).Seconds()*1000,
-				)
-			})
-		},
-		done: func(elapsed time.Duration) {
-			fmt.Println("\noptype__elapsed_____ops(total)___ops/sec(cum)__avg(ms)__p50(ms)__p95(ms)__p99(ms)_pMax(ms)")
-			reg.Tick(func(tick histogramTick) {
-				h := tick.Cumulative
-				fmt.Printf("%6s %7.1fs %14d %14.1f %8.1f %8.1f %8.1f %8.1f %8.1f\n",
-					tick.Name, elapsed.Seconds(), h.TotalCount(),
-					float64(h.TotalCount())/elapsed.Seconds(),
-					time.Duration(h.Mean()).Seconds()*1000,
-					time.Duration(h.ValueAtQuantile(50)).Seconds()*1000,
-					time.Duration(h.ValueAtQuantile(95)).Seconds()*1000,
-					time.Duration(h.ValueAtQuantile(99)).Seconds()*1000,
-					time.Duration(h.ValueAtQuantile(100)).Seconds()*1000)
-			})
-
-			if empty := atomic.LoadInt64(&numEmptyResults); empty != 0 {
-				fmt.Printf("Number of reads that didn't return any results: %d.\n", empty)
-			}
-			seq := atomic.LoadInt64(&seq.val)
-			var ch string
-			if ycsbConfig.sequential {
-				ch = "S"
-			} else {
-				ch = "R"
+				b = db.NewBatch()
 			}
-			fmt.Printf("Highest sequence written: %d. Can be passed as --write-seq=%s%d to the next run.\n",
-				seq, ch, seq)
-		},
+			_ = b.Set(y.makeKey(uint64(i)), y.randBytes(rng), nil)
+		}
+		if err := b.Commit(y.writeOpts); err != nil {
+			log.Fatal(err)
+		}
+
+		y.keyNum = ackseq.New(uint64(ycsbConfig.initialKeys))
+	}
+
+	wg.Add(concurrency)
+	for i := 0; i < concurrency; i++ {
+		go y.run(db, wg)
+	}
+}
+
+func (y *ycsb) run(db *pebble.DB, wg *sync.WaitGroup) {
+	defer wg.Done()
+
+	rng := randvar.NewRand()
+	for {
+		start := time.Now()
+
+		op := y.ops.Int()
+		switch op {
+		case ycsbInsert:
+			y.insert(db, rng)
+		case ycsbRead:
+			y.read(db, rng)
+		case ycsbScan:
+			y.scan(db, rng)
+		case ycsbUpdate:
+			y.update(db, rng)
+		default:
+			panic("not reached")
+		}
+
+		y.latency[op].Record(time.Since(start))
+		if ycsbConfig.numOps > 0 &&
+			atomic.AddUint64(&y.numOps, 1) >= ycsbConfig.numOps {
+			break
+		}
+	}
+}
+
+func (y *ycsb) hashKey(key uint64) uint64 {
+	// Inlined version of fnv.New64 + Write.
+	const offset64 = 14695981039346656037
+	const prime64 = 1099511628211
+
+	h := uint64(offset64)
+	for i := 0; i < 8; i++ {
+		h *= prime64
+		h ^= uint64(key & 0xff)
+		key >>= 8
+	}
+	return h
+}
+
+func (y *ycsb) makeKey(keyNum uint64) []byte {
+	key := make([]byte, 4, 24)
+	copy(key, "user")
+	return strconv.AppendUint(key, y.hashKey(keyNum), 10)
+}
+
+func (y *ycsb) nextReadKey() []byte {
+	// NB: the range of values returned by keyDist is tied to the range returned
+	// by keyNum.Base. See how these are both incremented by ycsb.insert().
+	keyNum := y.keyDist.Uint64()
+	return y.makeKey(keyNum)
+}
+
+func (y *ycsb) randBytes(rng *rand.Rand) []byte {
+	length := int(y.valueDist.Uint64())
+	return randomBlock(rng, length, y.targetCompression)
+}
+
+func (y *ycsb) insert(db *pebble.DB, rng *rand.Rand) {
+	count := y.batchDist.Uint64()
+	keyNums := make([]uint64, count)
+
+	b := db.NewBatch()
+	for i := range keyNums {
+		keyNums[i] = y.keyNum.Next()
+		_ = b.Set(y.makeKey(keyNums[i]), y.randBytes(rng), nil)
+	}
+	if err := b.Commit(y.writeOpts); err != nil {
+		log.Fatal(err)
+	}
+	atomic.AddUint64(&y.numKeys[ycsbInsert], uint64(len(keyNums)))
+
+	for i := range keyNums {
+		delta, err := y.keyNum.Ack(keyNums[i])
+		if err != nil {
+			log.Fatal(err)
+		}
+		if delta > 0 {
+			y.keyDist.IncMax(delta)
+		}
+	}
+}
+
+func (y *ycsb) read(db *pebble.DB, rng *rand.Rand) {
+	key := y.nextReadKey()
+	iter := db.NewIter(nil)
+	iter.SeekGE(key)
+	if err := iter.Close(); err != nil {
+		log.Fatal(err)
+	}
+	atomic.AddUint64(&y.numKeys[ycsbRead], 1)
+}
+
+func (y *ycsb) scan(db *pebble.DB, rng *rand.Rand) {
+	i, count := 0, y.scanDist.Uint64()
+	iter := db.NewIter(nil)
+	key := y.nextReadKey()
+	for valid := iter.SeekGE([]byte(key)); valid; valid = iter.Next() {
+		i++
+		if i >= int(count) {
+			break
+		}
+	}
+	if err := iter.Close(); err != nil {
+		log.Fatal(err)
+	}
+	atomic.AddUint64(&y.numKeys[ycsbScan], count)
+}
+
+func (y *ycsb) update(db *pebble.DB, rng *rand.Rand) {
+	count := int(y.batchDist.Uint64())
+	b := db.NewBatch()
+	for i := 0; i < count; i++ {
+		_ = b.Set(y.nextReadKey(), y.randBytes(rng), nil)
+	}
+	if err := b.Commit(y.writeOpts); err != nil {
+		log.Fatal(err)
+	}
+	atomic.AddUint64(&y.numKeys[ycsbUpdate], uint64(count))
+}
+
+func (y *ycsb) tick(elapsed time.Duration, i int) {
+	if i%20 == 0 {
+		fmt.Println("____optype__elapsed____ops/sec___keys/sec__p50(ms)__p95(ms)__p99(ms)_pMax(ms)")
+	}
+	y.reg.Tick(func(tick histogramTick) {
+		op := y.opsMap[tick.Name]
+		numKeys := atomic.LoadUint64(&y.numKeys[op])
+		h := tick.Hist
+
+		fmt.Printf("%10s %8s %10.1f %10.1f %8.1f %8.1f %8.1f %8.1f\n",
+			tick.Name,
+			time.Duration(elapsed.Seconds()+0.5)*time.Second,
+			float64(h.TotalCount())/tick.Elapsed.Seconds(),
+			float64(numKeys-y.prevNumKeys[op])/tick.Elapsed.Seconds(),
+			time.Duration(h.ValueAtQuantile(50)).Seconds()*1000,
+			time.Duration(h.ValueAtQuantile(95)).Seconds()*1000,
+			time.Duration(h.ValueAtQuantile(99)).Seconds()*1000,
+			time.Duration(h.ValueAtQuantile(100)).Seconds()*1000,
+		)
+
+		y.prevNumKeys[op] = numKeys
 	})
-	return nil
+}
+
+func (y *ycsb) done(elapsed time.Duration) {
+	fmt.Println("\n____optype__elapsed_____ops(total)___ops/sec(cum)__keys/sec(cum)__avg(ms)__p50(ms)__p95(ms)__p99(ms)_pMax(ms)")
+	y.reg.Tick(func(tick histogramTick) {
+		op := y.opsMap[tick.Name]
+		numKeys := atomic.LoadUint64(&y.numKeys[op])
+		h := tick.Cumulative
+
+		fmt.Printf("%10s %7.1fs %14d %14.1f %14.1f %8.1f %8.1f %8.1f %8.1f %8.1f\n",
+			tick.Name, elapsed.Seconds(), h.TotalCount(),
+			float64(h.TotalCount())/elapsed.Seconds(),
+			float64(numKeys)/elapsed.Seconds(),
+			time.Duration(h.Mean()).Seconds()*1000,
+			time.Duration(h.ValueAtQuantile(50)).Seconds()*1000,
+			time.Duration(h.ValueAtQuantile(95)).Seconds()*1000,
+			time.Duration(h.ValueAtQuantile(99)).Seconds()*1000,
+			time.Duration(h.ValueAtQuantile(100)).Seconds()*1000)
+	})
+	fmt.Println()
 }
diff --git a/compaction_picker.go b/compaction_picker.go
index f313fe0..2d75ea0 100644
--- a/compaction_picker.go
+++ b/compaction_picker.go
@@ -49,14 +49,14 @@ func (p *compactionPicker) compactionNeeded() bool {
 func (p *compactionPicker) initLevelMaxBytes(v *version, opts *Options) {
 	// Determine the first non-empty level and the maximum size of any level.
 	firstNonEmptyLevel := -1
-	var maxLevelSize int64
+	var bottomLevelSize int64
 	for level := 1; level < numLevels; level++ {
 		levelSize := int64(totalSize(v.files[level]))
-		if levelSize > 0 && firstNonEmptyLevel == -1 {
-			firstNonEmptyLevel = level
-		}
-		if maxLevelSize < levelSize {
-			maxLevelSize = levelSize
+		if levelSize > 0 {
+			if firstNonEmptyLevel == -1 {
+				firstNonEmptyLevel = level
+			}
+			bottomLevelSize = levelSize
 		}
 	}
 
@@ -67,7 +67,7 @@ func (p *compactionPicker) initLevelMaxBytes(v *version, opts *Options) {
 		p.levelMaxBytes[level] = math.MaxInt64
 	}
 
-	if maxLevelSize == 0 {
+	if bottomLevelSize == 0 {
 		// No levels for L1 and up contain any data. Target L0 compactions for the
 		// last level.
 		p.baseLevel = numLevels - 1
@@ -79,17 +79,15 @@ func (p *compactionPicker) initLevelMaxBytes(v *version, opts *Options) {
 	baseBytesMax := opts.LBaseMaxBytes
 	baseBytesMin := int64(float64(baseBytesMax) / levelMultiplier)
 
-	curLevelSize := maxLevelSize
+	curLevelSize := bottomLevelSize
 	for level := numLevels - 2; level >= firstNonEmptyLevel; level-- {
 		curLevelSize = int64(float64(curLevelSize) / levelMultiplier)
 	}
 
-	var baseLevelSize int64
 	if curLevelSize <= baseBytesMin {
-		// If we make target size of last level to be maxLevelSize, target size of
+		// If we make target size of last level to be bottomLevelSize, target size of
 		// the first non-empty level would be smaller than baseBytesMin. We set it
 		// be baseBytesMin.
-		baseLevelSize = baseBytesMin + 1
 		p.baseLevel = firstNonEmptyLevel
 	} else {
 		// Compute base level (where L0 data is compacted to).
@@ -98,23 +96,29 @@ func (p *compactionPicker) initLevelMaxBytes(v *version, opts *Options) {
 			p.baseLevel--
 			curLevelSize = int64(float64(curLevelSize) / levelMultiplier)
 		}
-		if curLevelSize > baseBytesMax {
-			baseLevelSize = baseBytesMax
-		} else {
-			baseLevelSize = curLevelSize
-		}
 	}
 
-	levelSize := baseLevelSize
+	var smoothedLevelMultiplier float64
+	if p.baseLevel < numLevels-1 {
+		smoothedLevelMultiplier = math.Pow(
+			float64(bottomLevelSize)/float64(baseBytesMax),
+			1.0/float64(numLevels-p.baseLevel-1))
+	} else {
+		smoothedLevelMultiplier = 1.0
+	}
+
+	levelSize := float64(baseBytesMax)
 	for level := p.baseLevel; level < numLevels; level++ {
-		if level > p.baseLevel {
-			if levelSize > 0 && float64(math.MaxInt64/levelSize) >= levelMultiplier {
-				levelSize = int64(float64(levelSize) * levelMultiplier)
-			}
+		if level > p.baseLevel && levelSize > 0 {
+			levelSize *= smoothedLevelMultiplier
 		}
-		p.levelMaxBytes[level] = levelSize
-		if p.levelMaxBytes[level] < baseBytesMax {
-			p.levelMaxBytes[level] = baseBytesMax
+		// Round the result since test cases use small target level sizes, which
+		// can be impacted by floating-point imprecision + integer truncation.
+		roundedLevelSize := math.Round(levelSize)
+		if roundedLevelSize > float64(math.MaxInt64) {
+			p.levelMaxBytes[level] = math.MaxInt64
+		} else {
+			p.levelMaxBytes[level] = int64(roundedLevelSize)
 		}
 	}
 }
diff --git a/internal/ackseq/ackseq.go b/internal/ackseq/ackseq.go
new file mode 100644
index 0000000..04f3920
--- /dev/null
+++ b/internal/ackseq/ackseq.go
@@ -0,0 +1,83 @@
+// Copyright 2019 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package ackseq
+
+import (
+	"fmt"
+	"sync"
+	"sync/atomic"
+)
+
+const (
+	// The window size constants. These values specify a window that can hold ~1m
+	// pending unacknowledged sequence numbers using 128 KB of memory.
+	windowSize  = 1 << 20
+	windowMask  = windowSize - 1
+	windowBytes = (windowSize + 7) / 8
+)
+
+// S keeps track of the largest sequence number such that all sequence numbers
+// in the range [0,v) have been acknowledged.
+type S struct {
+	next uint64
+	mu   struct {
+		sync.Mutex
+		base   uint64
+		window [windowBytes]uint8
+	}
+}
+
+// New creates a new acknowledged sequence tracker with the specified base
+// sequence number. All of the sequence numbers in the range [0,base) are
+// considered acknowledged. Next() will return base upon first call.
+func New(base uint64) *S {
+	s := &S{
+		next: base,
+	}
+	s.mu.base = base
+	return s
+}
+
+// Next returns the next sequence number to use.
+func (s *S) Next() uint64 {
+	return atomic.AddUint64(&s.next, 1) - 1
+}
+
+// Ack acknowledges the specified seqNum, adjusting base as necessary,
+// returning the number of newly acknowledged sequence numbers.
+func (s *S) Ack(seqNum uint64) (int, error) {
+	s.mu.Lock()
+	if s.getLocked(seqNum) {
+		defer s.mu.Unlock()
+		return 0, fmt.Errorf(
+			"pending acks exceeds window size: %d has been acked, but %d has not",
+			seqNum, s.mu.base)
+	}
+
+	var count int
+	s.setLocked(seqNum)
+	for s.getLocked(s.mu.base) {
+		s.clearLocked(s.mu.base)
+		s.mu.base++
+		count++
+	}
+	s.mu.Unlock()
+	return count, nil
+}
+
+func (s *S) getLocked(seqNum uint64) bool {
+	bit := seqNum & windowMask
+	return (s.mu.window[bit/8] & (1 << (bit % 8))) != 0
+}
+
+func (s *S) setLocked(seqNum uint64) {
+	bit := seqNum & windowMask
+	s.mu.window[bit/8] |= (1 << (bit % 8))
+}
+
+func (s *S) clearLocked(seqNum uint64) {
+	bit := seqNum & windowMask
+	s.mu.window[bit/8] &^= (1 << (bit % 8))
+}
diff --git a/internal/randvar/deck.go b/internal/randvar/deck.go
new file mode 100644
index 0000000..873348f
--- /dev/null
+++ b/internal/randvar/deck.go
@@ -0,0 +1,62 @@
+// Copyright 2019 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package randvar
+
+import (
+	"sync"
+
+	"golang.org/x/exp/rand"
+)
+
+// Deck is a random number generator that generates numbers in the range
+// [0,len(weights)-1] where the probability of i is
+// weights(i)/sum(weights). Unlike Weighted, the weights are specified as
+// integers and used in a deck-of-cards style random number selection which
+// ensures that each element is returned with a desired frequency within the
+// size of the deck.
+type Deck struct {
+	rng *rand.Rand
+	mu  struct {
+		sync.Mutex
+		index int
+		deck  []int
+	}
+}
+
+// NewDeck returns a new deck random number generator.
+func NewDeck(rng *rand.Rand, weights ...int) *Deck {
+	var sum int
+	for i := range weights {
+		sum += weights[i]
+	}
+	deck := make([]int, 0, sum)
+	for i := range weights {
+		for j := 0; j < weights[i]; j++ {
+			deck = append(deck, i)
+		}
+	}
+	d := &Deck{
+		rng: ensureRand(rng),
+	}
+	d.mu.index = len(deck)
+	d.mu.deck = deck
+	return d
+}
+
+// Int returns a random number in the range [0,len(weights)-1] where the
+// probability of i is weights(i)/sum(weights).
+func (d *Deck) Int() int {
+	d.mu.Lock()
+	if d.mu.index == len(d.mu.deck) {
+		d.rng.Shuffle(len(d.mu.deck), func(i, j int) {
+			d.mu.deck[i], d.mu.deck[j] = d.mu.deck[j], d.mu.deck[i]
+		})
+		d.mu.index = 0
+	}
+	result := d.mu.deck[d.mu.index]
+	d.mu.index++
+	d.mu.Unlock()
+	return result
+}
diff --git a/internal/randvar/rand.go b/internal/randvar/rand.go
new file mode 100644
index 0000000..f7d18a2
--- /dev/null
+++ b/internal/randvar/rand.go
@@ -0,0 +1,23 @@
+// Copyright 2019 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package randvar
+
+import (
+	"time"
+
+	"golang.org/x/exp/rand"
+)
+
+// NewRand creates a new random number generator seeded with the current time.
+func NewRand() *rand.Rand {
+	return rand.New(rand.NewSource(uint64(time.Now().UnixNano())))
+}
+
+func ensureRand(rng *rand.Rand) *rand.Rand {
+	if rng != nil {
+		return rng
+	}
+	return NewRand()
+}
diff --git a/internal/randvar/randvar.go b/internal/randvar/randvar.go
new file mode 100644
index 0000000..db7f763
--- /dev/null
+++ b/internal/randvar/randvar.go
@@ -0,0 +1,20 @@
+// Copyright 2019 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package randvar
+
+// Static models a random variable that pulls from a distribution with static
+// bounds
+type Static interface {
+	Uint64() uint64
+}
+
+// Dynamic models a random variable that pulls from a distribution with an
+// upper bound that can change dynamically using the IncMax method.
+type Dynamic interface {
+	Static
+
+	// Increment the max value the variable will return.
+	IncMax(delta int)
+}
diff --git a/internal/randvar/uniform.go b/internal/randvar/uniform.go
new file mode 100644
index 0000000..6ced2e0
--- /dev/null
+++ b/internal/randvar/uniform.go
@@ -0,0 +1,59 @@
+// Copyright 2018 The Cockroach Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied. See the License for the specific language governing
+// permissions and limitations under the License. See the AUTHORS file
+// for names of contributors.
+
+package randvar
+
+import (
+	"sync"
+
+	"golang.org/x/exp/rand"
+)
+
+// Uniform is a random number generator that generates draws from a uniform
+// distribution.
+type Uniform struct {
+	min uint64
+	mu  struct {
+		sync.Mutex
+		rng *rand.Rand
+		max uint64
+	}
+}
+
+// NewUniform constructs a new Uniform generator with the given
+// parameters. Returns an error if the parameters are outside the accepted
+// range.
+func NewUniform(rng *rand.Rand, min, max uint64) *Uniform {
+	g := &Uniform{min: min}
+	g.mu.rng = ensureRand(rng)
+	g.mu.max = max
+	return g
+}
+
+// IncMax increments max.
+func (g *Uniform) IncMax(delta int) {
+	g.mu.Lock()
+	g.mu.max += uint64(delta)
+	g.mu.Unlock()
+}
+
+// Uint64 returns a random Uint64 between min and max, drawn from a uniform
+// distribution.
+func (g *Uniform) Uint64() uint64 {
+	g.mu.Lock()
+	result := g.mu.rng.Uint64n(g.mu.max-g.min+1) + g.min
+	g.mu.Unlock()
+	return result
+}
diff --git a/internal/randvar/weighted.go b/internal/randvar/weighted.go
new file mode 100644
index 0000000..3294757
--- /dev/null
+++ b/internal/randvar/weighted.go
@@ -0,0 +1,41 @@
+// Copyright 2019 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package randvar
+
+import "golang.org/x/exp/rand"
+
+// Weighted is a random number generator that generates numbers in the range
+// [0,len(weights)-1] where the probability of i is weights(i)/sum(weights).
+type Weighted struct {
+	rng     *rand.Rand
+	sum     float64
+	weights []float64
+}
+
+// NewWeighted returns a new weighted random number generator.
+func NewWeighted(rng *rand.Rand, weights ...float64) *Weighted {
+	var sum float64
+	for i := range weights {
+		sum += weights[i]
+	}
+	return &Weighted{
+		rng:     ensureRand(rng),
+		sum:     sum,
+		weights: weights,
+	}
+}
+
+// Int returns a random number in the range [0,len(weights)-1] where the
+// probability of i is weights(i)/sum(weights).
+func (w *Weighted) Int() int {
+	p := w.rng.Float64() * w.sum
+	for i, weight := range w.weights {
+		if p <= weight {
+			return i
+		}
+		p -= weight
+	}
+	return len(w.weights) - 1
+}
diff --git a/internal/randvar/zipf.go b/internal/randvar/zipf.go
new file mode 100644
index 0000000..f04628f
--- /dev/null
+++ b/internal/randvar/zipf.go
@@ -0,0 +1,140 @@
+// Copyright 2017 The Cockroach Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied. See the License for the specific language governing
+// permissions and limitations under the License. See the AUTHORS file
+// for names of contributors.
+//
+// ZipfGenerator implements the Incrementing Zipfian Random Number Generator from
+// [1]: "Quickly Generating Billion-Record Synthetic Databases"
+// by Gray, Sundaresan, Englert, Baclawski, and Weinberger, SIGMOD 1994.
+
+package randvar
+
+import (
+	"fmt"
+	"math"
+	"sync"
+
+	"golang.org/x/exp/rand"
+)
+
+const (
+	// See https://github.com/brianfrankcooper/YCSB/blob/f886c1e7988f8f4965cb88a1fe2f6bad2c61b56d/core/src/main/java/com/yahoo/ycsb/generator/ScrambledZipfianGenerator.java#L33-L35
+	defaultMax   = 10000000000
+	defaultTheta = 0.99
+	defaultZetaN = 26.46902820178302
+)
+
+// Zipf is a random number generator that generates random numbers from a Zipf
+// distribution. Unlike rand.Zipf, this generator supports incrementing the max
+// parameter without performing an expensive recomputation of the underlying
+// hidden parameters, which is a pattern used in [1] for efficiently generating
+// large volumes of Zipf-distributed records for synthetic data. Second,
+// rand.Zipf only supports theta <= 1, we suppose all values of theta.
+type Zipf struct {
+	// Supplied constants.
+	theta float64
+	min   uint64
+	// Internally computed constants.
+	alpha, zeta2 float64
+	// Mutable state.
+	mu struct {
+		sync.Mutex
+		rng   *rand.Rand
+		max   uint64
+		eta   float64
+		zetaN float64
+	}
+}
+
+// NewDefaultZipf constructs a new Zipf generator with the default parameters.
+func NewDefaultZipf(rng *rand.Rand) (*Zipf, error) {
+	return NewZipf(rng, 1, defaultMax, defaultTheta)
+}
+
+// NewZipf constructs a new Zipf generator with the given parameters.  Returns
+// an error if the parameters are outside the accepted range.
+func NewZipf(rng *rand.Rand, min, max uint64, theta float64) (*Zipf, error) {
+	if min > max {
+		return nil, fmt.Errorf("min %d > max %d", min, max)
+	}
+	if theta < 0.0 || theta == 1.0 {
+		return nil, fmt.Errorf("0 < theta, and theta != 1")
+	}
+
+	z := &Zipf{
+		min:   min,
+		theta: theta,
+	}
+	z.mu.rng = ensureRand(rng)
+	z.mu.max = max
+
+	// Compute hidden parameters.
+	z.zeta2 = computeZetaFromScratch(2, theta)
+	z.mu.zetaN = computeZetaFromScratch(max+1-min, theta)
+	z.alpha = 1.0 / (1.0 - theta)
+	z.mu.eta = (1 - math.Pow(2.0/float64(z.mu.max+1-z.min), 1.0-theta)) / (1.0 - z.zeta2/z.mu.zetaN)
+	return z, nil
+}
+
+// computeZetaIncrementally recomputes zeta(max, theta), assuming that sum =
+// zeta(oldMax, theta). Returns zeta(max, theta), computed incrementally.
+func computeZetaIncrementally(oldMax, max uint64, theta float64, sum float64) float64 {
+	if max < oldMax {
+		panic(fmt.Errorf("unable to increment max backwards!"))
+	}
+	for i := oldMax + 1; i <= max; i++ {
+		sum += 1.0 / math.Pow(float64(i), theta)
+	}
+	return sum
+}
+
+// The function zeta computes the value
+// zeta(n, theta) = (1/1)^theta + (1/2)^theta + (1/3)^theta + ... + (1/n)^theta
+func computeZetaFromScratch(n uint64, theta float64) float64 {
+	if n == defaultMax && theta == defaultTheta {
+		// Precomputed value, borrowed from ScrambledZipfianGenerator.java. This is
+		// quite slow to calculate from scratch due to the large n value.
+		return defaultZetaN
+	}
+	return computeZetaIncrementally(0, n, theta, 0.0)
+}
+
+// IncMax increments max and recomputes the internal values that depend on
+// it. Returns an error if the recomputation failed.
+func (z *Zipf) IncMax(delta int) {
+	z.mu.Lock()
+	oldMax := z.mu.max
+	z.mu.max += uint64(delta)
+	z.mu.zetaN = computeZetaIncrementally(oldMax, z.mu.max, z.theta, z.mu.zetaN)
+	z.mu.eta = (1 - math.Pow(2.0/float64(z.mu.max-z.min), 1.0-z.theta)) / (1.0 - z.zeta2/z.mu.zetaN)
+	z.mu.Unlock()
+}
+
+// Uint64 draws a new value between min and max, with probabilities according
+// to the Zipf distribution.
+func (z *Zipf) Uint64() uint64 {
+	z.mu.Lock()
+	u := z.mu.rng.Float64()
+	uz := u * z.mu.zetaN
+	var result uint64
+	if uz < 1.0 {
+		result = z.min
+	} else if uz < 1.0+math.Pow(0.5, z.theta) {
+		result = z.min + 1
+	} else {
+		spread := float64(z.mu.max + 1 - z.min)
+		result = z.min + uint64(int64(spread*math.Pow(z.mu.eta*u-z.mu.eta+1.0, z.alpha)))
+	}
+	z.mu.Unlock()
+	return result
+}


Test patch:
diff --git a/batch_test.go b/batch_test.go
index cf0f1a4..b7243d0 100644
--- a/batch_test.go
+++ b/batch_test.go
@@ -285,7 +285,7 @@ func TestFlushableBatchIter(t *testing.T) {
 	})
 }
 
-func TestFlushableBatchSeqNum(t *testing.T) {
+func TestFlushableBatch(t *testing.T) {
 	var b *flushableBatch
 	datadriven.RunTest(t, "testdata/flushable_batch", func(d *datadriven.TestData) string {
 		switch d.Cmd {
@@ -307,6 +307,26 @@ func TestFlushableBatchSeqNum(t *testing.T) {
 			b = newFlushableBatch(batch, DefaultComparer)
 			return ""
 
+		case "iter":
+			var opts IterOptions
+			for _, arg := range d.CmdArgs {
+				if len(arg.Vals) != 1 {
+					return fmt.Sprintf("%s: %s=<value>", d.Cmd, arg.Key)
+				}
+				switch arg.Key {
+				case "lower":
+					opts.LowerBound = []byte(arg.Vals[0])
+				case "upper":
+					opts.UpperBound = []byte(arg.Vals[0])
+				default:
+					return fmt.Sprintf("%s: unknown arg: %s", d.Cmd, arg.Key)
+				}
+			}
+
+			iter := b.newIter(&opts)
+			defer iter.Close()
+			return runInternalIterCmd(d, iter)
+
 		case "dump":
 			if len(d.CmdArgs) != 1 || len(d.CmdArgs[0].Vals) != 1 || d.CmdArgs[0].Key != "seq" {
 				return fmt.Sprintf("dump seq=<value>\n")
diff --git a/internal/randvar/deck_test.go b/internal/randvar/deck_test.go
new file mode 100644
index 0000000..a58cd7e
--- /dev/null
+++ b/internal/randvar/deck_test.go
@@ -0,0 +1,20 @@
+// Copyright 2019 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package randvar
+
+import "testing"
+
+func TestDeck(t *testing.T) {
+	d := NewDeck(nil, 10, 20, 20, 0, 30)
+
+	x := make([]int, 10000)
+	for i := range x {
+		x[i] = d.Int()
+	}
+
+	if testing.Verbose() {
+		dumpSamples(x)
+	}
+}
diff --git a/internal/randvar/skewed_latest.go b/internal/randvar/skewed_latest.go
new file mode 100644
index 0000000..59eda3c
--- /dev/null
+++ b/internal/randvar/skewed_latest.go
@@ -0,0 +1,70 @@
+// Copyright 2018 The Cockroach Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied. See the License for the specific language governing
+// permissions and limitations under the License. See the AUTHORS file
+// for names of contributors.
+
+package randvar
+
+import (
+	"sync"
+
+	"golang.org/x/exp/rand"
+)
+
+// SkewedLatest is a random number generator that generates numbers in
+// the range [min, max], but skews it towards max using a zipfian
+// distribution.
+type SkewedLatest struct {
+	mu struct {
+		sync.Mutex
+		max  uint64
+		zipf *Zipf
+	}
+}
+
+// NewDefaultSkewedLatest constructs a new SkewedLatest generator with the
+// default parameters.
+func NewDefaultSkewedLatest(rng *rand.Rand) (*SkewedLatest, error) {
+	return NewSkewedLatest(rng, 1, defaultMax, defaultTheta)
+}
+
+// NewSkewedLatest constructs a new SkewedLatest generator with the given
+// parameters. It returns an error if the parameters are outside the accepted
+// range.
+func NewSkewedLatest(rng *rand.Rand, min, max uint64, theta float64) (*SkewedLatest, error) {
+	z := &SkewedLatest{}
+	z.mu.max = max
+	zipf, err := NewZipf(rng, 0, max-min, theta)
+	if err != nil {
+		return nil, err
+	}
+	z.mu.zipf = zipf
+	return z, nil
+}
+
+// IncMax increments max.
+func (z *SkewedLatest) IncMax(delta int) {
+	z.mu.Lock()
+	z.mu.zipf.IncMax(delta)
+	z.mu.max += uint64(delta)
+	z.mu.Unlock()
+}
+
+// Uint64 returns a random Uint64 between min and max, where keys near max are
+// most likely to be drawn.
+func (z *SkewedLatest) Uint64() uint64 {
+	z.mu.Lock()
+	result := z.mu.max - z.mu.zipf.Uint64()
+	z.mu.Unlock()
+	return result
+}
diff --git a/internal/randvar/skewed_latest_test.go b/internal/randvar/skewed_latest_test.go
new file mode 100644
index 0000000..be12b77
--- /dev/null
+++ b/internal/randvar/skewed_latest_test.go
@@ -0,0 +1,67 @@
+// Copyright 2018 The Cockroach Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied. See the License for the specific language governing
+// permissions and limitations under the License. See the AUTHORS file
+// for names of contributors.
+
+package randvar
+
+import (
+	"fmt"
+	"sort"
+	"testing"
+)
+
+func dumpSamples(x []int) {
+	sort.Ints(x)
+
+	max := x[len(x)-1]
+	step := max / 20
+	if step == 0 {
+		step = 1
+	}
+	index := 0
+	count := 0
+	for i := 0; i <= max; i += step {
+		count = 0
+		for index < len(x) {
+			if x[index] >= i+step {
+				break
+			}
+			index++
+			count++
+		}
+		fmt.Printf("[%3d-%3d) ", i, i+step)
+		for j := 0; j < count; j++ {
+			if j%50 == 0 {
+				fmt.Printf("%c", '∎')
+			}
+		}
+		fmt.Println()
+	}
+}
+
+func TestSkewedLatest(t *testing.T) {
+	z, err := NewSkewedLatest(nil, 0, 99, 0.99)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	x := make([]int, 10000)
+	for i := range x {
+		x[i] = int(z.Uint64())
+	}
+
+	if testing.Verbose() {
+		dumpSamples(x)
+	}
+}
diff --git a/internal/randvar/weighted_test.go b/internal/randvar/weighted_test.go
new file mode 100644
index 0000000..70601bc
--- /dev/null
+++ b/internal/randvar/weighted_test.go
@@ -0,0 +1,20 @@
+// Copyright 2019 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package randvar
+
+import "testing"
+
+func TestWeighted(t *testing.T) {
+	w := NewWeighted(nil, 1, 2, 2, 0, 3)
+
+	x := make([]int, 10000)
+	for i := range x {
+		x[i] = w.Int()
+	}
+
+	if testing.Verbose() {
+		dumpSamples(x)
+	}
+}
diff --git a/internal/randvar/zipf_test.go b/internal/randvar/zipf_test.go
new file mode 100644
index 0000000..df655f2
--- /dev/null
+++ b/internal/randvar/zipf_test.go
@@ -0,0 +1,105 @@
+// Copyright 2017 The Cockroach Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied. See the License for the specific language governing
+// permissions and limitations under the License. See the AUTHORS file
+// for names of contributors.
+
+package randvar
+
+import (
+	"math"
+	"testing"
+)
+
+func TestZeta(t *testing.T) {
+	var zetaTests = []struct {
+		n        uint64
+		theta    float64
+		expected float64
+	}{
+		{20, 0.99, 3.64309060779367},
+		{200, 0.99, 6.02031118558},
+		{1000, 0.99, 7.72895321728},
+		{2000, 0.99, 8.47398788329},
+		{10000, 0.99, 10.2243614596},
+		{100000, 0.99, 12.7783380626},
+		{1000000, 0.99, 15.391849746},
+		{10000000, 0.99, 18.066242575},
+		// TODO(peter): The last test case takes an excessively long time to run (7s).
+		// {100000000, 0.99, 20.80293049},
+	}
+
+	t.Run("FromScratch", func(t *testing.T) {
+		for _, test := range zetaTests {
+			computedZeta := computeZetaFromScratch(test.n, test.theta)
+			if math.Abs(computedZeta-test.expected) > 0.000000001 {
+				t.Fatalf("expected %6.4f, got %6.4f", test.expected, computedZeta)
+			}
+		}
+	})
+
+	t.Run("Incrementally", func(t *testing.T) {
+		// Theta cannot be 1 by definition, so this is a safe initial value.
+		oldTheta := 1.0
+		var oldZetaN float64
+		var oldN uint64
+		for _, test := range zetaTests {
+			// If theta has changed, recompute from scratch
+			if test.theta != oldTheta {
+				oldZetaN = computeZetaFromScratch(test.n, test.theta)
+				oldN = test.n
+				continue
+			}
+
+			computedZeta := computeZetaIncrementally(oldN, test.n, test.theta, oldZetaN)
+			if math.Abs(computedZeta-test.expected) > 0.000000001 {
+				t.Fatalf("expected %6.4f, got %6.4f", test.expected, computedZeta)
+			}
+
+			oldZetaN = computedZeta
+			oldN = test.n
+		}
+	})
+}
+
+func TestNewZipf(t *testing.T) {
+	var gens = []struct {
+		min, max uint64
+		theta    float64
+	}{
+		{0, 100, 0.99},
+		{0, 100, 1.01},
+	}
+
+	for _, gen := range gens {
+		_, err := NewZipf(nil, gen.min, gen.max, gen.theta)
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+func TestZipf(t *testing.T) {
+	z, err := NewZipf(nil, 0, 99, 0.99)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	x := make([]int, 10000)
+	for i := range x {
+		x[i] = int(z.Uint64())
+	}
+
+	if testing.Verbose() {
+		dumpSamples(x)
+	}
+}
diff --git a/testdata/compaction_picker_level_max_bytes b/testdata/compaction_picker_level_max_bytes
index a6b74f4..d610e5c 100644
--- a/testdata/compaction_picker_level_max_bytes
+++ b/testdata/compaction_picker_level_max_bytes
@@ -11,13 +11,13 @@ init 1
 6: 2
 ----
 5: 1
-6: 1
+6: 2
 
 init 1
 6: 2
 ----
 5: 1
-6: 1
+6: 2
 
 init 1
 3: 1
@@ -26,9 +26,9 @@ init 1
 6: 1
 ----
 3: 1
-4: 10
-5: 100
-6: 1000
+4: 1
+5: 1
+6: 1
 
 init 1
 1: 1
@@ -39,11 +39,11 @@ init 1
 6: 1
 ----
 1: 1
-2: 10
-3: 100
-4: 1000
-5: 10000
-6: 100000
+2: 1
+3: 1
+4: 1
+5: 1
+6: 1
 
 init 1
 1: 1
@@ -100,25 +100,45 @@ init 1
 5: 10000
 6: 100000
 
+# Smoothing multiplier is
+# `(size(Lbottom)/size(Lbase))^(1/(Lbottom-Lbase)) = (1000000/1)^(1/(6-1)) = 1000000^(1/5)`
+#
+# size(L1) = size(Lbase) = 1
+# size(L2) = size(L1) * 1000000^(1/5) ~= 16
+# size(L3) = size(L2) * 1000000^(1/5) ~= 251
+# size(L4) = size(L3) * 1000000^(1/5) ~= 3981
+# size(L5) = size(L4) * 1000000^(1/5) ~= 63096
+# size(L6) = size(L5) * 1000000^(1/5) ~= 1000000
+
 init 1
 6: 1000000
 ----
 1: 1
-2: 10
-3: 100
-4: 1000
-5: 10000
-6: 100000
+2: 16
+3: 251
+4: 3981
+5: 63096
+6: 1000000
+
+# Smoothing multiplier is
+# `(size(Lbottom)/size(Lbase))^(Lbottom-Lbase) = (64000000/64)^(1/(6-1)) = 1000000^(1/5)`
+#
+# size(L1) = size(Lbase) = 64
+# size(L2) = size(L1) * 1000000^(1/5) ~= 1014
+# size(L3) = size(L2) * 1000000^(1/5) ~= 16076
+# size(L4) = size(L3) * 1000000^(1/5) ~= 254789
+# size(L5) = size(L4) * 1000000^(1/5) ~= 4038127
+# size(L6) = size(L5) * 1000000^(1/5) ~= 64000000
 
 init 64
 6: 64000000
 ----
 1: 64
-2: 640
-3: 6400
-4: 64000
-5: 640000
-6: 6400000
+2: 1014
+3: 16076
+4: 254789
+5: 4038127
+6: 64000000
 
 init 1
 0: 4
diff --git a/testdata/compaction_picker_target_level b/testdata/compaction_picker_target_level
index ab0df71..3d22bd0 100644
--- a/testdata/compaction_picker_target_level
+++ b/testdata/compaction_picker_target_level
@@ -24,11 +24,18 @@ pick 1
 ----
 5: 2.0
 
+# Smoothing multiplier is
+# `(size(Lbottom)/size(Lbase))^(Lbottom-Lbase) = (30/1)^(1/(6-4)) = 30^(1/2)`
+#
+# size(L4) = size(Lbase) = 1
+# size(L5) = size(L4) * 30^(1/2) ~= 5
+# size(L6) = size(L5) * 30^(1/2) = 30
+
 pick 1
 5: 2
 6: 30
 ----
-5: 2.0
+5: 0.4
 
 pick 1
 4: 2
diff --git a/testdata/event_listener b/testdata/event_listener
index cb768a3..28f3577 100644
--- a/testdata/event_listener
+++ b/testdata/event_listener
@@ -101,5 +101,5 @@ level__files____size___score______in__ingest____move____read___write___w-amp
     3      0     0 B    0.00     0 B     0 B     0 B     0 B     0 B     0.0
     4      0     0 B    0.00     0 B     0 B     0 B     0 B     0 B     0.0
     5      1   959 B    0.00     0 B   959 B     0 B     0 B     0 B     0.0
-    6      1   959 B    0.00   1.9 K     0 B     0 B   1.9 K   959 B     0.5
+    6      1   959 B    1.00   1.9 K     0 B     0 B   1.9 K   959 B     0.5
 total      2   1.9 K    0.00   1.0 K   959 B     0 B   1.9 K   3.8 K     3.8
diff --git a/testdata/flushable_batch b/testdata/flushable_batch
index 315750c..2f84099 100644
--- a/testdata/flushable_batch
+++ b/testdata/flushable_batch
@@ -36,3 +36,107 @@ dump seq=1000
 a#1002,1:1
 b#1001,1:1
 c#1000,1:1
+
+iter
+first
+next
+next
+next
+----
+a:1
+b:1
+c:1
+.
+
+iter
+last
+prev
+prev
+prev
+----
+c:1
+b:1
+a:1
+.
+
+iter lower=b
+seek-ge b
+prev
+----
+b:1
+.
+
+iter lower=c
+seek-lt b
+----
+.
+
+iter lower=e
+last
+----
+.
+
+iter upper=1
+first
+----
+.
+
+iter upper=b
+first
+next
+----
+a:1
+.
+
+iter upper=b
+seek-ge b
+----
+.
+
+iter
+set-bounds lower=b
+seek-ge b
+prev
+----
+b:1
+.
+
+iter
+set-bounds lower=c
+seek-lt b
+----
+.
+
+iter
+set-bounds lower=e
+last
+----
+.
+
+iter
+set-bounds upper=1
+first
+----
+.
+
+iter
+set-bounds upper=b
+first
+next
+----
+a:1
+.
+
+iter
+set-bounds upper=b
+seek-ge b
+----
+.
+
+iter
+first
+set-bounds upper=b
+seek-ge b
+----
+a:1
+.