Repository: cockroachdb/pebble
Pull request #2134 (2022-11-16 16:13:41)

db: read path for values in value blocks
There are a lot of trivial plumbing changes. The substantive changes are in sstable, specifically the new valueBlockReader, and how it interacts with the ReaderProvider in order to read values even when the sstable iterator may be closed. LazyValue.Clone now also takes a *LazyFetcher to avoid allocation.

The changes in blockIter are designed to minimize the performance loss for the v2 sstables, and for v3 sstables that don't have value blocks. Despite that, there is some performance loss in microbenchmarks (see details below). LevelIterSeqSeekPrefixGE/skip=4/use-next=true-16 shows a 10% slowdown from 185ns to 204ns on gceworker (on my macbook there is a smaller slowdown from 69ns to 72ns), for v2 on this PR, compared to master. There is a ~5% slowdown in BlockIterSeekGE/restart=16. The only new code involved here is two simple if-conditions: one for deciding we can construct an InPlaceValue and the second related to the first logic in SeekGE. Profiles confirm this and don't show enough in those places to explain the slow down -- it is possible that we are slightly worse with cache or register allocation behavior because of the larger structs, or maybe the if-blocks are causing hiccups in the instruction pipeline.

master vs PR (neither using v3):

```
name                                                                                        old time/op    new time/op    delta
IteratorSeqSeekPrefixGENotFound/skip=1/two-level=false/bloom=false/with-tombstone=false-16     508ns ± 1%     509ns ± 0%     ~     (p=0.135 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=1/two-level=false/bloom=false/with-tombstone=true-16      421ns ± 0%     427ns ± 0%   +1.32%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=1/two-level=false/bloom=true/with-tombstone=false-16     1.11µs ± 1%    1.11µs ± 0%     ~     (p=0.968 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=1/two-level=false/bloom=true/with-tombstone=true-16       602ns ± 1%     603ns ± 1%     ~     (p=0.841 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=1/two-level=true/bloom=false/with-tombstone=false-16      594ns ± 0%     602ns ± 1%   +1.26%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=1/two-level=true/bloom=false/with-tombstone=true-16       444ns ± 0%     447ns ± 0%   +0.61%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=1/two-level=true/bloom=true/with-tombstone=false-16      1.14µs ± 0%    1.15µs ± 0%   +0.81%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=1/two-level=true/bloom=true/with-tombstone=true-16        609ns ± 2%     605ns ± 0%     ~     (p=0.690 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=2/two-level=false/bloom=false/with-tombstone=false-16     513ns ± 0%     512ns ± 1%     ~     (p=0.500 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=2/two-level=false/bloom=false/with-tombstone=true-16      428ns ± 0%     428ns ± 1%     ~     (p=0.810 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=2/two-level=false/bloom=true/with-tombstone=false-16     1.12µs ± 1%    1.11µs ± 0%   -0.61%  (p=0.048 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=2/two-level=false/bloom=true/with-tombstone=true-16       617ns ± 2%     611ns ± 1%     ~     (p=0.548 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=2/two-level=true/bloom=false/with-tombstone=false-16      599ns ± 0%     603ns ± 1%   +0.78%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=2/two-level=true/bloom=false/with-tombstone=true-16       450ns ± 0%     449ns ± 0%     ~     (p=0.151 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=2/two-level=true/bloom=true/with-tombstone=false-16      1.15µs ± 1%    1.15µs ± 1%     ~     (p=0.651 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=2/two-level=true/bloom=true/with-tombstone=true-16        621ns ± 1%     616ns ± 1%     ~     (p=0.310 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=4/two-level=false/bloom=false/with-tombstone=false-16     518ns ± 0%     517ns ± 1%     ~     (p=0.278 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=4/two-level=false/bloom=false/with-tombstone=true-16      434ns ± 1%     433ns ± 1%     ~     (p=0.548 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=4/two-level=false/bloom=true/with-tombstone=false-16     1.12µs ± 1%    1.13µs ± 1%     ~     (p=0.206 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=4/two-level=false/bloom=true/with-tombstone=true-16       621ns ± 1%     615ns ± 2%     ~     (p=0.310 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=4/two-level=true/bloom=false/with-tombstone=false-16      610ns ± 1%     612ns ± 1%     ~     (p=0.246 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=4/two-level=true/bloom=false/with-tombstone=true-16       459ns ± 1%     452ns ± 1%   -1.43%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=4/two-level=true/bloom=true/with-tombstone=false-16      1.15µs ± 1%    1.16µs ± 1%     ~     (p=0.095 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=4/two-level=true/bloom=true/with-tombstone=true-16        610ns ± 2%     615ns ± 1%     ~     (p=0.246 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=1/two-level=false/bloom=false/with-tombstone=false-16        635ns ± 0%     625ns ± 1%   -1.54%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=1/two-level=false/bloom=false/with-tombstone=true-16         523ns ± 0%     520ns ± 0%   -0.49%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=1/two-level=false/bloom=true/with-tombstone=false-16        1.36µs ± 0%    1.37µs ± 1%   +1.42%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=1/two-level=false/bloom=true/with-tombstone=true-16          855ns ± 0%     859ns ± 1%     ~     (p=0.151 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=1/two-level=true/bloom=false/with-tombstone=false-16         725ns ± 0%     717ns ± 1%   -1.02%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=1/two-level=true/bloom=false/with-tombstone=true-16          549ns ± 0%     544ns ± 0%   -0.85%  (p=0.016 n=5+4)
IteratorSeqSeekPrefixGEFound/skip=1/two-level=true/bloom=true/with-tombstone=false-16         1.40µs ± 1%    1.39µs ± 1%     ~     (p=0.690 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=1/two-level=true/bloom=true/with-tombstone=true-16           877ns ± 1%     886ns ± 0%   +1.00%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=2/two-level=false/bloom=false/with-tombstone=false-16        672ns ± 1%     663ns ± 0%   -1.39%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=2/two-level=false/bloom=false/with-tombstone=true-16         562ns ± 0%     563ns ± 0%     ~     (p=0.810 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=2/two-level=false/bloom=true/with-tombstone=false-16        1.40µs ± 0%    1.42µs ± 0%   +1.03%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=2/two-level=false/bloom=true/with-tombstone=true-16          902ns ± 1%     910ns ± 1%   +0.85%  (p=0.032 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=2/two-level=true/bloom=false/with-tombstone=false-16         765ns ± 1%     761ns ± 1%     ~     (p=0.151 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=2/two-level=true/bloom=false/with-tombstone=true-16          587ns ± 0%     584ns ± 1%   -0.48%  (p=0.024 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=2/two-level=true/bloom=true/with-tombstone=false-16         1.44µs ± 0%    1.45µs ± 1%   +0.85%  (p=0.016 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=2/two-level=true/bloom=true/with-tombstone=true-16           922ns ± 1%     936ns ± 1%   +1.56%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=4/two-level=false/bloom=false/with-tombstone=false-16        746ns ± 1%     752ns ± 1%   +0.89%  (p=0.048 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=4/two-level=false/bloom=false/with-tombstone=true-16         632ns ± 1%     643ns ± 0%   +1.65%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=4/two-level=false/bloom=true/with-tombstone=false-16        1.49µs ± 1%    1.51µs ± 1%   +1.32%  (p=0.016 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=4/two-level=false/bloom=true/with-tombstone=true-16          983ns ± 0%     994ns ± 1%   +1.12%  (p=0.016 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=4/two-level=true/bloom=false/with-tombstone=false-16         840ns ± 0%     848ns ± 0%   +1.04%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=4/two-level=true/bloom=false/with-tombstone=true-16          658ns ± 0%     663ns ± 0%   +0.80%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=4/two-level=true/bloom=true/with-tombstone=false-16         1.54µs ± 0%    1.55µs ± 1%   +0.83%  (p=0.016 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=4/two-level=true/bloom=true/with-tombstone=true-16          1.01µs ± 0%    1.03µs ± 1%   +2.09%  (p=0.008 n=5+5)
IteratorSeqSeekGEWithBounds/two-level=false-16                                                 783ns ± 2%     776ns ± 0%     ~     (p=0.190 n=5+4)
IteratorSeqSeekGEWithBounds/two-level=true-16                                                  773ns ± 0%     776ns ± 1%     ~     (p=0.071 n=5+5)
IteratorSeekGENoop/withLimit=false-16                                                         58.2ns ± 2%    57.3ns ± 1%   -1.53%  (p=0.016 n=5+5)
IteratorSeekGENoop/withLimit=true-16                                                          98.6ns ± 0%    98.9ns ± 0%   +0.31%  (p=0.008 n=5+5)
Iterator_RangeKeyMasking/range-keys-suffixes=@10/forward-16                                   5.35ms ± 1%    5.33ms ± 3%     ~     (p=0.548 n=5+5)
Iterator_RangeKeyMasking/range-keys-suffixes=@10/backward-16                                  6.82ms ± 1%    6.91ms ± 4%     ~     (p=0.151 n=5+5)
Iterator_RangeKeyMasking/range-keys-suffixes=@50/forward-16                                   4.85ms ± 4%    4.84ms ± 2%     ~     (p=0.548 n=5+5)
Iterator_RangeKeyMasking/range-keys-suffixes=@50/backward-16                                  5.97ms ± 2%    6.02ms ± 2%     ~     (p=0.548 n=5+5)
Iterator_RangeKeyMasking/range-keys-suffixes=@75/forward-16                                   2.66ms ± 0%    2.65ms ± 1%     ~     (p=0.190 n=4+5)
Iterator_RangeKeyMasking/range-keys-suffixes=@75/backward-16                                  3.53ms ± 0%    3.50ms ± 2%     ~     (p=0.730 n=4+5)
Iterator_RangeKeyMasking/range-keys-suffixes=@100/forward-16                                   361µs ± 1%     366µs ± 1%   +1.24%  (p=0.008 n=5+5)
Iterator_RangeKeyMasking/range-keys-suffixes=@100/backward-16                                  704µs ± 1%     727µs ± 0%   +3.36%  (p=0.008 n=5+5)
IteratorScan/keys=100,r-amp=1,key-types=points-only-16                                        11.5µs ±15%    11.1µs ± 3%     ~     (p=1.000 n=5+5)
IteratorScan/keys=100,r-amp=1,key-types=points-and-ranges-16                                  11.7µs ± 5%    12.2µs ± 7%     ~     (p=0.421 n=5+5)
IteratorScan/keys=100,r-amp=3,key-types=points-only-16                                        19.4µs ± 5%    19.6µs ± 5%     ~     (p=0.421 n=5+5)
IteratorScan/keys=100,r-amp=3,key-types=points-and-ranges-16                                  21.0µs ± 5%    21.8µs ± 8%     ~     (p=0.310 n=5+5)
IteratorScan/keys=100,r-amp=7,key-types=points-only-16                                        29.7µs ± 5%    28.4µs ± 5%     ~     (p=0.095 n=5+5)
IteratorScan/keys=100,r-amp=7,key-types=points-and-ranges-16                                  30.6µs ±11%    31.3µs ±10%     ~     (p=0.548 n=5+5)
IteratorScan/keys=100,r-amp=10,key-types=points-only-16                                       35.8µs ± 8%    35.9µs ± 8%     ~     (p=0.690 n=5+5)
IteratorScan/keys=100,r-amp=10,key-types=points-and-ranges-16                                 34.5µs ± 5%    34.6µs ± 8%     ~     (p=1.000 n=5+5)
IteratorScan/keys=1000,r-amp=1,key-types=points-only-16                                       85.9µs ± 0%    87.2µs ± 1%   +1.50%  (p=0.029 n=4+4)
IteratorScan/keys=1000,r-amp=1,key-types=points-and-ranges-16                                  100µs ±18%      95µs ±13%     ~     (p=1.000 n=5+5)
IteratorScan/keys=1000,r-amp=3,key-types=points-only-16                                        151µs ± 1%     151µs ± 1%     ~     (p=0.556 n=4+5)
IteratorScan/keys=1000,r-amp=3,key-types=points-and-ranges-16                                  155µs ± 1%     155µs ± 1%     ~     (p=0.905 n=5+4)
IteratorScan/keys=1000,r-amp=7,key-types=points-only-16                                        197µs ± 3%     196µs ± 1%     ~     (p=0.548 n=5+5)
IteratorScan/keys=1000,r-amp=7,key-types=points-and-ranges-16                                  199µs ± 0%     201µs ± 1%   +0.87%  (p=0.032 n=5+4)
IteratorScan/keys=1000,r-amp=10,key-types=points-only-16                                       216µs ± 3%     217µs ± 3%     ~     (p=0.548 n=5+5)
IteratorScan/keys=1000,r-amp=10,key-types=points-and-ranges-16                                 223µs ± 3%     222µs ± 2%     ~     (p=0.905 n=5+4)
IteratorScan/keys=10000,r-amp=1,key-types=points-only-16                                       855µs ± 3%     869µs ± 2%     ~     (p=0.095 n=5+5)
IteratorScan/keys=10000,r-amp=1,key-types=points-and-ranges-16                                 881µs ± 0%     917µs ± 3%   +4.09%  (p=0.008 n=5+5)
IteratorScan/keys=10000,r-amp=3,key-types=points-only-16                                      1.47ms ± 2%    1.46ms ± 2%     ~     (p=0.095 n=5+5)
IteratorScan/keys=10000,r-amp=3,key-types=points-and-ranges-16                                1.53ms ± 3%    1.50ms ± 1%   -2.32%  (p=0.008 n=5+5)
IteratorScan/keys=10000,r-amp=7,key-types=points-only-16                                      1.88ms ± 1%    1.87ms ± 1%     ~     (p=0.421 n=5+5)
IteratorScan/keys=10000,r-amp=7,key-types=points-and-ranges-16                                1.90ms ± 0%    1.91ms ± 3%     ~     (p=0.413 n=4+5)
IteratorScan/keys=10000,r-amp=10,key-types=points-only-16                                     2.01ms ± 0%    2.03ms ± 1%   +0.91%  (p=0.008 n=5+5)
IteratorScan/keys=10000,r-amp=10,key-types=points-and-ranges-16                               2.07ms ± 1%    2.07ms ± 1%     ~     (p=1.000 n=5+5)
CombinedIteratorSeek/range-key=false/batch=false-16                                           5.74µs ± 1%    5.74µs ± 1%     ~     (p=1.000 n=5+5)
CombinedIteratorSeek/range-key=false/batch=true-16                                            6.48µs ± 1%    6.56µs ± 1%   +1.28%  (p=0.008 n=5+5)
CombinedIteratorSeek/range-key=true/batch=false-16                                            13.7µs ± 1%    13.8µs ± 1%     ~     (p=0.151 n=5+5)
CombinedIteratorSeek/range-key=true/batch=true-16                                             14.5µs ± 0%    14.7µs ± 1%   +1.75%  (p=0.008 n=5+5)
CombinedIteratorSeek_Bounded-16                                                               87.6µs ± 1%    88.5µs ± 1%   +1.03%  (p=0.016 n=5+5)
CombinedIteratorSeekPrefix-16                                                                  179µs ± 0%     183µs ± 1%   +2.22%  (p=0.008 n=5+5)
LevelIterSeekGE/restart=16/count=5-16                                                         1.77µs ± 1%    1.89µs ± 5%   +6.87%  (p=0.008 n=5+5)
LevelIterSeqSeekGEWithBounds/restart=16/count=5-16                                             123ns ± 1%     123ns ± 0%     ~     (p=0.278 n=5+5)
LevelIterSeqSeekPrefixGE/skip=1/use-next=false-16                                              690ns ± 2%     691ns ± 0%     ~     (p=0.690 n=5+5)
LevelIterSeqSeekPrefixGE/skip=1/use-next=true-16                                              78.8ns ± 0%    82.4ns ± 0%   +4.59%  (p=0.008 n=5+5)
LevelIterSeqSeekPrefixGE/skip=2/use-next=false-16                                              700ns ± 1%     708ns ± 1%     ~     (p=0.175 n=5+5)
LevelIterSeqSeekPrefixGE/skip=2/use-next=true-16                                               113ns ± 1%     118ns ± 1%   +4.44%  (p=0.008 n=5+5)
LevelIterSeqSeekPrefixGE/skip=4/use-next=false-16                                              730ns ± 1%     730ns ± 1%     ~     (p=0.841 n=5+5)
LevelIterSeqSeekPrefixGE/skip=4/use-next=true-16                                               185ns ± 0%     204ns ± 1%  +10.13%  (p=0.008 n=5+5)
LevelIterSeqSeekPrefixGE/skip=8/use-next=false-16                                              856ns ± 1%     861ns ± 0%     ~     (p=0.421 n=5+5)
LevelIterSeqSeekPrefixGE/skip=8/use-next=true-16                                               954ns ± 1%     976ns ± 1%   +2.31%  (p=0.008 n=5+5)
LevelIterSeqSeekPrefixGE/skip=16/use-next=false-16                                             902ns ± 2%     900ns ± 4%     ~     (p=0.841 n=5+5)
LevelIterSeqSeekPrefixGE/skip=16/use-next=true-16                                              967ns ± 1%     971ns ± 3%     ~     (p=0.690 n=5+5)
LevelIterNext/restart=16/count=5-16                                                           24.4ns ± 1%    25.4ns ± 1%   +3.86%  (p=0.008 n=5+5)
LevelIterPrev/restart=16/count=5-16                                                           46.5ns ± 0%    46.7ns ± 0%     ~     (p=0.151 n=5+5)
MergingIterSeekGE/restart=16/count=1-16                                                       1.18µs ± 1%    1.20µs ± 2%     ~     (p=0.087 n=5+5)
MergingIterSeekGE/restart=16/count=2-16                                                       2.30µs ± 1%    2.34µs ± 1%   +1.75%  (p=0.016 n=5+5)
MergingIterSeekGE/restart=16/count=3-16                                                       3.53µs ± 2%    3.58µs ± 2%     ~     (p=0.151 n=5+5)
MergingIterSeekGE/restart=16/count=4-16                                                       4.81µs ± 1%    4.97µs ± 2%   +3.30%  (p=0.008 n=5+5)
MergingIterSeekGE/restart=16/count=5-16                                                       6.23µs ± 1%    6.37µs ± 2%   +2.18%  (p=0.032 n=5+5)
MergingIterNext/restart=16/count=1-16                                                         46.0ns ± 1%    47.6ns ± 1%   +3.40%  (p=0.008 n=5+5)
MergingIterNext/restart=16/count=2-16                                                         74.4ns ± 1%    77.2ns ± 3%   +3.71%  (p=0.008 n=5+5)
MergingIterNext/restart=16/count=3-16                                                         92.2ns ± 0%    95.3ns ± 0%   +3.38%  (p=0.008 n=5+5)
MergingIterNext/restart=16/count=4-16                                                          106ns ± 0%     109ns ± 0%   +2.82%  (p=0.008 n=5+5)
MergingIterNext/restart=16/count=5-16                                                          124ns ± 1%     129ns ± 2%   +3.36%  (p=0.008 n=5+5)
MergingIterPrev/restart=16/count=1-16                                                         66.8ns ± 1%    69.2ns ± 2%   +3.57%  (p=0.008 n=5+5)
MergingIterPrev/restart=16/count=2-16                                                         96.1ns ± 0%    99.3ns ± 1%   +3.31%  (p=0.008 n=5+5)
MergingIterPrev/restart=16/count=3-16                                                          115ns ± 0%     117ns ± 0%   +2.05%  (p=0.008 n=5+5)
MergingIterPrev/restart=16/count=4-16                                                          128ns ± 1%     129ns ± 1%   +1.13%  (p=0.048 n=5+5)
MergingIterPrev/restart=16/count=5-16                                                          144ns ± 1%     145ns ± 1%     ~     (p=0.087 n=5+5)
MergingIterSeqSeekGEWithBounds/levelCount=5-16                                                 613ns ± 1%     605ns ± 0%   -1.35%  (p=0.008 n=5+5)
MergingIterSeqSeekPrefixGE/skip=1/use-next=false-16                                           1.92µs ± 2%    1.93µs ± 0%     ~     (p=0.151 n=5+5)
MergingIterSeqSeekPrefixGE/skip=1/use-next=true-16                                             501ns ± 0%     499ns ± 1%     ~     (p=0.333 n=5+5)
MergingIterSeqSeekPrefixGE/skip=2/use-next=false-16                                           1.92µs ± 1%    1.95µs ± 0%   +1.64%  (p=0.008 n=5+5)
MergingIterSeqSeekPrefixGE/skip=2/use-next=true-16                                             536ns ± 1%     537ns ± 1%     ~     (p=0.548 n=5+5)
MergingIterSeqSeekPrefixGE/skip=4/use-next=false-16                                           1.97µs ± 0%    2.00µs ± 1%   +1.73%  (p=0.008 n=5+5)
MergingIterSeqSeekPrefixGE/skip=4/use-next=true-16                                             604ns ± 1%     619ns ± 1%   +2.45%  (p=0.008 n=5+5)
MergingIterSeqSeekPrefixGE/skip=8/use-next=false-16                                           2.12µs ± 1%    2.14µs ± 0%   +0.72%  (p=0.016 n=5+5)
MergingIterSeqSeekPrefixGE/skip=8/use-next=true-16                                            1.43µs ± 1%    1.45µs ± 2%   +1.38%  (p=0.008 n=5+5)
MergingIterSeqSeekPrefixGE/skip=16/use-next=false-16                                          2.27µs ± 1%    2.31µs ± 1%   +1.93%  (p=0.008 n=5+5)
MergingIterSeqSeekPrefixGE/skip=16/use-next=true-16                                           1.56µs ± 2%    1.58µs ± 1%     ~     (p=0.111 n=5+5)

name                                              old time/op    new time/op    delta
BlockIterSeekGE/restart=16-16                        460ns ± 0%     484ns ± 1%  +5.25%  (p=0.008 n=5+5)
BlockIterSeekLT/restart=16-16                        535ns ± 2%     532ns ± 1%    ~     (p=0.548 n=5+5)
BlockIterNext/restart=16-16                         18.4ns ± 0%    18.8ns ± 0%  +2.20%  (p=0.008 n=5+5)
BlockIterPrev/restart=16-16                         37.9ns ± 0%    39.2ns ± 0%  +3.60%  (p=0.008 n=5+5)
TableIterSeekGE/restart=16,compression=Snappy-16    1.58µs ± 3%    1.58µs ± 2%    ~     (p=1.000 n=5+5)
TableIterSeekGE/restart=16,compression=ZSTD-16      1.57µs ± 2%    1.57µs ± 1%    ~     (p=0.841 n=5+5)
TableIterSeekLT/restart=16,compression=Snappy-16    1.62µs ± 1%    1.63µs ± 1%    ~     (p=0.421 n=5+5)
TableIterSeekLT/restart=16,compression=ZSTD-16      1.62µs ± 1%    1.63µs ± 2%    ~     (p=0.310 n=5+5)
TableIterNext/restart=16,compression=Snappy-16      22.4ns ± 1%    23.7ns ± 1%  +6.19%  (p=0.008 n=5+5)
TableIterNext/restart=16,compression=ZSTD-16        22.5ns ± 1%    23.8ns ± 1%  +6.03%  (p=0.008 n=5+5)
TableIterPrev/restart=16,compression=Snappy-16      43.0ns ± 1%    44.9ns ± 1%  +4.53%  (p=0.008 n=5+5)
TableIterPrev/restart=16,compression=ZSTD-16        43.1ns ± 1%    45.2ns ± 1%  +4.85%  (p=0.008 n=5+5)
```

PR v2 or earlier vs PR v3:
```
IteratorSeqSeekPrefixGENotFound/skip=1/two-level=false/bloom=false/with-tombstone=false-16     509ns ± 0%     513ns ± 1%    ~     (p=0.087 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=1/two-level=false/bloom=false/with-tombstone=true-16      427ns ± 0%     426ns ± 0%    ~     (p=0.127 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=1/two-level=false/bloom=true/with-tombstone=false-16     1.11µs ± 0%    1.11µs ± 1%    ~     (p=1.000 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=1/two-level=false/bloom=true/with-tombstone=true-16       603ns ± 1%     608ns ± 1%    ~     (p=0.111 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=1/two-level=true/bloom=false/with-tombstone=false-16      602ns ± 1%     600ns ± 0%    ~     (p=0.452 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=1/two-level=true/bloom=false/with-tombstone=true-16       447ns ± 0%     446ns ± 0%    ~     (p=0.246 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=1/two-level=true/bloom=true/with-tombstone=false-16      1.15µs ± 0%    1.15µs ± 0%  +0.59%  (p=0.016 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=1/two-level=true/bloom=true/with-tombstone=true-16        605ns ± 0%     611ns ± 1%    ~     (p=0.222 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=2/two-level=false/bloom=false/with-tombstone=false-16     512ns ± 1%     512ns ± 0%    ~     (p=0.889 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=2/two-level=false/bloom=false/with-tombstone=true-16      428ns ± 1%     429ns ± 1%    ~     (p=0.548 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=2/two-level=false/bloom=true/with-tombstone=false-16     1.11µs ± 0%    1.11µs ± 1%    ~     (p=0.143 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=2/two-level=false/bloom=true/with-tombstone=true-16       611ns ± 1%     619ns ± 1%  +1.18%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=2/two-level=true/bloom=false/with-tombstone=false-16      603ns ± 1%     605ns ± 1%    ~     (p=0.095 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=2/two-level=true/bloom=false/with-tombstone=true-16       449ns ± 0%     449ns ± 0%    ~     (p=1.000 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=2/two-level=true/bloom=true/with-tombstone=false-16      1.15µs ± 1%    1.18µs ± 1%  +2.69%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=2/two-level=true/bloom=true/with-tombstone=true-16        616ns ± 1%     624ns ± 1%  +1.30%  (p=0.048 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=4/two-level=false/bloom=false/with-tombstone=false-16     517ns ± 1%     521ns ± 1%  +0.89%  (p=0.032 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=4/two-level=false/bloom=false/with-tombstone=true-16      433ns ± 1%     435ns ± 0%    ~     (p=0.095 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=4/two-level=false/bloom=true/with-tombstone=false-16     1.13µs ± 1%    1.12µs ± 1%    ~     (p=0.183 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=4/two-level=false/bloom=true/with-tombstone=true-16       615ns ± 2%     619ns ± 2%    ~     (p=0.310 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=4/two-level=true/bloom=false/with-tombstone=false-16      612ns ± 1%     613ns ± 1%    ~     (p=0.421 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=4/two-level=true/bloom=false/with-tombstone=true-16       452ns ± 1%     456ns ± 1%  +0.91%  (p=0.032 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=4/two-level=true/bloom=true/with-tombstone=false-16      1.16µs ± 1%    1.18µs ± 1%  +1.67%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGENotFound/skip=4/two-level=true/bloom=true/with-tombstone=true-16        615ns ± 1%     615ns ± 2%    ~     (p=0.786 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=1/two-level=false/bloom=false/with-tombstone=false-16        625ns ± 1%     626ns ± 0%    ~     (p=0.421 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=1/two-level=false/bloom=false/with-tombstone=true-16         520ns ± 0%     523ns ± 0%  +0.62%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=1/two-level=false/bloom=true/with-tombstone=false-16        1.37µs ± 1%    1.36µs ± 0%  -0.89%  (p=0.016 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=1/two-level=false/bloom=true/with-tombstone=true-16          859ns ± 1%     856ns ± 0%    ~     (p=0.548 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=1/two-level=true/bloom=false/with-tombstone=false-16         717ns ± 1%     721ns ± 0%  +0.49%  (p=0.048 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=1/two-level=true/bloom=false/with-tombstone=true-16          544ns ± 0%     545ns ± 0%    ~     (p=0.730 n=4+5)
IteratorSeqSeekPrefixGEFound/skip=1/two-level=true/bloom=true/with-tombstone=false-16         1.39µs ± 1%    1.41µs ± 1%  +1.31%  (p=0.016 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=1/two-level=true/bloom=true/with-tombstone=true-16           886ns ± 0%     885ns ± 2%    ~     (p=0.310 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=2/two-level=false/bloom=false/with-tombstone=false-16        663ns ± 0%     667ns ± 1%  +0.64%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=2/two-level=false/bloom=false/with-tombstone=true-16         563ns ± 0%     566ns ± 1%  +0.62%  (p=0.016 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=2/two-level=false/bloom=true/with-tombstone=false-16        1.42µs ± 0%    1.40µs ± 1%  -1.13%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=2/two-level=false/bloom=true/with-tombstone=true-16          910ns ± 1%     913ns ± 1%    ~     (p=0.397 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=2/two-level=true/bloom=false/with-tombstone=false-16         761ns ± 1%     767ns ± 0%  +0.85%  (p=0.032 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=2/two-level=true/bloom=false/with-tombstone=true-16          584ns ± 1%     589ns ± 1%  +0.82%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=2/two-level=true/bloom=true/with-tombstone=false-16         1.45µs ± 1%    1.46µs ± 0%    ~     (p=0.794 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=2/two-level=true/bloom=true/with-tombstone=true-16           936ns ± 1%     932ns ± 1%    ~     (p=0.238 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=4/two-level=false/bloom=false/with-tombstone=false-16        752ns ± 1%     760ns ± 1%  +1.01%  (p=0.048 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=4/two-level=false/bloom=false/with-tombstone=true-16         643ns ± 0%     653ns ± 1%  +1.62%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=4/two-level=false/bloom=true/with-tombstone=false-16        1.51µs ± 1%    1.51µs ± 1%    ~     (p=1.000 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=4/two-level=false/bloom=true/with-tombstone=true-16          994ns ± 1%    1003ns ± 1%    ~     (p=0.214 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=4/two-level=true/bloom=false/with-tombstone=false-16         848ns ± 0%     856ns ± 1%  +0.85%  (p=0.016 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=4/two-level=true/bloom=false/with-tombstone=true-16          663ns ± 0%     673ns ± 0%  +1.48%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=4/two-level=true/bloom=true/with-tombstone=false-16         1.55µs ± 1%    1.57µs ± 1%  +1.51%  (p=0.008 n=5+5)
IteratorSeqSeekPrefixGEFound/skip=4/two-level=true/bloom=true/with-tombstone=true-16          1.03µs ± 1%    1.02µs ± 1%    ~     (p=0.500 n=5+5)
IteratorSeqSeekGEWithBounds/two-level=false-16                                                 776ns ± 0%     779ns ± 1%    ~     (p=0.413 n=4+5)
IteratorSeqSeekGEWithBounds/two-level=true-16                                                  776ns ± 1%     776ns ± 0%    ~     (p=0.841 n=5+5)
IteratorSeekGENoop/withLimit=false-16                                                         57.3ns ± 1%    57.2ns ± 0%    ~     (p=0.548 n=5+5)
IteratorSeekGENoop/withLimit=true-16                                                          98.9ns ± 0%    98.8ns ± 0%    ~     (p=0.278 n=5+5)
Iterator_RangeKeyMasking/range-keys-suffixes=@10/forward-16                                   5.33ms ± 3%    5.27ms ± 2%    ~     (p=0.548 n=5+5)
Iterator_RangeKeyMasking/range-keys-suffixes=@10/backward-16                                  6.91ms ± 4%    6.76ms ± 2%    ~     (p=0.095 n=5+5)
Iterator_RangeKeyMasking/range-keys-suffixes=@50/forward-16                                   4.84ms ± 2%    4.90ms ± 3%    ~     (p=0.548 n=5+5)
Iterator_RangeKeyMasking/range-keys-suffixes=@50/backward-16                                  6.02ms ± 2%    6.07ms ± 1%    ~     (p=0.310 n=5+5)
Iterator_RangeKeyMasking/range-keys-suffixes=@75/forward-16                                   2.65ms ± 1%    2.69ms ± 1%  +1.65%  (p=0.032 n=5+5)
Iterator_RangeKeyMasking/range-keys-suffixes=@75/backward-16                                  3.50ms ± 2%    3.50ms ± 1%    ~     (p=1.000 n=5+5)
Iterator_RangeKeyMasking/range-keys-suffixes=@100/forward-16                                   366µs ± 1%     364µs ± 1%    ~     (p=0.151 n=5+5)
Iterator_RangeKeyMasking/range-keys-suffixes=@100/backward-16                                  727µs ± 0%     726µs ± 1%    ~     (p=0.548 n=5+5)
IteratorScan/keys=100,r-amp=1,key-types=points-only-16                                        11.1µs ± 3%    11.4µs ±21%    ~     (p=0.421 n=5+5)
IteratorScan/keys=100,r-amp=1,key-types=points-and-ranges-16                                  12.2µs ± 7%    11.4µs ± 8%    ~     (p=0.095 n=5+5)
IteratorScan/keys=100,r-amp=3,key-types=points-only-16                                        19.6µs ± 5%    21.1µs ± 9%    ~     (p=0.056 n=5+5)
IteratorScan/keys=100,r-amp=3,key-types=points-and-ranges-16                                  21.8µs ± 8%    21.5µs ± 5%    ~     (p=0.841 n=5+5)
IteratorScan/keys=100,r-amp=7,key-types=points-only-16                                        28.4µs ± 5%    29.2µs ± 4%    ~     (p=0.222 n=5+5)
IteratorScan/keys=100,r-amp=7,key-types=points-and-ranges-16                                  31.3µs ±10%    28.7µs ± 2%  -8.49%  (p=0.016 n=5+5)
IteratorScan/keys=100,r-amp=10,key-types=points-only-16                                       35.9µs ± 8%    36.2µs ±12%    ~     (p=1.000 n=5+5)
IteratorScan/keys=100,r-amp=10,key-types=points-and-ranges-16                                 34.6µs ± 8%    34.5µs ± 7%    ~     (p=0.690 n=5+5)
IteratorScan/keys=1000,r-amp=1,key-types=points-only-16                                       87.2µs ± 1%    96.9µs ±20%    ~     (p=0.905 n=4+5)
IteratorScan/keys=1000,r-amp=1,key-types=points-and-ranges-16                                 95.5µs ±13%    90.1µs ± 0%  -5.65%  (p=0.032 n=5+4)
IteratorScan/keys=1000,r-amp=3,key-types=points-only-16                                        151µs ± 1%     149µs ± 0%    ~     (p=0.111 n=5+4)
IteratorScan/keys=1000,r-amp=3,key-types=points-and-ranges-16                                  155µs ± 1%     164µs ± 6%    ~     (p=0.286 n=4+5)
IteratorScan/keys=1000,r-amp=7,key-types=points-only-16                                        196µs ± 1%     195µs ± 1%    ~     (p=0.190 n=5+4)
IteratorScan/keys=1000,r-amp=7,key-types=points-and-ranges-16                                  201µs ± 1%     199µs ± 1%    ~     (p=0.200 n=4+4)
IteratorScan/keys=1000,r-amp=10,key-types=points-only-16                                       217µs ± 3%     215µs ± 3%    ~     (p=0.310 n=5+5)
IteratorScan/keys=1000,r-amp=10,key-types=points-and-ranges-16                                 222µs ± 2%     220µs ± 2%    ~     (p=0.905 n=4+5)
IteratorScan/keys=10000,r-amp=1,key-types=points-only-16                                       869µs ± 2%     866µs ± 2%    ~     (p=0.841 n=5+5)
IteratorScan/keys=10000,r-amp=1,key-types=points-and-ranges-16                                 917µs ± 3%     906µs ± 1%    ~     (p=0.310 n=5+5)
IteratorScan/keys=10000,r-amp=3,key-types=points-only-16                                      1.46ms ± 2%    1.45ms ± 0%  -1.03%  (p=0.008 n=5+5)
IteratorScan/keys=10000,r-amp=3,key-types=points-and-ranges-16                                1.50ms ± 1%    1.49ms ± 0%    ~     (p=0.905 n=5+4)
IteratorScan/keys=10000,r-amp=7,key-types=points-only-16                                      1.87ms ± 1%    1.84ms ± 1%  -1.14%  (p=0.032 n=5+5)
IteratorScan/keys=10000,r-amp=7,key-types=points-and-ranges-16                                1.91ms ± 3%    1.89ms ± 0%  -1.17%  (p=0.032 n=5+5)
IteratorScan/keys=10000,r-amp=10,key-types=points-only-16                                     2.03ms ± 1%    2.00ms ± 0%  -1.64%  (p=0.008 n=5+5)
IteratorScan/keys=10000,r-amp=10,key-types=points-and-ranges-16                               2.07ms ± 1%    2.05ms ± 2%    ~     (p=0.151 n=5+5)
CombinedIteratorSeek/range-key=false/batch=false-16                                           5.74µs ± 1%    5.78µs ± 1%  +0.73%  (p=0.016 n=5+5)
CombinedIteratorSeek/range-key=false/batch=true-16                                            6.56µs ± 1%    6.59µs ± 0%    ~     (p=0.151 n=5+5)
CombinedIteratorSeek/range-key=true/batch=false-16                                            13.8µs ± 1%    13.7µs ± 1%    ~     (p=0.056 n=5+5)
CombinedIteratorSeek/range-key=true/batch=true-16                                             14.7µs ± 1%    14.6µs ± 0%  -1.22%  (p=0.008 n=5+5)
CombinedIteratorSeek_Bounded-16                                                               88.5µs ± 1%    87.9µs ± 1%    ~     (p=0.056 n=5+5)
CombinedIteratorSeekPrefix-16                                                                  183µs ± 1%     183µs ± 0%    ~     (p=1.000 n=5+5)
LevelIterSeekGE/restart=16/count=5-16                                                         1.89µs ± 5%    1.82µs ± 1%    ~     (p=0.095 n=5+5)
LevelIterSeqSeekGEWithBounds/restart=16/count=5-16                                             123ns ± 0%     123ns ± 1%    ~     (p=0.730 n=5+5)
LevelIterSeqSeekPrefixGE/skip=1/use-next=false-16                                              691ns ± 0%     703ns ± 1%  +1.67%  (p=0.008 n=5+5)
LevelIterSeqSeekPrefixGE/skip=1/use-next=true-16                                              82.4ns ± 0%    81.5ns ± 0%  -1.15%  (p=0.008 n=5+5)
LevelIterSeqSeekPrefixGE/skip=2/use-next=false-16                                              708ns ± 1%     715ns ± 1%  +1.03%  (p=0.008 n=5+5)
LevelIterSeqSeekPrefixGE/skip=2/use-next=true-16                                               118ns ± 1%     117ns ± 1%    ~     (p=0.675 n=5+5)
LevelIterSeqSeekPrefixGE/skip=4/use-next=false-16                                              730ns ± 1%     747ns ± 1%  +2.46%  (p=0.008 n=5+5)
LevelIterSeqSeekPrefixGE/skip=4/use-next=true-16                                               204ns ± 1%     204ns ± 1%    ~     (p=1.000 n=5+5)
LevelIterSeqSeekPrefixGE/skip=8/use-next=false-16                                              861ns ± 0%     844ns ± 4%    ~     (p=0.222 n=5+5)
LevelIterSeqSeekPrefixGE/skip=8/use-next=true-16                                               976ns ± 1%     972ns ± 3%    ~     (p=1.000 n=5+5)
LevelIterSeqSeekPrefixGE/skip=16/use-next=false-16                                             900ns ± 4%     915ns ± 2%    ~     (p=0.310 n=5+5)
LevelIterSeqSeekPrefixGE/skip=16/use-next=true-16                                              971ns ± 3%    1018ns ± 1%  +4.83%  (p=0.008 n=5+5)
LevelIterNext/restart=16/count=5-16                                                           25.4ns ± 1%    26.8ns ± 1%  +5.64%  (p=0.008 n=5+5)
LevelIterPrev/restart=16/count=5-16                                                           46.7ns ± 0%    48.0ns ± 1%  +2.87%  (p=0.008 n=5+5)
MergingIterSeekGE/restart=16/count=1-16                                                       1.20µs ± 2%    1.19µs ± 3%    ~     (p=0.246 n=5+5)
MergingIterSeekGE/restart=16/count=2-16                                                       2.34µs ± 1%    2.32µs ± 1%    ~     (p=0.151 n=5+5)
MergingIterSeekGE/restart=16/count=3-16                                                       3.58µs ± 2%    3.56µs ± 2%    ~     (p=0.421 n=5+5)
MergingIterSeekGE/restart=16/count=4-16                                                       4.97µs ± 2%    4.86µs ± 1%  -2.19%  (p=0.016 n=5+5)
MergingIterSeekGE/restart=16/count=5-16                                                       6.37µs ± 2%    6.50µs ± 2%    ~     (p=0.111 n=5+5)
MergingIterNext/restart=16/count=1-16                                                         47.6ns ± 1%    47.6ns ± 3%    ~     (p=1.000 n=5+5)
MergingIterNext/restart=16/count=2-16                                                         77.2ns ± 3%    76.5ns ± 1%    ~     (p=0.095 n=5+5)
MergingIterNext/restart=16/count=3-16                                                         95.3ns ± 0%    94.3ns ± 0%  -1.03%  (p=0.008 n=5+5)
MergingIterNext/restart=16/count=4-16                                                          109ns ± 0%     110ns ± 1%    ~     (p=0.667 n=5+5)
MergingIterNext/restart=16/count=5-16                                                          129ns ± 2%     127ns ± 1%    ~     (p=0.111 n=5+5)
MergingIterPrev/restart=16/count=1-16                                                         69.2ns ± 2%    69.8ns ± 1%    ~     (p=0.222 n=5+5)
MergingIterPrev/restart=16/count=2-16                                                         99.3ns ± 1%    99.6ns ± 1%    ~     (p=0.222 n=5+5)
MergingIterPrev/restart=16/count=3-16                                                          117ns ± 0%     117ns ± 0%    ~     (p=0.675 n=5+5)
MergingIterPrev/restart=16/count=4-16                                                          129ns ± 1%     129ns ± 0%    ~     (p=0.571 n=5+5)
MergingIterPrev/restart=16/count=5-16                                                          145ns ± 1%     147ns ± 1%  +0.85%  (p=0.032 n=5+5)
MergingIterSeqSeekGEWithBounds/levelCount=5-16                                                 605ns ± 0%     609ns ± 1%  +0.74%  (p=0.032 n=5+5)
MergingIterSeqSeekPrefixGE/skip=1/use-next=false-16                                           1.93µs ± 0%    1.96µs ± 1%  +1.57%  (p=0.008 n=5+5)
MergingIterSeqSeekPrefixGE/skip=1/use-next=true-16                                             499ns ± 1%     500ns ± 1%    ~     (p=1.000 n=5+5)
MergingIterSeqSeekPrefixGE/skip=2/use-next=false-16                                           1.95µs ± 0%    1.98µs ± 1%  +1.61%  (p=0.008 n=5+5)
MergingIterSeqSeekPrefixGE/skip=2/use-next=true-16                                             537ns ± 1%     540ns ± 1%    ~     (p=0.397 n=5+5)
MergingIterSeqSeekPrefixGE/skip=4/use-next=false-16                                           2.00µs ± 1%    2.01µs ± 0%    ~     (p=0.643 n=5+5)
MergingIterSeqSeekPrefixGE/skip=4/use-next=true-16                                             619ns ± 1%     627ns ± 2%  +1.34%  (p=0.032 n=5+5)
MergingIterSeqSeekPrefixGE/skip=8/use-next=false-16                                           2.14µs ± 0%    2.12µs ± 0%  -0.79%  (p=0.008 n=5+5)
MergingIterSeqSeekPrefixGE/skip=8/use-next=true-16                                            1.45µs ± 2%    1.41µs ± 2%  -2.65%  (p=0.024 n=5+5)
MergingIterSeqSeekPrefixGE/skip=16/use-next=false-16                                          2.31µs ± 1%    2.18µs ± 1%  -5.94%  (p=0.008 n=5+5)
MergingIterSeqSeekPrefixGE/skip=16/use-next=true-16                                           1.58µs ± 1%    1.44µs ± 2%  -8.71%  (p=0.008 n=5+5)

name                                              old time/op    new time/op    delta
BlockIterSeekGE/restart=16-16                        484ns ± 1%     484ns ± 0%    ~     (p=0.310 n=5+5)
BlockIterSeekLT/restart=16-16                        532ns ± 1%     531ns ± 0%    ~     (p=0.889 n=5+5)
BlockIterNext/restart=16-16                         18.8ns ± 0%    18.9ns ± 0%    ~     (p=0.222 n=5+5)
BlockIterPrev/restart=16-16                         39.2ns ± 0%    39.3ns ± 0%    ~     (p=0.325 n=5+5)
TableIterSeekGE/restart=16,compression=Snappy-16    1.58µs ± 2%    1.56µs ± 2%    ~     (p=0.333 n=5+5)
TableIterSeekGE/restart=16,compression=ZSTD-16      1.57µs ± 1%    1.56µs ± 2%    ~     (p=0.548 n=5+5)
TableIterSeekLT/restart=16,compression=Snappy-16    1.63µs ± 1%    1.63µs ± 3%    ~     (p=0.690 n=5+5)
TableIterSeekLT/restart=16,compression=ZSTD-16      1.63µs ± 2%    1.62µs ± 3%    ~     (p=0.595 n=5+5)
TableIterNext/restart=16,compression=Snappy-16      23.7ns ± 1%    24.1ns ± 1%  +1.41%  (p=0.016 n=5+5)
TableIterNext/restart=16,compression=ZSTD-16        23.8ns ± 1%    24.3ns ± 1%  +2.12%  (p=0.032 n=5+4)
TableIterPrev/restart=16,compression=Snappy-16      44.9ns ± 1%    45.3ns ± 1%  +0.91%  (p=0.016 n=5+5)
TableIterPrev/restart=16,compression=ZSTD-16        45.2ns ± 1%    45.3ns ± 1%    ~     (p=0.310 n=5+5)
```

Informs https://github.com/cockroachdb/pebble/issues/1170

cockroach-teamcity: 

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

sumeerbhola: TFTR!

Patch:
diff --git a/external_iterator.go b/external_iterator.go
index 29f5bb53..effe073c 100644
--- a/external_iterator.go
+++ b/external_iterator.go
@@ -199,6 +199,7 @@ func createExternalPointIter(it *Iterator) (internalIterator, error) {
 				nil,   /* BlockPropertiesFilterer */
 				false, /* useFilterBlock */
 				&it.stats.InternalStats,
+				sstable.TrivialReaderProvider{Reader: r},
 			)
 			if err != nil {
 				return nil, err
diff --git a/internal/base/lazy_value.go b/internal/base/lazy_value.go
index ef0c600b..846f9570 100644
--- a/internal/base/lazy_value.go
+++ b/internal/base/lazy_value.go
@@ -163,7 +163,8 @@ type LazyFetcher struct {
 	// If the fetcher attempted to use buf *and* len(buf) was insufficient, it
 	// will allocate a new slice for the value. In either case it will set
 	// callerOwned to true.
-	ValueFetcher func(handle []byte, buf []byte) (val []byte, callerOwned bool, err error)
+	ValueFetcher func(
+		handle []byte, valLen int32, buf []byte) (val []byte, callerOwned bool, err error)
 	// Attribute includes the short attribute and value length.
 	Attribute   AttributeAndLen
 	fetched     bool
@@ -194,7 +195,8 @@ func (lv *LazyValue) fetchValue(buf []byte) (val []byte, callerOwned bool, err e
 	f := lv.Fetcher
 	if !f.fetched {
 		f.fetched = true
-		f.value, f.callerOwned, f.err = lv.Fetcher.ValueFetcher(lv.ValueOrHandle, buf)
+		f.value, f.callerOwned, f.err = lv.Fetcher.ValueFetcher(
+			lv.ValueOrHandle, lv.Fetcher.Attribute.ValueLen, buf)
 	}
 	return f.value, f.callerOwned, f.err
 }
@@ -226,17 +228,29 @@ func (lv *LazyValue) TryGetShortAttribute() (ShortAttribute, bool) {
 }
 
 // Clone creates a stable copy of the LazyValue, by appending bytes to buf.
+// The fetcher parameter must be non-nil and may be over-written and used
+// inside the returned LazyValue -- this is needed to avoid an allocation.
+// Most callers have at most K cloned LazyValues, where K is hard-coded, so
+// they can have a pool of exactly K LazyFetcher structs they can reuse in
+// these calls. The alternative of allocating LazyFetchers from a sync.Pool is
+// not viable since we have no code trigger for returning to the pool
+// (LazyValues are simply GC'd).
+//
 // REQUIRES: LazyValue.Value() has not been called.
 // Ensure that this is only called before LazyValue.Value is called. We don't
-// actually care if it was in-place but we don't want confusing things about whether
-// we need to clone the value inside the fetcher.
-func (lv *LazyValue) Clone(buf []byte) (LazyValue, []byte) {
+// actually care if it was in-place but we don't want to complicate behavior
+// regarding whether we need to clone the value inside the fetcher.
+func (lv *LazyValue) Clone(buf []byte, fetcher *LazyFetcher) (LazyValue, []byte) {
 	if invariants.Enabled && lv.Fetcher != nil && lv.Fetcher.fetched {
 		panic("value has already been fetched")
 	}
 	// INVARIANT: LazyFetcher.value is nil, so there is nothing to copy there.
 	vLen := len(lv.ValueOrHandle)
-	lvCopy := LazyValue{Fetcher: lv.Fetcher}
+	var lvCopy LazyValue
+	if lv.Fetcher != nil {
+		*fetcher = *lv.Fetcher
+		lvCopy.Fetcher = fetcher
+	}
 	if vLen == 0 {
 		return lvCopy, buf
 	}
diff --git a/internal/metamorphic/options.go b/internal/metamorphic/options.go
index 13947243..65007610 100644
--- a/internal/metamorphic/options.go
+++ b/internal/metamorphic/options.go
@@ -253,6 +253,10 @@ func standardOptions() []*testOptions {
 		24: `
 [TestOptions]
   threads=1
+`,
+		25: `
+[Options]
+  enable_value_blocks=true
 `,
 	}
 
diff --git a/iterator.go b/iterator.go
index 0e57fab9..1fb2d5a3 100644
--- a/iterator.go
+++ b/iterator.go
@@ -173,6 +173,7 @@ type Iterator struct {
 	value  LazyValue
 	// For use in LazyValue.Clone.
 	valueBuf []byte
+	fetcher  base.LazyFetcher
 	// For use in LazyValue.Value.
 	lazyValueBuf []byte
 	valueCloser  io.Closer
@@ -922,7 +923,7 @@ func (i *Iterator) findPrevEntry(limit []byte) {
 			// call, so use valueBuf instead. Note that valueBuf is only used
 			// in this one instance; everywhere else (eg. in findNextEntry),
 			// we just point i.value to the unsafe i.iter-owned value buffer.
-			i.value, i.valueBuf = i.iterValue.Clone(i.valueBuf[:0])
+			i.value, i.valueBuf = i.iterValue.Clone(i.valueBuf[:0], &i.fetcher)
 			i.saveRangeKey()
 			i.iterValidityState = IterValid
 			i.iterKey, i.iterValue = i.iter.Prev()
diff --git a/options.go b/options.go
index 340d57b3..9259b1a8 100644
--- a/options.go
+++ b/options.go
@@ -1072,6 +1072,9 @@ func (o *Options) String() string {
 	fmt.Fprintf(&buf, "  compaction_debt_concurrency=%d\n", o.Experimental.CompactionDebtConcurrency)
 	fmt.Fprintf(&buf, "  comparer=%s\n", o.Comparer.Name)
 	fmt.Fprintf(&buf, "  disable_wal=%t\n", o.DisableWAL)
+	if o.Experimental.EnableValueBlocks {
+		fmt.Fprintf(&buf, "  enable_value_blocks=%t\n", o.Experimental.EnableValueBlocks)
+	}
 	fmt.Fprintf(&buf, "  flush_delay_delete_range=%s\n", o.FlushDelayDeleteRange)
 	fmt.Fprintf(&buf, "  flush_delay_range_key=%s\n", o.FlushDelayRangeKey)
 	fmt.Fprintf(&buf, "  flush_split_bytes=%d\n", o.FlushSplitBytes)
@@ -1276,6 +1279,8 @@ func (o *Options) Parse(s string, hooks *ParseHooks) error {
 				o.private.disableLazyCombinedIteration, err = strconv.ParseBool(value)
 			case "disable_wal":
 				o.DisableWAL, err = strconv.ParseBool(value)
+			case "enable_value_blocks":
+				o.Experimental.EnableValueBlocks, err = strconv.ParseBool(value)
 			case "flush_delay_delete_range":
 				o.FlushDelayDeleteRange, err = time.ParseDuration(value)
 			case "flush_delay_range_key":
@@ -1550,7 +1555,6 @@ func (o *Options) MakeWriterOptions(level int, format sstable.TableFormat) sstab
 		writerOpts.BlockPropertyCollectors = o.BlockPropertyCollectors
 	}
 	if format >= sstable.TableFormatPebblev3 {
-		writerOpts.EnableValueBlocks = o.Experimental.EnableValueBlocks
 		writerOpts.ShortAttributeExtractor = o.Experimental.ShortAttributeExtractor
 		writerOpts.RequiredInPlaceValueBound = o.Experimental.RequiredInPlaceValueBound
 	}
diff --git a/sstable/block.go b/sstable/block.go
index 0a2ff6af..3d7ae55e 100644
--- a/sstable/block.go
+++ b/sstable/block.go
@@ -263,6 +263,34 @@ type blockEntry struct {
 // range keys since there is only a single range deletion and range key block
 // per sstable and the blockIter will not release the bytes for the block until
 // it is closed.
+//
+// Note on why blockIter knows about lazyValueHandling:
+//
+// blockIter's positioning functions (that return a LazyValue), are too
+// complex to inline even prior to lazyValueHandling. blockIter.Next and
+// blockIter.First were by far the cheapest and had costs 195 and 180
+// respectively, which exceeds the budget of 80. We initially tried to keep
+// the lazyValueHandling logic out of blockIter by wrapping it with a
+// lazyValueDataBlockIter. singleLevelIter and twoLevelIter would use this
+// wrapped iter. The functions in lazyValueDataBlockIter were simple, in that
+// they called the corresponding blockIter func and then decided whether the
+// value was in fact in-place (so return immediately) or needed further
+// handling. But these also turned out too costly for mid-stack inlining since
+// simple calls like the following have a high cost that is barely under the
+// budget of 80
+//  k, v := i.data.SeekGE(key, flags)  // cost 74
+//  k, v := i.data.Next()              // cost 72
+//
+// We have 2 options for minimizing performance regressions:
+// - Include the lazyValueHandling logic in the already non-inlineable
+//   blockIter functions: Since most of the time is spent in data block iters,
+//   it is acceptable to take the small hit of unnecessary branching (which
+//   hopefully branch prediction will predict correctly) for other kinds of
+//   blocks.
+// - Duplicate the logic of singleLevelIterator and twoLevelIterator for the
+//   v3 sstable and only use the aforementioned lazyValueDataBlockIter for a
+//   v3 sstable. We would want to manage these copies via code generation.
+// We have picked the first option here.
 type blockIter struct {
 	cmp Compare
 	// offset is the byte index that marks where the current key/value is
@@ -300,6 +328,9 @@ type blockIter struct {
 	// val contains the value the iterator is currently pointed at. If non-nil,
 	// this points to a slice of the block data.
 	val []byte
+	// lazyValue is val turned into a LazyValue, whenever a positioning method
+	// returns a non-nil key-value pair.
+	lazyValue base.LazyValue
 	// ikey contains the decoded InternalKey the iterator is currently pointed
 	// at. Note that the memory backing ikey.UserKey is either data stored
 	// directly in the block, fullKey, or cachedBuf. The key stability guarantee
@@ -325,7 +356,11 @@ type blockIter struct {
 	cacheHandle cache.Handle
 	// The first key in the block. This is used by the caller to set bounds
 	// for block iteration for already loaded blocks.
-	firstKey InternalKey
+	firstKey          InternalKey
+	lazyValueHandling struct {
+		vbr            *valueBlockReader
+		hasValuePrefix bool
+	}
 }
 
 // blockIter implements the base.InternalIterator interface.
@@ -679,12 +714,27 @@ func (i *blockIter) SeekGE(key []byte, flags base.SeekGEFlags) (*InternalKey, ba
 	i.decodeInternalKey(i.key)
 
 	// Iterate from that restart point to somewhere >= the key sought.
-	for ; i.valid(); i.Next() {
+	if !i.valid() {
+		return nil, base.LazyValue{}
+	}
+	if base.InternalCompare(i.cmp, i.ikey, ikey) >= 0 {
+		// Initialize i.lazyValue
+		if !i.lazyValueHandling.hasValuePrefix ||
+			base.TrailerKind(i.ikey.Trailer) != InternalKeyKindSet {
+			i.lazyValue = base.MakeInPlaceValue(i.val)
+		} else if i.lazyValueHandling.vbr == nil || !isValueHandle(valuePrefix(i.val[0])) {
+			i.lazyValue = base.MakeInPlaceValue(i.val[1:])
+		} else {
+			i.lazyValue = i.lazyValueHandling.vbr.getLazyValueForPrefixAndValueHandle(i.val)
+		}
+		return &i.ikey, i.lazyValue
+	}
+	for i.Next(); i.valid(); i.Next() {
 		if base.InternalCompare(i.cmp, i.ikey, ikey) >= 0 {
-			return &i.ikey, base.MakeInPlaceValue(i.val)
+			// i.Next() has already initialized i.lazyValue.
+			return &i.ikey, i.lazyValue
 		}
 	}
-
 	return nil, base.LazyValue{}
 }
 
@@ -809,11 +859,12 @@ func (i *blockIter) SeekLT(key []byte, flags base.SeekLTFlags) (*InternalKey, ba
 
 		if i.cmp(i.ikey.UserKey, ikey.UserKey) >= 0 {
 			// The current key is greater than or equal to our search key. Back up to
-			// the previous key which was less than our search key. Note that his for
+			// the previous key which was less than our search key. Note that this for
 			// loop will execute at least once with this if-block not being true, so
 			// the key we are backing up to is the last one this loop cached.
 			i.Prev()
-			return &i.ikey, base.MakeInPlaceValue(i.val)
+			// i.Prev() has already initialized i.lazyValue.
+			return &i.ikey, i.lazyValue
 		}
 
 		if i.nextOffset >= targetOffset {
@@ -831,7 +882,15 @@ func (i *blockIter) SeekLT(key []byte, flags base.SeekLTFlags) (*InternalKey, ba
 	if !i.valid() {
 		return nil, base.LazyValue{}
 	}
-	return &i.ikey, base.MakeInPlaceValue(i.val)
+	if !i.lazyValueHandling.hasValuePrefix ||
+		base.TrailerKind(i.ikey.Trailer) != InternalKeyKindSet {
+		i.lazyValue = base.MakeInPlaceValue(i.val)
+	} else if i.lazyValueHandling.vbr == nil || !isValueHandle(valuePrefix(i.val[0])) {
+		i.lazyValue = base.MakeInPlaceValue(i.val[1:])
+	} else {
+		i.lazyValue = i.lazyValueHandling.vbr.getLazyValueForPrefixAndValueHandle(i.val)
+	}
+	return &i.ikey, i.lazyValue
 }
 
 // First implements internalIterator.First, as documented in the pebble
@@ -844,7 +903,15 @@ func (i *blockIter) First() (*InternalKey, base.LazyValue) {
 	i.clearCache()
 	i.readEntry()
 	i.decodeInternalKey(i.key)
-	return &i.ikey, base.MakeInPlaceValue(i.val)
+	if !i.lazyValueHandling.hasValuePrefix ||
+		base.TrailerKind(i.ikey.Trailer) != InternalKeyKindSet {
+		i.lazyValue = base.MakeInPlaceValue(i.val)
+	} else if i.lazyValueHandling.vbr == nil || !isValueHandle(valuePrefix(i.val[0])) {
+		i.lazyValue = base.MakeInPlaceValue(i.val[1:])
+	} else {
+		i.lazyValue = i.lazyValueHandling.vbr.getLazyValueForPrefixAndValueHandle(i.val)
+	}
+	return &i.ikey, i.lazyValue
 }
 
 func decodeRestart(b []byte) int32 {
@@ -871,7 +938,15 @@ func (i *blockIter) Last() (*InternalKey, base.LazyValue) {
 	}
 
 	i.decodeInternalKey(i.key)
-	return &i.ikey, base.MakeInPlaceValue(i.val)
+	if !i.lazyValueHandling.hasValuePrefix ||
+		base.TrailerKind(i.ikey.Trailer) != InternalKeyKindSet {
+		i.lazyValue = base.MakeInPlaceValue(i.val)
+	} else if i.lazyValueHandling.vbr == nil || !isValueHandle(valuePrefix(i.val[0])) {
+		i.lazyValue = base.MakeInPlaceValue(i.val[1:])
+	} else {
+		i.lazyValue = i.lazyValueHandling.vbr.getLazyValueForPrefixAndValueHandle(i.val)
+	}
+	return &i.ikey, i.lazyValue
 }
 
 // Next implements internalIterator.Next, as documented in the pebble
@@ -908,7 +983,15 @@ func (i *blockIter) Next() (*InternalKey, base.LazyValue) {
 		i.ikey.Trailer = uint64(InternalKeyKindInvalid)
 		i.ikey.UserKey = nil
 	}
-	return &i.ikey, base.MakeInPlaceValue(i.val)
+	if !i.lazyValueHandling.hasValuePrefix ||
+		base.TrailerKind(i.ikey.Trailer) != InternalKeyKindSet {
+		i.lazyValue = base.MakeInPlaceValue(i.val)
+	} else if i.lazyValueHandling.vbr == nil || !isValueHandle(valuePrefix(i.val[0])) {
+		i.lazyValue = base.MakeInPlaceValue(i.val[1:])
+	} else {
+		i.lazyValue = i.lazyValueHandling.vbr.getLazyValueForPrefixAndValueHandle(i.val)
+	}
+	return &i.ikey, i.lazyValue
 }
 
 // Prev implements internalIterator.Prev, as documented in the pebble
@@ -932,7 +1015,15 @@ func (i *blockIter) Prev() (*InternalKey, base.LazyValue) {
 			i.ikey.UserKey = nil
 		}
 		i.cached = i.cached[:n]
-		return &i.ikey, base.MakeInPlaceValue(i.val)
+		if !i.lazyValueHandling.hasValuePrefix ||
+			base.TrailerKind(i.ikey.Trailer) != InternalKeyKindSet {
+			i.lazyValue = base.MakeInPlaceValue(i.val)
+		} else if i.lazyValueHandling.vbr == nil || !isValueHandle(valuePrefix(i.val[0])) {
+			i.lazyValue = base.MakeInPlaceValue(i.val[1:])
+		} else {
+			i.lazyValue = i.lazyValueHandling.vbr.getLazyValueForPrefixAndValueHandle(i.val)
+		}
+		return &i.ikey, i.lazyValue
 	}
 
 	i.clearCache()
@@ -979,7 +1070,15 @@ func (i *blockIter) Prev() (*InternalKey, base.LazyValue) {
 	}
 
 	i.decodeInternalKey(i.key)
-	return &i.ikey, base.MakeInPlaceValue(i.val)
+	if !i.lazyValueHandling.hasValuePrefix ||
+		base.TrailerKind(i.ikey.Trailer) != InternalKeyKindSet {
+		i.lazyValue = base.MakeInPlaceValue(i.val)
+	} else if i.lazyValueHandling.vbr == nil || !isValueHandle(valuePrefix(i.val[0])) {
+		i.lazyValue = base.MakeInPlaceValue(i.val[1:])
+	} else {
+		i.lazyValue = i.lazyValueHandling.vbr.getLazyValueForPrefixAndValueHandle(i.val)
+	}
+	return &i.ikey, i.lazyValue
 }
 
 // Key implements internalIterator.Key, as documented in the pebble package.
@@ -988,7 +1087,7 @@ func (i *blockIter) Key() *InternalKey {
 }
 
 func (i *blockIter) value() base.LazyValue {
-	return base.MakeInPlaceValue(i.val)
+	return i.lazyValue
 }
 
 // Error implements internalIterator.Error, as documented in the pebble
@@ -1003,6 +1102,8 @@ func (i *blockIter) Close() error {
 	i.cacheHandle.Release()
 	i.cacheHandle = cache.Handle{}
 	i.val = nil
+	i.lazyValue = base.LazyValue{}
+	i.lazyValueHandling.vbr = nil
 	return nil
 }
 
diff --git a/sstable/options.go b/sstable/options.go
index 0b0117b8..716b5dc1 100644
--- a/sstable/options.go
+++ b/sstable/options.go
@@ -222,10 +222,6 @@ type WriterOptions struct {
 	// Writer client goroutine.
 	Parallelism bool
 
-	// EnableValueBlocks mirrors Options.Experimental.EnableValueBlocks. Must
-	// be false if the TableFormat is < TableFormatPebblev3.
-	EnableValueBlocks bool
-
 	// ShortAttributeExtractor mirrors
 	// Options.Experimental.ShortAttributeExtractor.
 	ShortAttributeExtractor base.ShortAttributeExtractor
diff --git a/sstable/reader.go b/sstable/reader.go
index 110b7d38..e5cc3ca5 100644
--- a/sstable/reader.go
+++ b/sstable/reader.go
@@ -210,6 +210,7 @@ type singleLevelIterator struct {
 	// loading. It may not actually have loaded the block, due to an error or
 	// because it was considered irrelevant.
 	dataBH    BlockHandle
+	vbReader  *valueBlockReader
 	err       error
 	closeHook func(i Iterator) error
 	stats     *base.InternalIteratorStats
@@ -388,6 +389,7 @@ func (i *singleLevelIterator) init(
 	filterer *BlockPropertiesFilterer,
 	useFilter bool,
 	stats *base.InternalIteratorStats,
+	rp ReaderProvider,
 ) error {
 	if r.err != nil {
 		return r.err
@@ -411,6 +413,18 @@ func (i *singleLevelIterator) init(
 		return err
 	}
 	i.dataRS.size = initialReadaheadSize
+	if r.tableFormat == TableFormatPebblev3 {
+		if r.Properties.NumValueBlocks > 0 {
+			i.vbReader = &valueBlockReader{
+				bpOpen: i,
+				rp:     rp,
+				vbih:   r.valueBIH,
+				stats:  stats,
+			}
+			i.data.lazyValueHandling.vbr = i.vbReader
+		}
+		i.data.lazyValueHandling.hasValuePrefix = true
+	}
 	return nil
 }
 
@@ -527,6 +541,18 @@ func (i *singleLevelIterator) loadBlock(dir int8) loadBlockResult {
 	return loadBlockOK
 }
 
+// readBlockForVBR implements the blockProviderWhenOpen interface for use by
+// the valueBlockReader. We could use a readaheadState for this (that would be
+// different from the readaheadState for the data blocks), but choose to use
+// nil since (a) for user-facing reads we expect access to the value blocks to
+// be rare, (b) for compactions, we are not using the logic in readaheadState
+// and deferring to OS-level readahead.
+func (i *singleLevelIterator) readBlockForVBR(
+	h BlockHandle, stats *base.InternalIteratorStats,
+) (cache.Handle, error) {
+	return i.reader.readBlock(h, nil /* transform */, nil /* raState */, stats)
+}
+
 // resolveMaybeExcluded is invoked when the block-property filterer has found
 // that a block is excluded according to its properties but only if its bounds
 // fall within the filter's current bounds.  This function consults the
@@ -1307,6 +1333,9 @@ func (i *singleLevelIterator) Close() error {
 	if i.bpfs != nil {
 		releaseBlockPropertiesFilterer(i.bpfs)
 	}
+	if i.vbReader != nil {
+		i.vbReader.close()
+	}
 	*i = i.resetForReuse()
 	singleLevelIterPool.Put(i)
 	return err
@@ -1572,6 +1601,7 @@ func (i *twoLevelIterator) init(
 	filterer *BlockPropertiesFilterer,
 	useFilter bool,
 	stats *base.InternalIteratorStats,
+	rp ReaderProvider,
 ) error {
 	if r.err != nil {
 		return r.err
@@ -1594,6 +1624,19 @@ func (i *twoLevelIterator) init(
 		_ = i.topLevelIndex.Close()
 		return err
 	}
+	i.dataRS.size = initialReadaheadSize
+	if r.tableFormat == TableFormatPebblev3 {
+		if r.Properties.NumValueBlocks > 0 {
+			i.vbReader = &valueBlockReader{
+				bpOpen: i,
+				rp:     rp,
+				vbih:   r.valueBIH,
+				stats:  stats,
+			}
+			i.data.lazyValueHandling.vbr = i.vbReader
+		}
+		i.data.lazyValueHandling.hasValuePrefix = true
+	}
 	return nil
 }
 
@@ -2200,6 +2243,9 @@ func (i *twoLevelIterator) Close() error {
 	if i.bpfs != nil {
 		releaseBlockPropertiesFilterer(i.bpfs)
 	}
+	if i.vbReader != nil {
+		i.vbReader.close()
+	}
 	*i = twoLevelIterator{
 		singleLevelIterator: i.singleLevelIterator.resetForReuse(),
 		topLevelIndex:       i.topLevelIndex.resetForReuse(),
@@ -2328,6 +2374,11 @@ type readaheadState struct {
 	// reading blocks in a sequential access pattern. Once this is non-nil,
 	// the other variables in readaheadState don't matter much as we defer
 	// to OS-level readahead.
+	//
+	// For TableFormatPebblev3, there are potentially two different sequential
+	// accesses, for the data block and the value blocks. It is unclear to me
+	// how FADV_SEQUENTIAL will function for such accesses. We do expect that
+	// the average workload will have most data in data blocks.
 	sequentialFile vfs.File
 }
 
@@ -2613,6 +2664,7 @@ type Reader struct {
 	rangeDelBH        BlockHandle
 	rangeKeyBH        BlockHandle
 	rangeDelTransform blockTransform
+	valueBIH          valueBlocksIndexHandle
 	propertiesBH      BlockHandle
 	metaIndexBH       BlockHandle
 	footerBH          BlockHandle
@@ -2661,13 +2713,14 @@ func (r *Reader) NewIterWithBlockPropertyFilters(
 	filterer *BlockPropertiesFilterer,
 	useFilterBlock bool,
 	stats *base.InternalIteratorStats,
+	rp ReaderProvider,
 ) (Iterator, error) {
 	// NB: pebble.tableCache wraps the returned iterator with one which performs
 	// reference counting on the Reader, preventing the Reader from being closed
 	// until the final iterator closes.
 	if r.Properties.IndexType == twoLevelIndex {
 		i := twoLevelIterPool.Get().(*twoLevelIterator)
-		err := i.init(r, lower, upper, filterer, useFilterBlock, stats)
+		err := i.init(r, lower, upper, filterer, useFilterBlock, stats, rp)
 		if err != nil {
 			return nil, err
 		}
@@ -2675,7 +2728,7 @@ func (r *Reader) NewIterWithBlockPropertyFilters(
 	}
 
 	i := singleLevelIterPool.Get().(*singleLevelIterator)
-	err := i.init(r, lower, upper, filterer, useFilterBlock, stats)
+	err := i.init(r, lower, upper, filterer, useFilterBlock, stats, rp)
 	if err != nil {
 		return nil, err
 	}
@@ -2683,18 +2736,23 @@ func (r *Reader) NewIterWithBlockPropertyFilters(
 }
 
 // NewIter returns an iterator for the contents of the table. If an error
-// occurs, NewIter cleans up after itself and returns a nil iterator.
+// occurs, NewIter cleans up after itself and returns a nil iterator. NewIter
+// must only be used when the Reader is guaranteed to outlive any LazyValues
+// returned from the iter.
 func (r *Reader) NewIter(lower, upper []byte) (Iterator, error) {
-	return r.NewIterWithBlockPropertyFilters(lower, upper, nil, true /* useFilterBlock */, nil /* stats */)
+	return r.NewIterWithBlockPropertyFilters(
+		lower, upper, nil, true /* useFilterBlock */, nil, /* stats */
+		TrivialReaderProvider{Reader: r})
 }
 
 // NewCompactionIter returns an iterator similar to NewIter but it also increments
 // the number of bytes iterated. If an error occurs, NewCompactionIter cleans up
 // after itself and returns a nil iterator.
-func (r *Reader) NewCompactionIter(bytesIterated *uint64) (Iterator, error) {
+func (r *Reader) NewCompactionIter(bytesIterated *uint64, rp ReaderProvider) (Iterator, error) {
 	if r.Properties.IndexType == twoLevelIndex {
 		i := twoLevelIterPool.Get().(*twoLevelIterator)
-		err := i.init(r, nil /* lower */, nil /* upper */, nil, false /* useFilter */, nil /* stats */)
+		err := i.init(
+			r, nil /* lower */, nil /* upper */, nil, false /* useFilter */, nil /* stats */, rp)
 		if err != nil {
 			return nil, err
 		}
@@ -2705,7 +2763,8 @@ func (r *Reader) NewCompactionIter(bytesIterated *uint64) (Iterator, error) {
 		}, nil
 	}
 	i := singleLevelIterPool.Get().(*singleLevelIterator)
-	err := i.init(r, nil /* lower */, nil /* upper */, nil, false /* useFilter */, nil /* stats */)
+	err := i.init(
+		r, nil /* lower */, nil /* upper */, nil, false /* useFilter */, nil /* stats */, rp)
 	if err != nil {
 		return nil, err
 	}
@@ -2969,11 +3028,23 @@ func (r *Reader) readMetaindex(metaindexBH BlockHandle) error {
 
 	meta := map[string]BlockHandle{}
 	for valid := i.First(); valid; valid = i.Next() {
-		bh, n := decodeBlockHandle(i.Value())
-		if n == 0 {
-			return base.CorruptionErrorf("pebble/table: invalid table (bad filter block handle)")
+		value := i.Value()
+		if bytes.Equal(i.Key().UserKey, []byte(metaValueIndexName)) {
+			vbih, n, err := decodeValueBlocksIndexHandle(i.Value())
+			if err != nil {
+				return err
+			}
+			if n == 0 || n != len(value) {
+				return base.CorruptionErrorf("pebble/table: invalid table (bad value blocks index handle)")
+			}
+			r.valueBIH = vbih
+		} else {
+			bh, n := decodeBlockHandle(value)
+			if n == 0 || n != len(value) {
+				return base.CorruptionErrorf("pebble/table: invalid table (bad block handle)")
+			}
+			meta[string(i.Key().UserKey)] = bh
 		}
-		meta[string(i.Key().UserKey)] = bh
 	}
 	if err := i.Close(); err != nil {
 		return err
@@ -3046,9 +3117,11 @@ func (r *Reader) Layout() (*Layout, error) {
 		Filter:     r.filterBH,
 		RangeDel:   r.rangeDelBH,
 		RangeKey:   r.rangeKeyBH,
+		ValueIndex: r.valueBIH.h,
 		Properties: r.propertiesBH,
 		MetaIndex:  r.metaIndexBH,
 		Footer:     r.footerBH,
+		Format:     r.tableFormat,
 	}
 
 	indexH, err := r.readIndex(nil /* stats */)
@@ -3115,6 +3188,39 @@ func (r *Reader) Layout() (*Layout, error) {
 			*iter = iter.resetForReuse()
 		}
 	}
+	if r.valueBIH.h.Length != 0 {
+		vbiH, err := r.readBlock(r.valueBIH.h, nil, nil, nil)
+		if err != nil {
+			return nil, err
+		}
+		defer vbiH.Release()
+		vbiBlock := vbiH.Get()
+		indexEntryLen := int(r.valueBIH.blockNumByteLength + r.valueBIH.blockOffsetByteLength +
+			r.valueBIH.blockLengthByteLength)
+		i := 0
+		for len(vbiBlock) != 0 {
+			if len(vbiBlock) < indexEntryLen {
+				return nil, errors.Errorf(
+					"remaining value index block %d does not contain a full entry of length %d",
+					len(vbiBlock), indexEntryLen)
+			}
+			n := int(r.valueBIH.blockNumByteLength)
+			bn := int(littleEndianGet(vbiBlock, n))
+			if bn != i {
+				return nil, errors.Errorf("unexpected block num %d, expected %d",
+					bn, i)
+			}
+			i++
+			vbiBlock = vbiBlock[n:]
+			n = int(r.valueBIH.blockOffsetByteLength)
+			blockOffset := littleEndianGet(vbiBlock, n)
+			vbiBlock = vbiBlock[n:]
+			n = int(r.valueBIH.blockLengthByteLength)
+			blockLen := littleEndianGet(vbiBlock, n)
+			vbiBlock = vbiBlock[n:]
+			l.ValueBlock = append(l.ValueBlock, BlockHandle{Offset: blockOffset, Length: blockLen})
+		}
+	}
 
 	return l, nil
 }
@@ -3405,9 +3511,12 @@ type Layout struct {
 	Filter     BlockHandle
 	RangeDel   BlockHandle
 	RangeKey   BlockHandle
+	ValueBlock []BlockHandle
+	ValueIndex BlockHandle
 	Properties BlockHandle
 	MetaIndex  BlockHandle
 	Footer     BlockHandle
+	Format     TableFormat
 }
 
 // Describe returns a description of the layout. If the verbose parameter is
@@ -3439,6 +3548,12 @@ func (l *Layout) Describe(
 	if l.RangeKey.Length != 0 {
 		blocks = append(blocks, block{l.RangeKey, "range-key"})
 	}
+	for i := range l.ValueBlock {
+		blocks = append(blocks, block{l.ValueBlock[i], "value-block"})
+	}
+	if l.ValueIndex.Length != 0 {
+		blocks = append(blocks, block{l.ValueIndex, "value-index"})
+	}
 	if l.Properties.Length != 0 {
 		blocks = append(blocks, block{l.Properties, "properties"})
 	}
@@ -3575,12 +3690,23 @@ func (l *Layout) Describe(
 				formatIsRestart(iter.data, iter.restarts, iter.numRestarts, iter.offset)
 				if fmtRecord != nil {
 					fmt.Fprintf(w, "              ")
-					// InPlaceValue() will succeed even for data blocks where the actual
-					// value is in a different location, since this value was fetched
-					// from blockIter, which only has local knowledge.
-					// TODO(sumeer): adjust the formatting to account for the difference
-					// between the real value and a handle to that value.
-					fmtRecord(key, value.InPlaceValue())
+					if l.Format != TableFormatPebblev3 {
+						fmtRecord(key, value.InPlaceValue())
+					} else {
+						// InPlaceValue() will succeed even for data blocks where the
+						// actual value is in a different location, since this value was
+						// fetched from a blockIter which does not know about value
+						// blocks.
+						v := value.InPlaceValue()
+						if base.TrailerKind(key.Trailer) != InternalKeyKindSet {
+							fmtRecord(key, v)
+						} else if !isValueHandle(valuePrefix(v[0])) {
+							fmtRecord(key, v[1:])
+						} else {
+							vh := decodeValueHandle(v[1:])
+							fmtRecord(key, []byte(fmt.Sprintf("value handle %+v", vh)))
+						}
+					}
 				}
 
 				if base.InternalCompare(r.Compare, lastKey, *key) >= 0 {
@@ -3618,19 +3744,39 @@ func (l *Layout) Describe(
 			iter, _ := newRawBlockIter(r.Compare, h.Get())
 			for valid := iter.First(); valid; valid = iter.Next() {
 				value := iter.Value()
-				bh, n := decodeBlockHandle(value)
+				var bh BlockHandle
+				var n int
+				var vbih valueBlocksIndexHandle
+				isValueBlocksIndexHandle := false
+				if bytes.Equal(iter.Key().UserKey, []byte(metaValueIndexName)) {
+					vbih, n, err = decodeValueBlocksIndexHandle(value)
+					bh = vbih.h
+					isValueBlocksIndexHandle = true
+				} else {
+					bh, n = decodeBlockHandle(value)
+				}
 				if n == 0 || n != len(value) {
 					fmt.Fprintf(w, "%10d    [err: %s]\n", b.Offset+uint64(iter.offset), err)
 					continue
 				}
-
-				fmt.Fprintf(w, "%10d    %s block:%d/%d",
+				var vbihStr string
+				if isValueBlocksIndexHandle {
+					vbihStr = fmt.Sprintf(" value-blocks-index-lengths: %d(num), %d(offset), %d(length)",
+						vbih.blockNumByteLength, vbih.blockOffsetByteLength, vbih.blockLengthByteLength)
+				}
+				fmt.Fprintf(w, "%10d    %s block:%d/%d%s",
 					b.Offset+uint64(iter.offset), iter.Key().UserKey,
-					bh.Offset, bh.Length)
+					bh.Offset, bh.Length, vbihStr)
 				formatIsRestart(iter.data, iter.restarts, iter.numRestarts, iter.offset)
 			}
 			formatRestarts(iter.data, iter.restarts, iter.numRestarts)
 			formatTrailer()
+		case "value-block":
+			// We don't peer into the value-block since it can't be interpreted
+			// without the valueHandles.
+		case "value-index":
+			// We have already read the value-index to construct the list of
+			// value-blocks, so no need to do it again.
 		}
 
 		h.Release()
diff --git a/sstable/suffix_rewriter.go b/sstable/suffix_rewriter.go
index 2e137399..e75bf98c 100644
--- a/sstable/suffix_rewriter.go
+++ b/sstable/suffix_rewriter.go
@@ -39,14 +39,6 @@ func RewriteKeySuffixes(
 	if err != nil {
 		return nil, err
 	}
-	// TODO(sumeer): Add support for value blocks. The valueHandles do not need
-	// to change. We need to (a) copy the value blocks, (b) rewrite the meta
-	// value index block as the block offsets can change, (c) update the entry
-	// in the metaindex block entry for the meta value index block. Also handle
-	// the set-has-same-prefix bit in restarts.
-	if r.tableFormat > TableFormatPebblev2 {
-		return nil, errors.Errorf("efficient key suffix rewriting is unsupported with value blocks")
-	}
 	defer r.Close()
 	return rewriteKeySuffixesInBlocks(r, out, o, from, to, concurrency)
 }
@@ -60,6 +52,9 @@ func rewriteKeySuffixesInBlocks(
 	if concurrency < 1 {
 		return nil, errors.New("concurrency must be >= 1")
 	}
+	if r.Properties.NumValueBlocks > 0 {
+		return nil, errors.New("sstable with a single suffix should not have value blocks")
+	}
 
 	w := NewWriter(out, o)
 	defer w.Close()
diff --git a/sstable/value_block.go b/sstable/value_block.go
index 0cd99c1d..852af716 100644
--- a/sstable/value_block.go
+++ b/sstable/value_block.go
@@ -12,6 +12,7 @@ import (
 
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/cache"
 	"github.com/cockroachdb/pebble/internal/invariants"
 	"golang.org/x/exp/rand"
 )
@@ -190,41 +191,76 @@ func setHasSamePrefix(b valuePrefix) bool {
 	return b&setHasSameKeyPrefixMask == setHasSameKeyPrefixMask
 }
 
-func decodeValueHandle(src []byte) (valueHandle, error) {
+func decodeLenFromValueHandle(src []byte) (uint32, []byte) {
+	ptr := unsafe.Pointer(&src[0])
+	var v uint32
+	if a := *((*uint8)(ptr)); a < 128 {
+		v = uint32(a)
+		src = src[1:]
+	} else if a, b := a&0x7f, *((*uint8)(unsafe.Pointer(uintptr(ptr) + 1))); b < 128 {
+		v = uint32(b)<<7 | uint32(a)
+		src = src[2:]
+	} else if b, c := b&0x7f, *((*uint8)(unsafe.Pointer(uintptr(ptr) + 2))); c < 128 {
+		v = uint32(c)<<14 | uint32(b)<<7 | uint32(a)
+		src = src[3:]
+	} else if c, d := c&0x7f, *((*uint8)(unsafe.Pointer(uintptr(ptr) + 3))); d < 128 {
+		v = uint32(d)<<21 | uint32(c)<<14 | uint32(b)<<7 | uint32(a)
+		src = src[4:]
+	} else {
+		d, e := d&0x7f, *((*uint8)(unsafe.Pointer(uintptr(ptr) + 4)))
+		v = uint32(e)<<28 | uint32(d)<<21 | uint32(c)<<14 | uint32(b)<<7 | uint32(a)
+		src = src[5:]
+	}
+	return v, src
+}
+
+func decodeRemainingValueHandle(src []byte) valueHandle {
 	var vh valueHandle
 	ptr := unsafe.Pointer(&src[0])
-	for i := 0; i < 3; i++ {
-		// Manually inlined uvarint decoding. Saves ~25% in
-		// BenchmarkValueBlocks/valueSize=100/versions=10/needValue=true/hasCache=true.
-		// If needed we can also unroll the loop, as done in blockIter.readEntry.
-		var v uint32
-		if a := *((*uint8)(ptr)); a < 128 {
-			v = uint32(a)
-			ptr = unsafe.Pointer(uintptr(ptr) + 1)
-		} else if a, b := a&0x7f, *((*uint8)(unsafe.Pointer(uintptr(ptr) + 1))); b < 128 {
-			v = uint32(b)<<7 | uint32(a)
-			ptr = unsafe.Pointer(uintptr(ptr) + 2)
-		} else if b, c := b&0x7f, *((*uint8)(unsafe.Pointer(uintptr(ptr) + 2))); c < 128 {
-			v = uint32(c)<<14 | uint32(b)<<7 | uint32(a)
-			ptr = unsafe.Pointer(uintptr(ptr) + 3)
-		} else if c, d := c&0x7f, *((*uint8)(unsafe.Pointer(uintptr(ptr) + 3))); d < 128 {
-			v = uint32(d)<<21 | uint32(c)<<14 | uint32(b)<<7 | uint32(a)
-			ptr = unsafe.Pointer(uintptr(ptr) + 4)
-		} else {
-			d, e := d&0x7f, *((*uint8)(unsafe.Pointer(uintptr(ptr) + 4)))
-			v = uint32(e)<<28 | uint32(d)<<21 | uint32(c)<<14 | uint32(b)<<7 | uint32(a)
-			ptr = unsafe.Pointer(uintptr(ptr) + 5)
-		}
-		switch i {
-		case 0:
-			vh.valueLen = v
-		case 1:
-			vh.blockNum = v
-		case 2:
-			vh.offsetInBlock = v
-		}
+	// Manually inlined uvarint decoding. Saves ~25% in benchmarks. Unrolling
+	// a loop for i:=0; i<2; i++, saves ~6%.
+	var v uint32
+	if a := *((*uint8)(ptr)); a < 128 {
+		v = uint32(a)
+		ptr = unsafe.Pointer(uintptr(ptr) + 1)
+	} else if a, b := a&0x7f, *((*uint8)(unsafe.Pointer(uintptr(ptr) + 1))); b < 128 {
+		v = uint32(b)<<7 | uint32(a)
+		ptr = unsafe.Pointer(uintptr(ptr) + 2)
+	} else if b, c := b&0x7f, *((*uint8)(unsafe.Pointer(uintptr(ptr) + 2))); c < 128 {
+		v = uint32(c)<<14 | uint32(b)<<7 | uint32(a)
+		ptr = unsafe.Pointer(uintptr(ptr) + 3)
+	} else if c, d := c&0x7f, *((*uint8)(unsafe.Pointer(uintptr(ptr) + 3))); d < 128 {
+		v = uint32(d)<<21 | uint32(c)<<14 | uint32(b)<<7 | uint32(a)
+		ptr = unsafe.Pointer(uintptr(ptr) + 4)
+	} else {
+		d, e := d&0x7f, *((*uint8)(unsafe.Pointer(uintptr(ptr) + 4)))
+		v = uint32(e)<<28 | uint32(d)<<21 | uint32(c)<<14 | uint32(b)<<7 | uint32(a)
+		ptr = unsafe.Pointer(uintptr(ptr) + 5)
 	}
-	return vh, nil
+	vh.blockNum = v
+
+	if a := *((*uint8)(ptr)); a < 128 {
+		v = uint32(a)
+	} else if a, b := a&0x7f, *((*uint8)(unsafe.Pointer(uintptr(ptr) + 1))); b < 128 {
+		v = uint32(b)<<7 | uint32(a)
+	} else if b, c := b&0x7f, *((*uint8)(unsafe.Pointer(uintptr(ptr) + 2))); c < 128 {
+		v = uint32(c)<<14 | uint32(b)<<7 | uint32(a)
+	} else if c, d := c&0x7f, *((*uint8)(unsafe.Pointer(uintptr(ptr) + 3))); d < 128 {
+		v = uint32(d)<<21 | uint32(c)<<14 | uint32(b)<<7 | uint32(a)
+	} else {
+		d, e := d&0x7f, *((*uint8)(unsafe.Pointer(uintptr(ptr) + 4)))
+		v = uint32(e)<<28 | uint32(d)<<21 | uint32(c)<<14 | uint32(b)<<7 | uint32(a)
+	}
+	vh.offsetInBlock = v
+
+	return vh
+}
+
+func decodeValueHandle(src []byte) valueHandle {
+	valLen, src := decodeLenFromValueHandle(src)
+	vh := decodeRemainingValueHandle(src)
+	vh.valueLen = valLen
+	return vh
 }
 
 // valueBlocksIndexHandle is placed in the metaindex if there are any value
@@ -335,8 +371,13 @@ func releaseToValueBlockBufPool(pool *sync.Pool, b *blockBuffer) {
 		return
 	}
 	if invariants.Enabled {
-		// Set the bytes to a random value.
-		b.b = b.b[:cap(b.b)]
+		// Set the bytes to a random value. Cap the number of bytes being
+		// randomized to prevent test timeouts.
+		length := cap(b.b)
+		if length > 1000 {
+			length = 1000
+		}
+		b.b = b.b[:length:length]
 		rand.Read(b.b)
 	}
 	pool.Put(b)
@@ -396,6 +437,9 @@ func releaseValueBlockWriter(w *valueBlockWriter) {
 }
 
 func (w *valueBlockWriter) addValue(v []byte) (valueHandle, error) {
+	if invariants.Enabled && len(v) == 0 {
+		return valueHandle{}, errors.Errorf("cannot write empty value to value block")
+	}
 	w.numValues++
 	blockLen := len(w.buf.b)
 	valueLen := len(v)
@@ -449,12 +493,10 @@ func (w *valueBlockWriter) compressAndFlush() {
 		}
 	}
 	n := len(b.b)
-	allocated := false
 	if n+blockTrailerLen > cap(b.b) {
 		block := make([]byte, n+blockTrailerLen)
 		copy(block, b.b)
 		b.b = block
-		allocated = true
 	} else {
 		b.b = b.b[:n+blockTrailerLen]
 	}
@@ -470,13 +512,11 @@ func (w *valueBlockWriter) compressAndFlush() {
 		handle:     bh,
 		compressed: compressed,
 	})
-	if !allocated {
-		// Handed off a buffer to w.blocks, so need get a new one.
-		if compressed {
-			w.compressedBuf = compressedValueBlockBufPool.Get().(*blockBuffer)
-		} else {
-			w.buf = uncompressedValueBlockBufPool.Get().(*blockBuffer)
-		}
+	// Handed off a buffer to w.blocks, so need get a new one.
+	if compressed {
+		w.compressedBuf = compressedValueBlockBufPool.Get().(*blockBuffer)
+	} else {
+		w.buf = uncompressedValueBlockBufPool.Get().(*blockBuffer)
 	}
 	w.buf.b = w.buf.b[:0]
 }
@@ -590,6 +630,15 @@ func lenLittleEndian(v uint64) int {
 	return n
 }
 
+func littleEndianGet(b []byte, n int) uint64 {
+	_ = b[n-1] // bounds check
+	v := uint64(b[0])
+	for i := 1; i < n; i++ {
+		v |= uint64(b[i]) << (8 * i)
+	}
+	return v
+}
+
 // UserKeyPrefixBound represents a [Lower,Upper) bound of user key prefixes.
 // If both are nil, there is no bound specified. Else, Compare(Lower,Upper)
 // must be < 0.
@@ -604,3 +653,195 @@ type UserKeyPrefixBound struct {
 func (ukb *UserKeyPrefixBound) IsEmpty() bool {
 	return len(ukb.Lower) == 0 && len(ukb.Upper) == 0
 }
+
+type blockProviderWhenOpen interface {
+	readBlockForVBR(h BlockHandle, stats *base.InternalIteratorStats) (cache.Handle, error)
+}
+
+type blockProviderWhenClosed struct {
+	rp ReaderProvider
+	r  *Reader
+}
+
+func (bpwc *blockProviderWhenClosed) open() error {
+	var err error
+	bpwc.r, err = bpwc.rp.GetReader()
+	return err
+}
+
+func (bpwc *blockProviderWhenClosed) close() {
+	bpwc.rp.Close()
+	bpwc.r = nil
+}
+
+func (bpwc blockProviderWhenClosed) readBlockForVBR(
+	h BlockHandle, stats *base.InternalIteratorStats,
+) (cache.Handle, error) {
+	return bpwc.r.readBlock(h, nil, nil, stats)
+}
+
+// ReaderProvider supports the implementation of blockProviderWhenClosed.
+// GetReader and Close can be called multiple times in pairs.
+type ReaderProvider interface {
+	GetReader() (r *Reader, err error)
+	Close()
+}
+
+// TrivialReaderProvider implements ReaderProvider for a Reader that will
+// outlive the top-level iterator in the iterator tree.
+type TrivialReaderProvider struct {
+	*Reader
+}
+
+var _ ReaderProvider = TrivialReaderProvider{}
+
+// GetReader implements ReaderProvider.
+func (trp TrivialReaderProvider) GetReader() (*Reader, error) {
+	return trp.Reader, nil
+}
+
+// Close implements ReaderProvider.
+func (trp TrivialReaderProvider) Close() {}
+
+// valueBlockReader is used to retrieve values in value
+// blocks. It is used when the sstable was written with
+// Properties.ValueBlocksAreEnabled.
+type valueBlockReader struct {
+	bpOpen blockProviderWhenOpen
+	rp     ReaderProvider
+	vbih   valueBlocksIndexHandle
+	stats  *base.InternalIteratorStats
+
+	// The value blocks index is lazily retrieved the first time the reader
+	// needs to read a value that resides in a value block.
+	vbiBlock []byte
+	vbiCache cache.Handle
+	// When sequentially iterating through all key-value pairs, the cost of
+	// repeatedly getting a block that is already in the cache and releasing the
+	// cache.Handle can be ~40% of the cpu overhead. So the reader remembers the
+	// last value block it retrieved, in case there is locality of access, and
+	// this value block can be used for the next value retrieval.
+	valueBlockNum uint32
+	valueBlock    []byte
+	valueBlockPtr unsafe.Pointer
+	valueCache    cache.Handle
+	lazyFetcher   base.LazyFetcher
+	closed        bool
+}
+
+func (r *valueBlockReader) getLazyValueForPrefixAndValueHandle(handle []byte) base.LazyValue {
+	fetcher := &r.lazyFetcher
+	valLen, h := decodeLenFromValueHandle(handle[1:])
+	*fetcher = base.LazyFetcher{
+		ValueFetcher: r.getValue,
+		Attribute: base.AttributeAndLen{
+			ValueLen:       int32(valLen),
+			ShortAttribute: getShortAttribute(valuePrefix(handle[0])),
+		},
+	}
+	return base.LazyValue{
+		ValueOrHandle: h,
+		Fetcher:       fetcher,
+	}
+}
+
+func (r *valueBlockReader) close() {
+	r.bpOpen = nil
+	r.vbiBlock = nil
+	r.vbiCache.Release()
+	// Set the handle to empty since Release does not nil the Handle.value. If
+	// we were to reopen this valueBlockReader and retrieve the same
+	// Handle.value from the cache, we don't want to accidentally unref it when
+	// attempting to unref the old handle.
+	r.vbiCache = cache.Handle{}
+	r.valueBlock = nil
+	r.valueBlockPtr = nil
+	r.valueCache.Release()
+	// See comment above.
+	r.valueCache = cache.Handle{}
+	r.closed = true
+	// rp, vbih, stats remain valid, so that LazyFetcher.ValueFetcher can be
+	// implemented.
+}
+
+func (r *valueBlockReader) getValue(
+	handle []byte, valLen int32, buf []byte,
+) (val []byte, callerOwned bool, err error) {
+	if !r.closed {
+		val, err := r.getValueInternal(handle, valLen)
+		return val, false, err
+	}
+
+	bp := blockProviderWhenClosed{rp: r.rp}
+	err = bp.open()
+	if err != nil {
+		return nil, false, err
+	}
+	defer bp.close()
+	defer r.close()
+	r.bpOpen = bp
+	var v []byte
+	v, err = r.getValueInternal(handle, valLen)
+	if err != nil {
+		return nil, false, err
+	}
+	if invariants.Enabled && callerOwned {
+		panic("callerOwned must be false")
+	}
+	buf = append(buf[:0], v...)
+	return buf, true, nil
+}
+
+func (r *valueBlockReader) getValueInternal(handle []byte, valLen int32) (val []byte, err error) {
+	vh := decodeRemainingValueHandle(handle)
+	vh.valueLen = uint32(valLen)
+	if r.vbiBlock == nil {
+		ch, err := r.bpOpen.readBlockForVBR(r.vbih.h, r.stats)
+		if err != nil {
+			return nil, err
+		}
+		r.vbiCache = ch
+		r.vbiBlock = ch.Get()
+	}
+	if r.valueBlock == nil || r.valueBlockNum != vh.blockNum {
+		vbh, err := r.getBlockHandle(vh.blockNum)
+		if err != nil {
+			return nil, err
+		}
+		vbCacheHandle, err := r.bpOpen.readBlockForVBR(vbh, r.stats)
+		if err != nil {
+			return nil, err
+		}
+		r.valueBlockNum = vh.blockNum
+		r.valueCache.Release()
+		r.valueCache = vbCacheHandle
+		r.valueBlock = vbCacheHandle.Get()
+		r.valueBlockPtr = unsafe.Pointer(&r.valueBlock[0])
+	}
+	return r.valueBlock[vh.offsetInBlock : vh.offsetInBlock+vh.valueLen], nil
+}
+
+func (r *valueBlockReader) getBlockHandle(blockNum uint32) (BlockHandle, error) {
+	indexEntryLen :=
+		int(r.vbih.blockNumByteLength + r.vbih.blockOffsetByteLength + r.vbih.blockLengthByteLength)
+	offsetInIndex := indexEntryLen * int(blockNum)
+	if len(r.vbiBlock) < offsetInIndex+indexEntryLen {
+		return BlockHandle{}, errors.Errorf(
+			"cannot read at offset %d and length %d from block of length %d",
+			offsetInIndex, indexEntryLen, len(r.vbiBlock))
+	}
+	b := r.vbiBlock[offsetInIndex : offsetInIndex+indexEntryLen]
+	n := int(r.vbih.blockNumByteLength)
+	bn := littleEndianGet(b, n)
+	if uint32(bn) != blockNum {
+		return BlockHandle{},
+			errors.Errorf("expected block num %d but found %d", blockNum, bn)
+	}
+	b = b[n:]
+	n = int(r.vbih.blockOffsetByteLength)
+	blockOffset := littleEndianGet(b, n)
+	b = b[n:]
+	n = int(r.vbih.blockLengthByteLength)
+	blockLen := littleEndianGet(b, n)
+	return BlockHandle{Offset: blockOffset, Length: blockLen}, nil
+}
diff --git a/sstable/writer.go b/sstable/writer.go
index 13dbdd2f..f9554623 100644
--- a/sstable/writer.go
+++ b/sstable/writer.go
@@ -770,7 +770,7 @@ func (w *Writer) makeAddPointDecisionV2(key InternalKey) error {
 }
 
 func (w *Writer) makeAddPointDecisionV3(
-	key InternalKey,
+	key InternalKey, valueLen int,
 ) (setHasSamePrefix bool, writeToValueBlock bool, err error) {
 	prevPointKeyInfo := w.lastPointKeyInfo
 	w.lastPointKeyInfo.userKeyLen = len(key.UserKey)
@@ -831,17 +831,22 @@ func (w *Writer) makeAddPointDecisionV3(
 	// NB: it is possible that cmpUser == 0, i.e., these two SETs have identical
 	// user keys (because of an open snapshot). This should be the rare case.
 	setHasSamePrefix = cmpPrefix == 0
-	// Currently we write to the value block exactly when consecutive sets have
-	// the same prefix. We may expand the cases we write to the value block in
-	// the future.
-	return setHasSamePrefix, setHasSamePrefix, nil
+	considerWriteToValueBlock = setHasSamePrefix
+	// Use of 0 here is somewhat arbitrary. Given the minimum 3 byte encoding of
+	// valueHandle, this should be > 3. But tiny values are common in test and
+	// unlikely in production, so we use 0 here for better test coverage.
+	const tinyValueThreshold = 0
+	if considerWriteToValueBlock && valueLen <= tinyValueThreshold {
+		considerWriteToValueBlock = false
+	}
+	return setHasSamePrefix, considerWriteToValueBlock, nil
 }
 
 func (w *Writer) addPoint(key InternalKey, value []byte) error {
 	var err error
 	var setHasSameKeyPrefix, writeToValueBlock, addPrefixToValueStoredWithKey bool
 	if w.valueBlockWriter != nil {
-		setHasSameKeyPrefix, writeToValueBlock, err = w.makeAddPointDecisionV3(key)
+		setHasSameKeyPrefix, writeToValueBlock, err = w.makeAddPointDecisionV3(key, len(value))
 		addPrefixToValueStoredWithKey = base.TrailerKind(key.Trailer) == InternalKeyKindSet
 	} else {
 		err = w.makeAddPointDecisionV2(key)
@@ -2128,11 +2133,7 @@ func NewWriter(f writeCloseSyncer, o WriterOptions, extraOpts ...WriterOption) *
 			Format: o.Comparer.FormatKey,
 		},
 	}
-	if o.EnableValueBlocks {
-		if w.tableFormat < TableFormatPebblev3 {
-			w.err = errors.New("value blocks not supported at this table format version")
-			return w
-		}
+	if w.tableFormat == TableFormatPebblev3 {
 		w.shortAttributeExtractor = o.ShortAttributeExtractor
 		w.requiredInPlaceValueBound = o.RequiredInPlaceValueBound
 		w.valueBlockWriter = newValueBlockWriter(
diff --git a/table_cache.go b/table_cache.go
index 88c20c5e..292f2bd8 100644
--- a/table_cache.go
+++ b/table_cache.go
@@ -413,11 +413,19 @@ func (c *tableCacheShard) newIters(
 	if opts != nil {
 		useFilter = manifest.LevelToInt(opts.level) != 6 || opts.UseL6Filters
 	}
+	tableFormat, err := v.reader.TableFormat()
+	if err != nil {
+		return nil, nil, err
+	}
+	var rp sstable.ReaderProvider
+	if tableFormat == sstable.TableFormatPebblev3 && v.reader.Properties.NumValueBlocks > 0 {
+		rp = &tableCacheShardReaderProvider{c: c, file: file, dbOpts: dbOpts}
+	}
 	if internalOpts.bytesIterated != nil {
-		iter, err = v.reader.NewCompactionIter(internalOpts.bytesIterated)
+		iter, err = v.reader.NewCompactionIter(internalOpts.bytesIterated, rp)
 	} else {
 		iter, err = v.reader.NewIterWithBlockPropertyFilters(
-			opts.GetLowerBound(), opts.GetUpperBound(), filterer, useFilter, internalOpts.stats)
+			opts.GetLowerBound(), opts.GetUpperBound(), filterer, useFilter, internalOpts.stats, rp)
 	}
 	if err != nil {
 		if rangeDelIter != nil {
@@ -427,7 +435,7 @@ func (c *tableCacheShard) newIters(
 		return nil, nil, err
 	}
 	// NB: v.closeHook takes responsibility for calling unrefValue(v) here. Take
-	// care to avoid introduceingan allocation here by adding a closure.
+	// care to avoid introducing an allocation here by adding a closure.
 	iter.SetCloseHook(v.closeHook)
 
 	atomic.AddInt32(&c.atomic.iterCount, 1)
@@ -488,6 +496,41 @@ func (c *tableCacheShard) newRangeKeyIter(
 	return iter, nil
 }
 
+type tableCacheShardReaderProvider struct {
+	c      *tableCacheShard
+	file   *manifest.FileMetadata
+	dbOpts *tableCacheOpts
+	v      *tableCacheValue
+}
+
+var _ sstable.ReaderProvider = &tableCacheShardReaderProvider{}
+
+// GetReader implements sstable.ReaderProvider. Note that it is not the
+// responsibility of tableCacheShardReaderProvider to ensure that the file
+// continues to exist. The ReaderProvider is used in iterators where the
+// top-level iterator is pinning the read state and preventing the files from
+// being deleted.
+//
+// The caller must call tableCacheShardReaderProvider.Close.
+func (rp *tableCacheShardReaderProvider) GetReader() (*sstable.Reader, error) {
+	// Calling findNode gives us the responsibility of decrementing v's
+	// refCount.
+	v := rp.c.findNode(rp.file, rp.dbOpts)
+	if v.err != nil {
+		defer rp.c.unrefValue(v)
+		base.MustExist(rp.dbOpts.fs, v.filename, rp.dbOpts.logger, v.err)
+		return nil, v.err
+	}
+	rp.v = v
+	return v.reader, nil
+}
+
+// Close implements sstable.ReaderProvider.
+func (rp *tableCacheShardReaderProvider) Close() {
+	rp.c.unrefValue(rp.v)
+	rp.v = nil
+}
+
 // getTableProperties return sst table properties for target file
 func (c *tableCacheShard) getTableProperties(
 	file *fileMetadata, dbOpts *tableCacheOpts,


Test patch:
diff --git a/data_test.go b/data_test.go
index 20fb57c3..97382fa1 100644
--- a/data_test.go
+++ b/data_test.go
@@ -590,21 +590,30 @@ func runBuildCmd(td *datadriven.TestData, d *DB, fs vfs.FS) error {
 		return err
 	}
 
-	if len(td.CmdArgs) != 1 {
+	if len(td.CmdArgs) < 1 {
 		return errors.New("build <path>: argument missing")
 	}
 	path := td.CmdArgs[0].String()
 
 	// Override table format, if provided.
 	tableFormat := d.opts.FormatMajorVersion.MaxTableFormat()
-	for _, cmdArg := range td.CmdArgs {
+	for _, cmdArg := range td.CmdArgs[1:] {
 		switch cmdArg.Key {
 		case "format":
-			v, err := strconv.Atoi(cmdArg.Vals[0])
-			if err != nil {
-				return err
+			switch cmdArg.Vals[0] {
+			case "leveldb":
+				tableFormat = sstable.TableFormatLevelDB
+			case "rocksdbv2":
+				tableFormat = sstable.TableFormatRocksDBv2
+			case "pebblev1":
+				tableFormat = sstable.TableFormatPebblev1
+			case "pebblev2":
+				tableFormat = sstable.TableFormatPebblev2
+			case "pebblev3":
+				tableFormat = sstable.TableFormatPebblev3
+			default:
+				return errors.Errorf("unknown format string %s", cmdArg.Vals[0])
 			}
-			tableFormat = sstable.TableFormat(v)
 		}
 	}
 
diff --git a/external_iterator_test.go b/external_iterator_test.go
index e72370e1..967abbe4 100644
--- a/external_iterator_test.go
+++ b/external_iterator_test.go
@@ -241,7 +241,9 @@ func TestIterRandomizedMaybeFilteredKeys(t *testing.T) {
 			require.NoError(t, err)
 
 			var iter sstable.Iterator
-			iter, err = r.NewIterWithBlockPropertyFilters(nil, nil, filterer, false /* useFilterBlock */, nil /* stats */)
+			iter, err = r.NewIterWithBlockPropertyFilters(
+				nil, nil, filterer, false /* useFilterBlock */, nil, /* stats */
+				sstable.TrivialReaderProvider{Reader: r})
 			require.NoError(t, err)
 			defer iter.Close()
 			var lastSeekKey, lowerBound, upperBound []byte
diff --git a/format_major_version_test.go b/format_major_version_test.go
index 92865949..dba2a828 100644
--- a/format_major_version_test.go
+++ b/format_major_version_test.go
@@ -415,6 +415,8 @@ func TestPebblev1Migration(t *testing.T) {
 				if err := runBuildCmd(td, d, d.opts.FS); err != nil {
 					return err.Error()
 				}
+				// Only the first arg is a filename.
+				td.CmdArgs = td.CmdArgs[:1]
 				if err := runIngestCmd(td, d, d.opts.FS); err != nil {
 					return err.Error()
 				}
diff --git a/internal/base/lazy_value_test.go b/internal/base/lazy_value_test.go
index 89b694f4..21105ad8 100644
--- a/internal/base/lazy_value_test.go
+++ b/internal/base/lazy_value_test.go
@@ -23,7 +23,7 @@ func TestLazyValue(t *testing.T) {
 	require.Equal(t, 3, fooLV1.Len())
 	_, hasAttr := fooLV1.TryGetShortAttribute()
 	require.False(t, hasAttr)
-	fooLV2, fooBytes2 := fooLV1.Clone(nil)
+	fooLV2, fooBytes2 := fooLV1.Clone(nil, &LazyFetcher{})
 	require.Equal(t, 3, fooLV2.Len())
 	_, hasAttr = fooLV2.TryGetShortAttribute()
 	require.False(t, hasAttr)
@@ -43,9 +43,10 @@ func TestLazyValue(t *testing.T) {
 		fooLV3 := LazyValue{
 			ValueOrHandle: []byte("foo-handle"),
 			Fetcher: &LazyFetcher{
-				ValueFetcher: func(handle []byte, buf []byte) ([]byte, bool, error) {
+				ValueFetcher: func(handle []byte, valLen int32, buf []byte) ([]byte, bool, error) {
 					numCalls++
 					require.Equal(t, []byte("foo-handle"), handle)
+					require.Equal(t, int32(3), valLen)
 					return fooBytes1, callerOwned, nil
 				},
 				Attribute: AttributeAndLen{ValueLen: 3, ShortAttribute: 7},
diff --git a/level_iter_test.go b/level_iter_test.go
index d5b31fb6..50e1e1b6 100644
--- a/level_iter_test.go
+++ b/level_iter_test.go
@@ -159,7 +159,9 @@ func (lt *levelIterTest) newIters(
 	file *manifest.FileMetadata, opts *IterOptions, iio internalIterOpts,
 ) (internalIterator, keyspan.FragmentIterator, error) {
 	lt.itersCreated++
-	iter, err := lt.readers[file.FileNum].NewIterWithBlockPropertyFilters(opts.LowerBound, opts.UpperBound, nil, true, iio.stats)
+	iter, err := lt.readers[file.FileNum].NewIterWithBlockPropertyFilters(
+		opts.LowerBound, opts.UpperBound, nil, true, iio.stats,
+		sstable.TrivialReaderProvider{Reader: lt.readers[file.FileNum]})
 	if err != nil {
 		return nil, nil, err
 	}
diff --git a/merging_iter_test.go b/merging_iter_test.go
index 19614d8e..ecf57742 100644
--- a/merging_iter_test.go
+++ b/merging_iter_test.go
@@ -157,7 +157,9 @@ func TestMergingIterCornerCases(t *testing.T) {
 			if err != nil {
 				return nil, nil, err
 			}
-			iter, err := r.NewIterWithBlockPropertyFilters(opts.GetLowerBound(), opts.GetUpperBound(), nil, true /* useFilterBlock */, iio.stats)
+			iter, err := r.NewIterWithBlockPropertyFilters(
+				opts.GetLowerBound(), opts.GetUpperBound(), nil, true /* useFilterBlock */, iio.stats,
+				sstable.TrivialReaderProvider{Reader: r})
 			if err != nil {
 				return nil, nil, err
 			}
diff --git a/options_test.go b/options_test.go
index 5915856e..a812cb7c 100644
--- a/options_test.go
+++ b/options_test.go
@@ -234,6 +234,7 @@ func TestOptionsParse(t *testing.T) {
 			opts.Experimental.TableCacheShards = 500
 			opts.Experimental.MaxWriterConcurrency = 1
 			opts.Experimental.ForceWriterParallelism = true
+			opts.Experimental.EnableValueBlocks = true
 			opts.EnsureDefaults()
 			str := opts.String()
 
diff --git a/sstable/block_property_test.go b/sstable/block_property_test.go
index 6ce404ba..54890b9f 100644
--- a/sstable/block_property_test.go
+++ b/sstable/block_property_test.go
@@ -1007,11 +1007,13 @@ func TestBlockProperties(t *testing.T) {
 			} else if !ok {
 				return "filter excludes entire table"
 			}
-			iter, err := r.NewIterWithBlockPropertyFilters(lower, upper, filterer, false /* use (bloom) filter */, &stats)
+			iter, err := r.NewIterWithBlockPropertyFilters(
+				lower, upper, filterer, false /* use (bloom) filter */, &stats,
+				TrivialReaderProvider{Reader: r})
 			if err != nil {
 				return err.Error()
 			}
-			return runIterCmd(td, iter, runIterCmdEveryOpAfter(func(w io.Writer) {
+			return runIterCmd(td, iter, false, runIterCmdEveryOpAfter(func(w io.Writer) {
 				// After every op, point the value of MaybeFilteredKeys.
 				fmt.Fprintf(w, " MaybeFilteredKeys()=%t", iter.MaybeFilteredKeys())
 			}))
@@ -1085,11 +1087,13 @@ func TestBlockProperties_BoundLimited(t *testing.T) {
 			} else if !ok {
 				return "filter excludes entire table"
 			}
-			iter, err := r.NewIterWithBlockPropertyFilters(lower, upper, filterer, false /* use (bloom) filter */, &stats)
+			iter, err := r.NewIterWithBlockPropertyFilters(
+				lower, upper, filterer, false /* use (bloom) filter */, &stats,
+				TrivialReaderProvider{Reader: r})
 			if err != nil {
 				return err.Error()
 			}
-			return runIterCmd(td, iter, runIterCmdEveryOp(func(w io.Writer) {
+			return runIterCmd(td, iter, false, runIterCmdEveryOp(func(w io.Writer) {
 				// Copy the bound-limited-wrapper's accumulated output to the
 				// iterator's writer. This interleaves its output with the
 				// iterator output.
diff --git a/sstable/data_test.go b/sstable/data_test.go
index bc81d07f..c6c9e4df 100644
--- a/sstable/data_test.go
+++ b/sstable/data_test.go
@@ -264,7 +264,9 @@ func runIterCmdStats(stats *base.InternalIteratorStats) runIterCmdOption {
 	return func(opts *runIterCmdOptions) { opts.stats = stats }
 }
 
-func runIterCmd(td *datadriven.TestData, origIter Iterator, opt ...runIterCmdOption) string {
+func runIterCmd(
+	td *datadriven.TestData, origIter Iterator, printValue bool, opt ...runIterCmdOption,
+) string {
 	var opts runIterCmdOptions
 	for _, o := range opt {
 		o(&opts)
@@ -363,6 +365,9 @@ func runIterCmd(td *datadriven.TestData, origIter Iterator, opt ...runIterCmdOpt
 		}
 		if iter.Valid() && checkValidPrefix(prefix, iter.Key().UserKey) {
 			fmt.Fprintf(&b, "<%s:%d>", iter.Key().UserKey, iter.Key().SeqNum())
+			if printValue {
+				fmt.Fprintf(&b, ":%s", string(iter.Value()))
+			}
 		} else if err := iter.Error(); err != nil {
 			fmt.Fprintf(&b, "<err=%v>", err)
 		} else {
diff --git a/sstable/reader_test.go b/sstable/reader_test.go
index c1a4bcf6..02c2f59e 100644
--- a/sstable/reader_test.go
+++ b/sstable/reader_test.go
@@ -194,7 +194,7 @@ func TestReader(t *testing.T) {
 	}
 
 	opts := map[string]*Comparer{
-		"default":      nil,
+		"default":      testkeys.Comparer,
 		"prefixFilter": fixtureComparer,
 	}
 
@@ -203,21 +203,24 @@ func TestReader(t *testing.T) {
 		"prefixFilter": "testdata/prefixreader",
 	}
 
-	for dName, blockSize := range blockSizes {
-		for iName, indexBlockSize := range blockSizes {
-			for lName, tableOpt := range writerOpts {
-				for oName, cmp := range opts {
-					tableOpt.BlockSize = blockSize
-					tableOpt.Comparer = cmp
-					tableOpt.IndexBlockSize = indexBlockSize
-
-					t.Run(
-						fmt.Sprintf("opts=%s,writerOpts=%s,blockSize=%s,indexSize=%s",
-							oName, lName, dName, iName),
-						func(t *testing.T) {
-							runTestReader(
-								t, tableOpt, testDirs[oName], nil /* Reader */, 0, false)
-						})
+	for _, format := range []TableFormat{TableFormatPebblev2, TableFormatPebblev3} {
+		for dName, blockSize := range blockSizes {
+			for iName, indexBlockSize := range blockSizes {
+				for lName, tableOpt := range writerOpts {
+					for oName, cmp := range opts {
+						tableOpt.BlockSize = blockSize
+						tableOpt.Comparer = cmp
+						tableOpt.IndexBlockSize = indexBlockSize
+						tableOpt.TableFormat = format
+
+						t.Run(
+							fmt.Sprintf("format=%d,opts=%s,writerOpts=%s,blockSize=%s,indexSize=%s",
+								format, oName, lName, dName, iName),
+							func(t *testing.T) {
+								runTestReader(
+									t, tableOpt, testDirs[oName], nil /* Reader */, 0, false, true)
+							})
+					}
 				}
 			}
 		}
@@ -245,7 +248,7 @@ func TestHamletReader(t *testing.T) {
 		t.Run(
 			fmt.Sprintf("sst=%s", prebuiltSST),
 			func(t *testing.T) {
-				runTestReader(t, WriterOptions{}, "testdata/hamletreader", r, 0, false)
+				runTestReader(t, WriterOptions{}, "testdata/hamletreader", r, 0, false, false)
 			},
 		)
 	}
@@ -256,19 +259,24 @@ func TestReaderStats(t *testing.T) {
 		BlockSize:      30,
 		IndexBlockSize: 30,
 	}
-	runTestReader(t, tableOpt, "testdata/readerstats", nil, 10000, false)
+	runTestReader(t, tableOpt, "testdata/readerstats", nil, 10000, false, false)
 }
 
 func TestReaderWithBlockPropertyFilter(t *testing.T) {
-	writerOpt := WriterOptions{
-		BlockSize:               1,
-		IndexBlockSize:          40,
-		Comparer:                testkeys.Comparer,
-		TableFormat:             TableFormatMax,
-		BlockPropertyCollectors: []func() BlockPropertyCollector{NewTestKeysBlockPropertyCollector},
+	for _, format := range []TableFormat{TableFormatPebblev2, TableFormatPebblev3} {
+		writerOpt := WriterOptions{
+			BlockSize:               1,
+			IndexBlockSize:          40,
+			Comparer:                testkeys.Comparer,
+			TableFormat:             format,
+			BlockPropertyCollectors: []func() BlockPropertyCollector{NewTestKeysBlockPropertyCollector},
+		}
+		tdFile := "testdata/reader_bpf"
+		if format == TableFormatPebblev3 {
+			tdFile = "testdata/reader_bpf_v3"
+		}
+		runTestReader(t, writerOpt, tdFile, nil /* Reader */, 0, true, false)
 	}
-	runTestReader(
-		t, writerOpt, "testdata/reader_bpf", nil /* Reader */, 0, true)
 }
 
 func TestInjectedErrors(t *testing.T) {
@@ -387,7 +395,13 @@ func indexLayoutString(t *testing.T, r *Reader) string {
 }
 
 func runTestReader(
-	t *testing.T, o WriterOptions, dir string, r *Reader, cacheSize int, printLayout bool,
+	t *testing.T,
+	o WriterOptions,
+	dir string,
+	r *Reader,
+	cacheSize int,
+	printLayout bool,
+	printValue bool,
 ) {
 	datadriven.Walk(t, dir, func(t *testing.T, path string) {
 		defer func() {
@@ -442,11 +456,12 @@ func runTestReader(
 					filterer,
 					true, /* use filter block */
 					&stats,
+					TrivialReaderProvider{Reader: r},
 				)
 				if err != nil {
 					return err.Error()
 				}
-				return runIterCmd(d, iter, runIterCmdStats(&stats))
+				return runIterCmd(d, iter, printValue, runIterCmdStats(&stats))
 
 			case "get":
 				var b bytes.Buffer
@@ -576,7 +591,7 @@ func testBytesIteratedWithCompression(
 			for _, numEntries := range []uint64{0, 1, maxNumEntries[i]} {
 				r := buildTestTable(t, numEntries, blockSize, indexBlockSize, compression)
 				var bytesIterated, prevIterated uint64
-				citer, err := r.NewCompactionIter(&bytesIterated)
+				citer, err := r.NewCompactionIter(&bytesIterated, TrivialReaderProvider{Reader: r})
 				require.NoError(t, err)
 
 				for key, _ := citer.First(); key != nil; key, _ = citer.Next() {
@@ -626,7 +641,7 @@ func TestCompactionIteratorSetupForCompaction(t *testing.T) {
 			for _, numEntries := range []uint64{0, 1, 1e5} {
 				r := buildTestTable(t, numEntries, blockSize, indexBlockSize, DefaultCompression)
 				var bytesIterated uint64
-				citer, err := r.NewCompactionIter(&bytesIterated)
+				citer, err := r.NewCompactionIter(&bytesIterated, TrivialReaderProvider{Reader: r})
 				require.NoError(t, err)
 				switch i := citer.(type) {
 				case *compactionIterator:
@@ -1098,6 +1113,7 @@ var basicBenchmarks = []struct {
 			BlockRestartInterval: 16,
 			FilterPolicy:         nil,
 			Compression:          SnappyCompression,
+			TableFormat:          TableFormatPebblev2,
 		},
 	},
 	{
@@ -1107,6 +1123,7 @@ var basicBenchmarks = []struct {
 			BlockRestartInterval: 16,
 			FilterPolicy:         nil,
 			Compression:          ZstdCompression,
+			TableFormat:          TableFormatPebblev2,
 		},
 	},
 }
diff --git a/sstable/suffix_rewriter_test.go b/sstable/suffix_rewriter_test.go
index 0258d8a4..1e8ac676 100644
--- a/sstable/suffix_rewriter_test.go
+++ b/sstable/suffix_rewriter_test.go
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"encoding/binary"
 	"fmt"
+	"math/rand"
 	"strconv"
 	"testing"
 
@@ -15,6 +16,11 @@ import (
 func TestRewriteSuffixProps(t *testing.T) {
 	from, to := []byte("_212"), []byte("_646")
 
+	format := TableFormatPebblev2
+	if rand.Intn(2) != 0 {
+		format = TableFormatPebblev3
+	}
+	t.Logf("table format: %s\n", format.String())
 	wOpts := WriterOptions{
 		FilterPolicy: bloom.FilterPolicy(10),
 		Comparer:     test4bSuffixComparer,
@@ -27,7 +33,7 @@ func TestRewriteSuffixProps(t *testing.T) {
 			intSuffixIntervalCollectorFn("bp2", 2),
 			intSuffixIntervalCollectorFn("bp1", 1),
 		},
-		TableFormat: TableFormatPebblev2,
+		TableFormat: format,
 	}
 
 	const keyCount = 1e5
diff --git a/sstable/testdata/prefixreader/iter b/sstable/testdata/prefixreader/iter
index 44a01c9f..d1b8ac8b 100644
--- a/sstable/testdata/prefixreader/iter
+++ b/sstable/testdata/prefixreader/iter
@@ -10,29 +10,29 @@ seek-prefix-ge a
 next
 next
 ----
-<a:1>
-<aa:2>
+<a:1>:A
+<aa:2>:AA
 .
 
 iter
 seek-prefix-ge aa
 next
 ----
-<aa:2>
+<aa:2>:AA
 .
 
 iter
 seek-prefix-ge aa
 prev
 ----
-<aa:2>
+<aa:2>:AA
 .
 
 iter
 seek-prefix-ge c
 prev
 ----
-<c:3>
+<c:3>:C
 .
 
 iter
@@ -44,14 +44,14 @@ iter
 seek-prefix-ge c
 next
 ----
-<c:3>
+<c:3>:C
 .
 
 iter
 seek-prefix-ge d
 next
 ----
-<d:4>
+<d:4>:D
 .
 
 iter
@@ -64,8 +64,8 @@ seek-prefix-ge c
 seek-prefix-ge d
 seek-prefix-ge e
 ----
-<c:3>
-<d:4>
+<c:3>:C
+<d:4>:D
 .
 
 iter
@@ -75,10 +75,10 @@ seek-prefix-ge a
 next
 next
 ----
-<c:3>
+<c:3>:C
 .
-<a:1>
-<aa:2>
+<a:1>:A
+<aa:2>:AA
 .
 
 iter
@@ -90,12 +90,12 @@ next
 seek-prefix-ge c
 next
 ----
-<aa:2>
+<aa:2>:AA
 .
-<a:1>
-<aa:2>
+<a:1>:A
+<aa:2>:AA
 .
-<c:3>
+<c:3>:C
 .
 
 iter
@@ -107,12 +107,12 @@ seek-prefix-ge a
 next
 next
 ----
-<c:3>
+<c:3>:C
 .
-<aa:2>
+<aa:2>:AA
 .
-<a:1>
-<aa:2>
+<a:1>:A
+<aa:2>:AA
 .
 
 iter
@@ -120,8 +120,8 @@ seek-prefix-ge a
 next
 next
 ----
-<a:1>
-<aa:2>
+<a:1>:A
+<aa:2>:AA
 .
 
 iter
@@ -130,16 +130,16 @@ next
 prev
 prev
 ----
-<a:1>
-<aa:2>
-<a:1>
+<a:1>:A
+<aa:2>:AA
+<a:1>:A
 .
 
 iter
 seek-prefix-ge a
 prev
 ----
-<a:1>
+<a:1>:A
 .
 
 iter
@@ -150,11 +150,11 @@ next
 next
 next
 ----
-<a:1>
-<a:1>
-<aa:2>
-<c:3>
-<d:4>
+<a:1>:A
+<a:1>:A
+<aa:2>:AA
+<c:3>:C
+<d:4>:D
 .
 
 iter
@@ -164,10 +164,10 @@ next
 next
 next
 ----
-<a:1>
-<aa:2>
-<c:3>
-<d:4>
+<a:1>:A
+<aa:2>:AA
+<c:3>:C
+<d:4>:D
 .
 
 iter
@@ -176,9 +176,9 @@ seek-ge c
 next
 next
 ----
-<aa:2>
-<c:3>
-<d:4>
+<aa:2>:AA
+<c:3>:C
+<d:4>:D
 .
 
 iter
@@ -188,10 +188,10 @@ next
 next
 next
 ----
-<aa:2>
-<aa:2>
-<c:3>
-<d:4>
+<aa:2>:AA
+<aa:2>:AA
+<c:3>:C
+<d:4>:D
 .
 
 iter
@@ -200,9 +200,9 @@ seek-lt c
 prev
 prev
 ----
-<aa:2>
-<aa:2>
-<a:1>
+<aa:2>:AA
+<aa:2>:AA
+<a:1>:A
 .
 
 iter
@@ -210,8 +210,8 @@ seek-lt c
 seek-prefix-ge aa
 prev
 ----
-<aa:2>
-<aa:2>
+<aa:2>:AA
+<aa:2>:AA
 .
 
 iter
@@ -221,9 +221,9 @@ next
 next
 
 ----
-<aa:2>
-<a:1>
-<aa:2>
+<aa:2>:AA
+<a:1>:A
+<aa:2>:AA
 .
 
 iter
@@ -233,9 +233,9 @@ next
 next
 
 ----
-<aa:2>
-<a:1>
-<aa:2>
+<aa:2>:AA
+<a:1>:A
+<aa:2>:AA
 .
 
 iter
@@ -263,8 +263,8 @@ seek-prefix-ge aa true
 seek-prefix-ge d true
 seek-prefix-ge c false
 ----
-<a:1>
-<a:1>
-<aa:2>
-<d:4>
-<c:3>
+<a:1>:A
+<a:1>:A
+<aa:2>:AA
+<d:4>:D
+<c:3>:C
diff --git a/sstable/testdata/reader/iter b/sstable/testdata/reader/iter
index 6ecfe82d..b06939cc 100644
--- a/sstable/testdata/reader/iter
+++ b/sstable/testdata/reader/iter
@@ -12,10 +12,10 @@ next
 next
 next
 ----
-<a:1>
-<b:2>
-<c:3>
-<d:4>
+<a:1>:A
+<b:2>:B
+<c:3>:C
+<d:4>:D
 .
 
 iter
@@ -25,10 +25,10 @@ next
 next
 next
 ----
-<a:1>
-<b:2>
-<c:3>
-<d:4>
+<a:1>:A
+<b:2>:B
+<c:3>:C
+<d:4>:D
 .
 
 iter
@@ -37,9 +37,9 @@ next
 next
 next
 ----
-<b:2>
-<c:3>
-<d:4>
+<b:2>:B
+<c:3>:C
+<d:4>:D
 .
 
 iter
@@ -47,15 +47,15 @@ seek-ge c
 next
 next
 ----
-<c:3>
-<d:4>
+<c:3>:C
+<d:4>:D
 .
 
 iter
 seek-ge d
 next
 ----
-<d:4>
+<d:4>:D
 .
 
 iter
@@ -67,7 +67,7 @@ iter
 seek-ge d
 seek-ge z
 ----
-<d:4>
+<d:4>:D
 .
 
 iter
@@ -76,9 +76,9 @@ seek-ge c
 seek-ge d
 seek-ge e
 ----
-<b:2>
-<c:3>
-<d:4>
+<b:2>:B
+<c:3>:C
+<d:4>:D
 .
 
 iter
@@ -88,10 +88,10 @@ prev
 prev
 prev
 ----
-<d:4>
-<c:3>
-<b:2>
-<a:1>
+<d:4>:D
+<c:3>:C
+<b:2>:B
+<a:1>:A
 .
 
 iter
@@ -101,10 +101,10 @@ prev
 prev
 prev
 ----
-<d:4>
-<c:3>
-<b:2>
-<a:1>
+<d:4>:D
+<c:3>:C
+<b:2>:B
+<a:1>:A
 .
 
 iter
@@ -113,9 +113,9 @@ prev
 prev
 prev
 ----
-<c:3>
-<b:2>
-<a:1>
+<c:3>:C
+<b:2>:B
+<a:1>:A
 .
 
 iter
@@ -123,15 +123,15 @@ seek-lt c
 prev
 prev
 ----
-<b:2>
-<a:1>
+<b:2>:B
+<a:1>:A
 .
 
 iter
 seek-lt b
 prev
 ----
-<a:1>
+<a:1>:A
 .
 
 iter
@@ -145,9 +145,9 @@ seek-lt c
 seek-lt b
 seek-lt a
 ----
-<c:3>
-<b:2>
-<a:1>
+<c:3>:C
+<b:2>:B
+<a:1>:A
 .
 
 iter globalSeqNum=1
@@ -157,10 +157,10 @@ next
 next
 next
 ----
-<a:1>
-<b:1>
-<c:1>
-<d:1>
+<a:1>:A
+<b:1>:B
+<c:1>:C
+<d:1>:D
 .
 
 iter globalSeqNum=10
@@ -170,16 +170,16 @@ next
 next
 next
 ----
-<a:10>
-<b:10>
-<c:10>
-<d:10>
+<a:10>:A
+<b:10>:B
+<c:10>:C
+<d:10>:D
 .
 
 iter globalSeqNum=0
 seek-lt x
 ----
-<d:4>
+<d:4>:D
 
 get
 b
@@ -207,14 +207,14 @@ set-bounds lower= upper=
 seek-ge c
 seek-lt b
 ----
-<c:3>
-<a:1>
+<c:3>:C
+<a:1>:A
 .
 .
 .
 .
-<c:3>
-<a:1>
+<c:3>:C
+<a:1>:A
 
 # Verify that seeking past the end of the sstable leaves the iterator
 # in a state where prev returns the last key in the table.
@@ -224,9 +224,9 @@ seek-lt d
 seek-ge f
 prev
 ----
-<c:3>
+<c:3>:C
 .
-<d:4>
+<d:4>:D
 
 # Verify that seeking before the beginning of the sstable leaves the
 # iterator in a state where next returns the first key in the table.
@@ -236,9 +236,9 @@ seek-ge b
 seek-lt a
 next
 ----
-<b:2>
+<b:2>:B
 .
-<a:1>
+<a:1>:A
 
 
 # Verify the optimization to use next when doing SeekGE.
@@ -251,11 +251,11 @@ seek-ge c true
 seek-ge d true
 seek-ge e true
 ----
-<a:1>
-<a:1>
-<b:2>
-<c:3>
-<d:4>
+<a:1>:A
+<a:1>:A
+<b:2>:B
+<c:3>:C
+<d:4>:D
 .
 
 # Verify the optimization to use next when doing SeekPrefixGE.
@@ -268,11 +268,11 @@ seek-prefix-ge c true
 seek-prefix-ge d true
 seek-prefix-ge e true
 ----
-<a:1>
-<a:1>
-<b:2>
-<c:3>
-<d:4>
+<a:1>:A
+<a:1>:A
+<b:2>:B
+<c:3>:C
+<d:4>:D
 .
 
 # Verify that iteration from before the beginning or after the end of
@@ -291,9 +291,9 @@ next
 next
 next
 ----
-<a:1>
+<a:1>:a
 .
-<a:1>
+<a:1>:a
 .
 .
 
@@ -304,9 +304,9 @@ prev
 prev
 prev
 ----
-<a:1>
+<a:1>:a
 .
-<a:1>
+<a:1>:a
 .
 .
 
@@ -418,13 +418,13 @@ next
 next
 next
 ----
-<a:1>
+<a:1>:A
 .
-<a:1>
-<aae:1>
-<aaf:1>
-<aag:1>
-<aah:1>
+<a:1>:A
+<aae:1>:E
+<aaf:1>:F
+<aag:1>:G
+<aah:1>:H
 
 iter
 last
@@ -433,11 +433,11 @@ prev
 prev
 prev
 ----
-<ddz:4>
+<ddz:4>:Z
 .
-<ddz:4>
-<ddy:4>
-<ddx:4>
+<ddz:4>:Z
+<ddy:4>:Y
+<ddx:4>:X
 
 iter
 first
@@ -448,13 +448,13 @@ seek-ge x
 prev
 prev
 ----
-<a:1>
+<a:1>:A
 .
-<a:1>
-<aae:1>
+<a:1>:A
+<aae:1>:E
 .
-<ddz:4>
-<ddy:4>
+<ddz:4>:Z
+<ddy:4>:Y
 
 iter
 first
@@ -465,10 +465,10 @@ seek-prefix-ge x
 prev
 prev
 ----
-<a:1>
+<a:1>:A
 .
-<a:1>
-<aae:1>
+<a:1>:A
+<aae:1>:E
 .
 .
 .
@@ -482,13 +482,13 @@ seek-lt a
 next
 next
 ----
-<ddz:4>
+<ddz:4>:Z
 .
-<ddz:4>
-<ddy:4>
+<ddz:4>:Z
+<ddy:4>:Y
 .
-<a:1>
-<aae:1>
+<a:1>:A
+<aae:1>:E
 
 # Test that SeekPrefixGE does not position the iterator far outside the iterator bounds.
 # Doing so would break the subsequent SeekGE that is utilizing the next instead of seek
@@ -502,11 +502,11 @@ seek-ge aae
 next
 ----
 .
-<a:1>
+<a:1>:A
 .
 .
-<aae:1>
-<aaf:1>
+<aae:1>:E
+<aaf:1>:F
 
 # Test that using Next does not mislead a twoLevelIterator into believing that the
 # iterator has been positioned based on the latest iterator bounds. The Next call
@@ -522,9 +522,52 @@ seek-ge bbf
 next
 ----
 .
-<bbq:2>
+<bbq:2>:Q
 .
 .
 .
-<bbf:2>
-<bbg:2>
+<bbf:2>:F
+<bbg:2>:G
+
+build
+a@10.SET.10:a10
+a@5.SET.5:a5
+b@20.SET.20:b20
+b@17.SET.17:b17
+----
+
+iter
+first
+next
+next
+next
+next
+----
+<a@10:10>:a10
+<a@5:5>:a5
+<b@20:20>:b20
+<b@17:17>:b17
+.
+
+iter
+seek-ge a@5
+prev
+seek-lt b
+next
+next
+seek-lt c
+prev
+seek-ge b@18
+prev
+next
+----
+<a@5:5>:a5
+<a@10:10>:a10
+<a@5:5>:a5
+<b@20:20>:b20
+<b@17:17>:b17
+<b@17:17>:b17
+<b@20:20>:b20
+<b@17:17>:b17
+<b@20:20>:b20
+<b@17:17>:b17
diff --git a/sstable/testdata/reader_bpf_v3/iter b/sstable/testdata/reader_bpf_v3/iter
new file mode 100644
index 00000000..14dabbed
--- /dev/null
+++ b/sstable/testdata/reader_bpf_v3/iter
@@ -0,0 +1,56 @@
+# Test case for bug https://github.com/cockroachdb/pebble/issues/2036 Build
+# sstable with two-level index, with two data blocks in each lower-level index
+# block.
+build
+c@10.SET.10:cAT10
+d@7.SET.9:dAT7
+e@15.SET.8:eAT15
+f@7.SET.5:fAT7
+----
+index entries:
+ d@7: size 53
+   c@10: size 29
+   d@7: size 27
+ g: size 51
+   e@15: size 29
+   g: size 27
+
+iter
+first
+next
+next
+next
+----
+<c@10:10>
+<d@7:9>
+<e@15:8>
+<f@7:5>
+
+
+# The block property filter matches data block 2 and 4.
+iter block-property-filter=(7,8)
+first
+next
+----
+<d@7:9>
+<f@7:5>
+
+# Use the same block property filter, but use seeks to find these entries.
+# With the bug the second seek-ge below would step to the second lower-level
+# index block and only see the entry in the data block 4.
+iter block-property-filter=(7,8)
+set-bounds lower=a upper=c
+seek-ge a
+seek-ge b true
+set-bounds lower=c upper=g
+seek-ge c
+next
+next
+----
+.
+.
+.
+.
+<d@7:9>
+<f@7:5>
+.
diff --git a/sstable/testdata/rewriter_v3 b/sstable/testdata/rewriter_v3
new file mode 100644
index 00000000..bf418884
--- /dev/null
+++ b/sstable/testdata/rewriter_v3
@@ -0,0 +1,237 @@
+build block-size=1 index-block-size=1 filter
+a_xyz.SET.1:a
+b_xyz.SET.1:b
+c_xyz.SET.1:c
+----
+point:    [a_xyz#1,1-c_xyz#1,1]
+seqnums:  [1-1]
+
+rewrite from=xyz to=123 block-size=1 index-block-size=1 filter
+----
+rewrite failed: a valid splitter is required to define suffix to replace replace suffix
+
+rewrite from=xyz to=123 block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+----
+rewrite failed: rewriting data blocks: mismatched Comparer leveldb.BytewiseComparator vs comparer-split-4b-suffix, replacement requires same splitter to copy filters
+
+build block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+aa_xyz.SET.1:a
+ba_xyz.SET.1:b
+ca_xyz.SET.1:c
+----
+point:    [aa_xyz#1,1-ca_xyz#1,1]
+seqnums:  [1-1]
+
+rewrite from=yz to=23 block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+----
+rewrite failed: rewriting data blocks: key has suffix "_xyz", expected "yz"
+
+rewrite from=a_xyz to=a_123 block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+----
+rewrite failed: rewriting data blocks: key has suffix "_xyz", expected "a_xyz"
+
+build block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+a_xyz.SET.1:a
+b_xyz.SET.1:b
+c_xyz.SET.1:c
+----
+point:    [a_xyz#1,1-c_xyz#1,1]
+seqnums:  [1-1]
+
+layout
+----
+         0  data (26)
+        31  data (26)
+        62  data (26)
+        93  filter (69)
+       167  index (22)
+       194  index (22)
+       221  index (22)
+       248  top-index (48)
+       301  properties (767)
+      1073  meta-index (79)
+      1157  footer (53)
+      1210  EOF
+
+scan
+----
+a_xyz#1,1:a
+b_xyz#1,1:b
+c_xyz#1,1:c
+
+get
+b_xyz
+f_xyz
+c_xyz
+----
+b
+get f_xyz: pebble: not found
+c
+
+rewrite from=_xyz to=_123 block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+----
+point:    [a_123#1,1-c_123#1,1]
+seqnums:  [1-1]
+
+layout
+----
+         0  data (26)
+        31  data (26)
+        62  data (26)
+        93  filter (69)
+       167  index (22)
+       194  index (22)
+       221  index (22)
+       248  top-index (48)
+       301  properties (767)
+      1073  meta-index (79)
+      1157  footer (53)
+      1210  EOF
+
+scan
+----
+a_123#1,1:a
+b_123#1,1:b
+c_123#1,1:c
+
+get
+b_123
+f_123
+c_123
+----
+b
+get f_123: pebble: not found
+c
+
+rewrite from=_123 to=_456 block-size=1 index-block-size=1 filter comparer-split-4b-suffix concurrency=2
+----
+point:    [a_456#1,1-c_456#1,1]
+seqnums:  [1-1]
+
+layout
+----
+         0  data (26)
+        31  data (26)
+        62  data (26)
+        93  filter (69)
+       167  index (22)
+       194  index (22)
+       221  index (22)
+       248  top-index (48)
+       301  properties (767)
+      1073  meta-index (79)
+      1157  footer (53)
+      1210  EOF
+
+scan
+----
+a_456#1,1:a
+b_456#1,1:b
+c_456#1,1:c
+
+get
+b_456
+f_456
+c_456
+----
+b
+get f_456: pebble: not found
+c
+
+rewrite from=_456 to=_xyz block-size=1 index-block-size=1 filter comparer-split-4b-suffix concurrency=3
+----
+point:    [a_xyz#1,1-c_xyz#1,1]
+seqnums:  [1-1]
+
+layout
+----
+         0  data (26)
+        31  data (26)
+        62  data (26)
+        93  filter (69)
+       167  index (22)
+       194  index (22)
+       221  index (22)
+       248  top-index (48)
+       301  properties (767)
+      1073  meta-index (79)
+      1157  footer (53)
+      1210  EOF
+
+scan
+----
+a_xyz#1,1:a
+b_xyz#1,1:b
+c_xyz#1,1:c
+
+get
+b_xyz
+f_xyz
+c_xyz
+----
+b
+get f_xyz: pebble: not found
+c
+
+
+rewrite from=_xyz to=_123 block-size=1 index-block-size=1 filter comparer-split-4b-suffix concurrency=4
+----
+point:    [a_123#1,1-c_123#1,1]
+seqnums:  [1-1]
+
+layout
+----
+         0  data (26)
+        31  data (26)
+        62  data (26)
+        93  filter (69)
+       167  index (22)
+       194  index (22)
+       221  index (22)
+       248  top-index (48)
+       301  properties (767)
+      1073  meta-index (79)
+      1157  footer (53)
+      1210  EOF
+
+scan
+----
+a_123#1,1:a
+b_123#1,1:b
+c_123#1,1:c
+
+get
+b_123
+f_123
+c_123
+----
+b
+get f_123: pebble: not found
+c
+
+# Rewrite a table that contain only range keys.
+
+build block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+rangekey: a-b:{(#1,RANGEKEYSET,_xyz)}
+rangekey: b-c:{(#1,RANGEKEYSET,_xyz)}
+rangekey: c-d:{(#1,RANGEKEYSET,_xyz)}
+----
+rangekey: [a#1,21-d#72057594037927935,21]
+seqnums:  [1-1]
+
+scan-range-key
+----
+a-b:{(#1,RANGEKEYSET,_xyz)}
+b-c:{(#1,RANGEKEYSET,_xyz)}
+c-d:{(#1,RANGEKEYSET,_xyz)}
+
+rewrite from=_xyz to=_123 block-size=1 index-block-size=1 filter comparer-split-4b-suffix
+----
+rangekey: [a#1,21-d#72057594037927935,21]
+seqnums:  [1-1]
+
+scan-range-key
+----
+a-b:{(#1,RANGEKEYSET,_123)}
+b-c:{(#1,RANGEKEYSET,_123)}
+c-d:{(#1,RANGEKEYSET,_123)}
diff --git a/sstable/testdata/writer_v3 b/sstable/testdata/writer_v3
new file mode 100644
index 00000000..5981b385
--- /dev/null
+++ b/sstable/testdata/writer_v3
@@ -0,0 +1,343 @@
+build
+a.SET.1:a
+----
+point:    [a#1,1-a#1,1]
+seqnums:  [1-1]
+
+scan
+----
+a#1,1:a
+
+scan-range-del
+----
+
+scan-range-key
+----
+
+build
+a.SET.1:a
+b.DEL.2:
+c.MERGE.3:c
+d.RANGEDEL.4:e
+f.SET.5:f
+g.DEL.6:
+h.MERGE.7:h
+i.RANGEDEL.8:j
+rangekey: j-k:{(#9,RANGEKEYDEL)}
+rangekey: k-l:{(#10,RANGEKEYUNSET,@t5)}
+rangekey: l-m:{(#11,RANGEKEYSET,@t10,foo)}
+----
+point:    [a#1,1-h#7,2]
+rangedel: [d#4,15-j#72057594037927935,15]
+rangekey: [j#9,19-m#72057594037927935,21]
+seqnums:  [1-11]
+
+build
+a.SET.1:a
+b.DEL.2:
+c.MERGE.3:c
+d.RANGEDEL.4:e
+f.SET.5:f
+g.DEL.6:
+h.MERGE.7:h
+i.RANGEDEL.8:j
+----
+point:    [a#1,1-h#7,2]
+rangedel: [d#4,15-j#72057594037927935,15]
+seqnums:  [1-8]
+
+scan
+----
+a#1,1:a
+b#2,0:
+c#3,2:c
+f#5,1:f
+g#6,0:
+h#7,2:h
+
+scan-range-del
+----
+d-e:{(#4,RANGEDEL)}
+i-j:{(#8,RANGEDEL)}
+
+# 3: a-----------m
+# 2:      f------------s
+# 1:          j---------------z
+
+build
+a.RANGEDEL.3:m
+f.RANGEDEL.2:s
+j.RANGEDEL.1:z
+----
+rangedel: [a#3,15-z#72057594037927935,15]
+seqnums:  [1-3]
+
+scan
+----
+
+scan-range-del
+----
+a-f:{(#3,RANGEDEL)}
+f-j:{(#3,RANGEDEL) (#2,RANGEDEL)}
+j-m:{(#3,RANGEDEL) (#2,RANGEDEL) (#1,RANGEDEL)}
+m-s:{(#2,RANGEDEL) (#1,RANGEDEL)}
+s-z:{(#1,RANGEDEL)}
+
+scan-range-key
+----
+
+# The range tombstone upper bound is exclusive, so a point operation
+# on that same key will be the actual boundary.
+
+build
+a.RANGEDEL.3:b
+b.SET.4:c
+----
+point:    [b#4,1-b#4,1]
+rangedel: [a#3,15-b#72057594037927935,15]
+seqnums:  [3-4]
+
+build
+a.RANGEDEL.3:b
+b.SET.2:c
+----
+point:    [b#2,1-b#2,1]
+rangedel: [a#3,15-b#72057594037927935,15]
+seqnums:  [2-3]
+
+build
+a.RANGEDEL.3:c
+b.SET.2:c
+----
+point:    [b#2,1-b#2,1]
+rangedel: [a#3,15-c#72057594037927935,15]
+seqnums:  [2-3]
+
+# Keys must be added in order.
+
+build
+a.SET.1:b
+a.SET.2:c
+----
+pebble: keys must be added in strictly increasing order: a#1,SET, a#2,SET
+
+build
+b.SET.1:a
+a.SET.2:b
+----
+pebble: keys must be added in strictly increasing order: b#1,SET, a#2,SET
+
+build
+b.RANGEDEL.1:c
+a.RANGEDEL.2:b
+----
+pebble: keys must be added in order: b > a
+
+build-raw
+.RANGEDEL.1:b
+----
+rangedel: [#1,15-b#72057594037927935,15]
+seqnums:  [1-1]
+
+build-raw
+a.RANGEDEL.1:c
+a.RANGEDEL.2:c
+----
+pebble: keys must be added in strictly increasing order: a#1,RANGEDEL, a#2,RANGEDEL
+
+build-raw
+a.RANGEDEL.1:c
+b.RANGEDEL.2:d
+----
+pebble: overlapping tombstones must be fragmented: a-c:{(#1,RANGEDEL)} vs b-d:{(#2,RANGEDEL)}
+
+build-raw
+a.RANGEDEL.2:c
+a.RANGEDEL.1:d
+----
+pebble: overlapping tombstones must be fragmented: a-c:{(#2,RANGEDEL)} vs a-d:{(#1,RANGEDEL)}
+
+build-raw
+a.RANGEDEL.1:c
+c.RANGEDEL.2:d
+----
+rangedel: [a#1,15-d#72057594037927935,15]
+seqnums:  [1-2]
+
+build-raw
+rangekey: a-b:{(#1,RANGEKEYSET,@t10,foo)}
+rangekey: a-b:{(#2,RANGEKEYSET,@t10,foo)}
+----
+rangekey: [a#2,21-b#72057594037927935,21]
+seqnums:  [2-2]
+
+build-raw
+rangekey: b-c:{(#2,RANGEKEYSET,@t10,foo)}
+rangekey: a-b:{(#1,RANGEKEYSET,@t10,foo)}
+----
+pebble: spans must be added in order: b > a
+
+build-raw
+a.RANGEKEYDEL.1:c
+b.RANGEKEYDEL.2:d
+----
+pebble: overlapping range keys must be fragmented: a#1,RANGEKEYDEL, b#2,RANGEKEYDEL
+
+build-raw
+a.RANGEKEYDEL.2:c
+a.RANGEKEYDEL.1:d
+----
+pebble: overlapping range keys must be fragmented: a#2,RANGEKEYDEL, a#1,RANGEKEYDEL
+
+build-raw
+rangekey: a-c:{(#1,RANGEKEYSET,@t10,foo)}
+rangekey: c-d:{(#2,RANGEKEYSET,@t10,foo)}
+----
+rangekey: [a#1,21-d#72057594037927935,21]
+seqnums:  [1-2]
+
+# Range keys may have perfectly aligned spans (including sequence numbers),
+# though the key kinds must be ordered (descending).
+
+build-raw
+a.RANGEKEYDEL.1:b
+a.RANGEKEYDEL.1:b
+----
+pebble: range keys starts must be added in increasing order: a#1,RANGEKEYDEL, a#1,RANGEKEYDEL
+
+build-raw
+rangekey: a-b:{(#1,RANGEKEYSET,@t10,foo) (#1,RANGEKEYUNSET,@t10) (#1,RANGEKEYDEL)}
+----
+rangekey: [a#1,21-b#72057594037927935,19]
+seqnums:  [1-1]
+
+# The range-del-v1 format supports unfragmented and unsorted range
+# tombstones.
+
+build-raw range-del-v1
+a.RANGEDEL.1:c
+a.RANGEDEL.2:c
+----
+rangedel: [a#2,15-c#72057594037927935,15]
+seqnums:  [1-2]
+
+scan-range-del
+----
+a-c:{(#2,RANGEDEL) (#1,RANGEDEL)}
+
+build-raw range-del-v1
+a.RANGEDEL.1:c
+b.RANGEDEL.2:d
+----
+rangedel: [a#1,15-d#72057594037927935,15]
+seqnums:  [1-2]
+
+scan-range-del
+----
+a-b:{(#1,RANGEDEL)}
+b-c:{(#2,RANGEDEL) (#1,RANGEDEL)}
+c-d:{(#2,RANGEDEL)}
+
+build-raw range-del-v1
+a.RANGEDEL.2:c
+a.RANGEDEL.1:d
+----
+rangedel: [a#2,15-d#72057594037927935,15]
+seqnums:  [1-2]
+
+scan-range-del
+----
+a-c:{(#2,RANGEDEL) (#1,RANGEDEL)}
+c-d:{(#1,RANGEDEL)}
+
+# This matches an early test case, except we're passing overlapping
+# range tombstones to the sstable writer and requiring them to be
+# fragmented at read time.
+
+build-raw range-del-v1
+j.RANGEDEL.1:z
+f.RANGEDEL.2:s
+a.RANGEDEL.3:m
+----
+rangedel: [a#3,15-z#72057594037927935,15]
+seqnums:  [1-3]
+
+scan-range-del
+----
+a-f:{(#3,RANGEDEL)}
+f-j:{(#3,RANGEDEL) (#2,RANGEDEL)}
+j-m:{(#3,RANGEDEL) (#2,RANGEDEL) (#1,RANGEDEL)}
+m-s:{(#2,RANGEDEL) (#1,RANGEDEL)}
+s-z:{(#1,RANGEDEL)}
+
+# Setting a very small index-block-size results in a two-level index.
+
+build block-size=1 index-block-size=1
+a.SET.1:a
+b.SET.1:b
+c.SET.1:c
+----
+point:    [a#1,1-c#1,1]
+seqnums:  [1-1]
+
+layout
+----
+         0  data (22)
+        27  data (22)
+        54  data (22)
+        81  index (22)
+       108  index (22)
+       135  index (22)
+       162  top-index (51)
+       218  properties (717)
+       940  meta-index (33)
+       978  footer (53)
+      1031  EOF
+
+scan
+----
+a#1,1:a
+b#1,1:b
+c#1,1:c
+
+# Enabling leveldb format disables the creation of a two-level index
+# (the input data here mirrors the test case above).
+
+build leveldb block-size=1 index-block-size=1
+a.SET.1:a
+b.SET.1:b
+c.SET.1:c
+----
+point:    [a#1,1-c#1,1]
+seqnums:  [1-1]
+
+layout
+----
+         0  data (21)
+        26  data (21)
+        52  data (21)
+        78  index (47)
+       130  properties (678)
+       813  meta-index (33)
+       851  leveldb-footer (48)
+       899  EOF
+
+# Range keys, if present, are shown in the layout.
+
+build
+rangekey: a-b:{(#3,RANGEKEYSET,@t3,foo)}
+rangekey: b-c:{(#2,RANGEKEYSET,@t2,bar)}
+rangekey: c-d:{(#1,RANGEKEYSET,@t1,baz)}
+----
+rangekey: [a#3,21-d#72057594037927935,21]
+seqnums:  [1-3]
+
+layout
+----
+         0  data (8)
+        13  index (21)
+        39  range-key (82)
+       126  properties (765)
+       896  meta-index (57)
+       958  footer (53)
+      1011  EOF
diff --git a/sstable/testdata/writer_value_blocks b/sstable/testdata/writer_value_blocks
index e98ce1e1..8ec066e7 100644
--- a/sstable/testdata/writer_value_blocks
+++ b/sstable/testdata/writer_value_blocks
@@ -18,6 +18,22 @@ b@4#3,0:
 b@3#2,1:in-place bat3, same-pre false
 b@2#1,1:value-handle len 5 block 0 offset 0, att 5, same-pre true
 
+scan
+----
+a@2#1,1:a2
+b@5#7,1:b5
+b@4#3,0:
+b@3#2,1:bat3
+b@2#1,1:vbat2
+
+scan-cloned-lazy-values
+----
+0(in-place: len 2): a2
+1(in-place: len 2): b5
+2(in-place: len 0): 
+3(in-place: len 4): bat3
+4(lazy: len 5, attr: 5): vbat2
+
 # Size of value index is 3 bytes plus 5 + 5 = 10 bytes of trailer of the value
 # block and value index block. So size 33 - 13 = 20 is the total size of the
 # values in the value block.
@@ -44,25 +60,159 @@ blue@3#10,1:value-handle len 5 block 0 offset 11, att 5, same-pre true
 red@9#18,1:in-place red9, same-pre false
 red@7#8,1:value-handle len 4 block 0 offset 16, att 4, same-pre true
 
+scan
+----
+blue@10#20,1:blue10
+blue@8#18,1:blue8
+blue@8#16,1:blue8s
+blue@6#14,0:
+blue@4#12,1:blue4
+blue@3#10,1:blue3
+red@9#18,1:red9
+red@7#8,1:red7
+
+scan-cloned-lazy-values
+----
+0(in-place: len 6): blue10
+1(lazy: len 5, attr: 5): blue8
+2(lazy: len 6, attr: 6): blue8s
+3(in-place: len 0): 
+4(in-place: len 5): blue4
+5(lazy: len 5, attr: 5): blue3
+6(in-place: len 4): red9
+7(lazy: len 4, attr: 4): red7
+
 # Multiple value blocks. Trailers of 5+5+5 for the two value blocks and the
-# value index block, totals to 15. The values are 5+6+11=22. The value index
+# value index block, totals to 15. The values are 5+6+15=26. The value index
 # block has to encode two tuples, each of 4 bytes (blockNumByteLength=1,
 # blockOffsetByteLength=2, blockLengthByteLength=1), so 2*4=8. The total is
-# 15+22+8=45 bytes, which corresponds to "size: 45" below.
+# 15+26+8=49 bytes, which corresponds to "size: 49" below.
 build block-size=8
 blue@10.SET.20:blue10
 blue@8.SET.18:blue8
 blue@8.SET.16:blue8s
-blue@6.SET.16:blue6islong
+blue@6.SET.16:blue6isverylong
 ----
-value-blocks: num-values 3, num-blocks: 2, size: 45
+value-blocks: num-values 3, num-blocks: 2, size: 49
 
 scan-raw
 ----
 blue@10#20,1:in-place blue10, same-pre false
 blue@8#18,1:value-handle len 5 block 0 offset 0, att 5, same-pre true
 blue@8#16,1:value-handle len 6 block 0 offset 5, att 6, same-pre true
-blue@6#16,1:value-handle len 11 block 1 offset 0, att 3, same-pre true
+blue@6#16,1:value-handle len 15 block 1 offset 0, att 7, same-pre true
+
+scan
+----
+blue@10#20,1:blue10
+blue@8#18,1:blue8
+blue@8#16,1:blue8s
+blue@6#16,1:blue6isverylong
+
+scan-cloned-lazy-values
+----
+0(in-place: len 6): blue10
+1(lazy: len 5, attr: 5): blue8
+2(lazy: len 6, attr: 6): blue8s
+3(lazy: len 15, attr: 7): blue6isverylong
+
+layout
+----
+         0  data (33)
+         0    record (25 = 3 [0] + 15 + 7) [restart]
+              blue@10#20,1:blue10        25    [restart 0]
+        33    [trailer compression=none checksum=0x5fb0d551]
+        38  data (29)
+        38    record (21 = 3 [0] + 14 + 4) [restart]
+              blue@8#18,1:value handle {valueLen:5 blockNum:0 offsetInBlock:0}        59    [restart 38]
+        67    [trailer compression=none checksum=0x628e4a10]
+        72  data (29)
+        72    record (21 = 3 [0] + 14 + 4) [restart]
+              blue@8#16,1:value handle {valueLen:6 blockNum:0 offsetInBlock:5}        93    [restart 72]
+       101    [trailer compression=none checksum=0xdc74261]
+       106  data (29)
+       106    record (21 = 3 [0] + 14 + 4) [restart]
+              blue@6#16,1:value handle {valueLen:15 blockNum:1 offsetInBlock:0}       127    [restart 106]
+       135    [trailer compression=none checksum=0x9f60e629]
+       140  index (28)
+       140    block:0/33 [restart]
+       160    [restart 140]
+       168    [trailer compression=none checksum=0x32b37f08]
+       173  index (27)
+       173    block:38/29 [restart]
+       192    [restart 173]
+       200    [trailer compression=none checksum=0x21d27815]
+       205  index (27)
+       205    block:72/29 [restart]
+       224    [restart 205]
+       232    [trailer compression=none checksum=0xbae26eb3]
+       237  index (22)
+       237    block:106/29 [restart]
+       251    [restart 237]
+       259    [trailer compression=none checksum=0x802be702]
+       264  top-index (77)
+       264    block:140/28 [restart]
+       285    block:173/27 [restart]
+       305    block:205/27 [restart]
+       325    block:237/22 [restart]
+       340    [restart 264]
+       344    [restart 285]
+       348    [restart 305]
+       352    [restart 325]
+       341    [trailer compression=snappy checksum=0x6b2d79b]
+       346  value-block (11)
+       362  value-block (15)
+       382  value-index (8)
+       395  properties (785)
+       395    pebble.num.value-blocks (27) [restart]
+       422    pebble.num.values.in.value-blocks (21)
+       443    pebble.value-blocks.size (21)
+       464    rocksdb.block.based.table.index.type (43)
+       507    rocksdb.block.based.table.prefix.filtering (20)
+       527    rocksdb.block.based.table.whole.key.filtering (23)
+       550    rocksdb.column.family.id (24)
+       574    rocksdb.comparator (35)
+       609    rocksdb.compression (16)
+       625    rocksdb.compression_options (106)
+       731    rocksdb.creation.time (16)
+       747    rocksdb.data.size (14)
+       761    rocksdb.deleted.keys (15)
+       776    rocksdb.external_sst_file.global_seqno (41)
+       817    rocksdb.external_sst_file.version (14)
+       831    rocksdb.filter.size (15)
+       846    rocksdb.fixed.key.length (18)
+       864    rocksdb.format.version (17)
+       881    rocksdb.index.key.is.user.key (25)
+       906    rocksdb.index.partitions (14)
+       920    rocksdb.index.size (9)
+       929    rocksdb.index.value.is.delta.encoded (26)
+       955    rocksdb.merge.operands (18)
+       973    rocksdb.merge.operator (24)
+       997    rocksdb.num.data.blocks (19)
+      1016    rocksdb.num.entries (11)
+      1027    rocksdb.num.range-deletions (19)
+      1046    rocksdb.oldest.key.time (19)
+      1065    rocksdb.prefix.extractor.name (31)
+      1096    rocksdb.property.collectors (22)
+      1118    rocksdb.raw.key.size (16)
+      1134    rocksdb.raw.value.size (14)
+      1148    rocksdb.top-level.index.size (24)
+      1172    [restart 395]
+      1180    [trailer compression=none checksum=0x4352b7fd]
+      1185  meta-index (64)
+      1185    pebble.value_index block:382/8 value-blocks-index-lengths: 1(num), 2(offset), 1(length) [restart]
+      1212    rocksdb.properties block:395/785 [restart]
+      1237    [restart 1185]
+      1241    [restart 1212]
+      1249    [trailer compression=none checksum=0xe7aed935]
+      1254  footer (53)
+      1254    checksum type: crc32c
+      1255    meta: offset=1185, length=64
+      1258    index: offset=264, length=77
+      1261    [padding]
+      1295    version: 3
+      1299    magic number: 0xf09faab3f09faab3
+      1307  EOF
 
 # Require that [c,e) must be in-place.
 build in-place-bound=(c,e)
@@ -83,3 +233,101 @@ c@10#16,1:in-place c10, same-pre false
 c@8#14,1:in-place c8, same-pre false
 e@20#25,1:in-place eat20, same-pre false
 e@18#23,1:value-handle len 5 block 0 offset 5, att 5, same-pre true
+
+scan
+----
+blue@10#20,1:blue10
+blue@8#18,1:blue8
+c@10#16,1:c10
+c@8#14,1:c8
+e@20#25,1:eat20
+e@18#23,1:eat18
+
+scan-cloned-lazy-values
+----
+0(in-place: len 6): blue10
+1(lazy: len 5, attr: 5): blue8
+2(in-place: len 3): c10
+3(in-place: len 2): c8
+4(in-place: len 5): eat20
+5(lazy: len 5, attr: 5): eat18
+
+# Try write empty values to value blocks.
+build
+b@5.SET.7:b5
+b@3.SET.2:
+c@6.DEL.7:
+c@5.DEL.6:
+----
+value-blocks: num-values 0, num-blocks: 0, size: 0
+
+scan-raw
+----
+b@5#7,1:in-place b5, same-pre false
+b@3#2,1:in-place , same-pre true
+c@6#7,0:
+c@5#6,0:
+
+scan
+----
+b@5#7,1:b5
+b@3#2,1:
+c@6#7,0:
+c@5#6,0:
+
+layout
+----
+         0  data (64)
+         0    record (17 = 3 [0] + 11 + 3) [restart]
+              b@5#7,1:b5        17    record (13 = 3 [2] + 9 + 1)
+              b@3#2,1:        30    record (14 = 3 [0] + 11 + 0)
+              c@6#7,0:        44    record (12 = 3 [2] + 9 + 0)
+              c@5#6,0:        56    [restart 0]
+        64    [trailer compression=none checksum=0xd099737a]
+        69  index (22)
+        69    block:0/64 [restart]
+        83    [restart 69]
+        91    [trailer compression=none checksum=0xf97e414e]
+        96  properties (676)
+        96    rocksdb.block.based.table.index.type (43) [restart]
+       139    rocksdb.block.based.table.prefix.filtering (20)
+       159    rocksdb.block.based.table.whole.key.filtering (23)
+       182    rocksdb.column.family.id (24)
+       206    rocksdb.comparator (35)
+       241    rocksdb.compression (16)
+       257    rocksdb.compression_options (106)
+       363    rocksdb.creation.time (16)
+       379    rocksdb.data.size (13)
+       392    rocksdb.deleted.keys (15)
+       407    rocksdb.external_sst_file.global_seqno (41)
+       448    rocksdb.external_sst_file.version (14)
+       462    rocksdb.filter.size (15)
+       477    rocksdb.fixed.key.length (18)
+       495    rocksdb.format.version (17)
+       512    rocksdb.index.key.is.user.key (25)
+       537    rocksdb.index.size (8)
+       545    rocksdb.index.value.is.delta.encoded (26)
+       571    rocksdb.merge.operands (18)
+       589    rocksdb.merge.operator (24)
+       613    rocksdb.num.data.blocks (19)
+       632    rocksdb.num.entries (11)
+       643    rocksdb.num.range-deletions (19)
+       662    rocksdb.oldest.key.time (19)
+       681    rocksdb.prefix.extractor.name (31)
+       712    rocksdb.property.collectors (22)
+       734    rocksdb.raw.key.size (16)
+       750    rocksdb.raw.value.size (14)
+       764    [restart 96]
+       772    [trailer compression=none checksum=0xf606ecc]
+       777  meta-index (32)
+       777    rocksdb.properties block:96/676 [restart]
+       801    [restart 777]
+       809    [trailer compression=none checksum=0xdd61ae3a]
+       814  footer (53)
+       814    checksum type: crc32c
+       815    meta: offset=777, length=32
+       818    index: offset=69, length=22
+       820    [padding]
+       855    version: 3
+       859    magic number: 0xf09faab3f09faab3
+       867  EOF
diff --git a/sstable/value_block_test.go b/sstable/value_block_test.go
index a605adba..f292c8ac 100644
--- a/sstable/value_block_test.go
+++ b/sstable/value_block_test.go
@@ -7,6 +7,7 @@ package sstable
 import (
 	"fmt"
 	"math"
+	"math/rand"
 	"testing"
 
 	"github.com/cockroachdb/pebble/internal/base"
@@ -22,8 +23,7 @@ func TestValueHandleEncodeDecode(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(fmt.Sprintf("%+v", tc), func(t *testing.T) {
 			n := encodeValueHandle(buf[:], tc)
-			vh, err := decodeValueHandle(buf[:n])
-			require.NoError(t, err)
+			vh := decodeValueHandle(buf[:n])
 			require.Equal(t, tc, vh)
 		})
 	}
@@ -94,3 +94,18 @@ func TestValueBlocksIndexHandleEncodeDecode(t *testing.T) {
 		})
 	}
 }
+
+func TestLittleEndianGetPut(t *testing.T) {
+	testCases := []uint64{
+		0, (1 << 10) - 1, (1 << 25) + 1, math.MaxUint32, math.MaxUint64, uint64(rand.Int63())}
+	var buf [8]byte
+	for _, tc := range testCases {
+		t.Run(fmt.Sprintf("%d", tc), func(t *testing.T) {
+			length := lenLittleEndian(tc)
+			b := buf[:length:length]
+			littleEndianPut(tc, b, length)
+			v := littleEndianGet(b, length)
+			require.Equal(t, tc, v)
+		})
+	}
+}
diff --git a/sstable/writer_test.go b/sstable/writer_test.go
index 145fba4e..97e811e6 100644
--- a/sstable/writer_test.go
+++ b/sstable/writer_test.go
@@ -25,34 +25,48 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func testWriterParallelism(t *testing.T, parallelism bool) {
+	for _, format := range []TableFormat{TableFormatPebblev2, TableFormatPebblev3} {
+		tdFile := "testdata/writer"
+		if format == TableFormatPebblev3 {
+			tdFile = "testdata/writer_v3"
+		}
+		t.Run(format.String(), func(t *testing.T) { runDataDriven(t, tdFile, format, parallelism) })
+	}
+}
 func TestWriter(t *testing.T) {
-	runDataDriven(t, "testdata/writer", false)
+	testWriterParallelism(t, false)
+}
+
+func testRewriterParallelism(t *testing.T, parallelism bool) {
+	for _, format := range []TableFormat{TableFormatPebblev2, TableFormatPebblev3} {
+		tdFile := "testdata/rewriter"
+		if format == TableFormatPebblev3 {
+			tdFile = "testdata/rewriter_v3"
+		}
+		t.Run(format.String(), func(t *testing.T) { runDataDriven(t, tdFile, format, parallelism) })
+	}
 }
 
 func TestRewriter(t *testing.T) {
-	runDataDriven(t, "testdata/rewriter", false)
+	testRewriterParallelism(t, false)
 }
 
 func TestWriterParallel(t *testing.T) {
-	runDataDriven(t, "testdata/writer", true)
+	testWriterParallelism(t, true)
 }
 
 func TestRewriterParallel(t *testing.T) {
-	runDataDriven(t, "testdata/rewriter", true)
+	testRewriterParallelism(t, true)
 }
 
-func runDataDriven(t *testing.T, file string, parallelism bool) {
+func runDataDriven(t *testing.T, file string, tableFormat TableFormat, parallelism bool) {
 	var r *Reader
 	defer func() {
 		if r != nil {
 			require.NoError(t, r.Close())
 		}
 	}()
-	formatVersion := TableFormatMax
-	if rand.Intn(2) == 0 {
-		formatVersion = TableFormatPebblev2
-	}
-	t.Logf("table format %s", formatVersion.String())
 
 	format := func(m *WriterMetadata) string {
 		var b bytes.Buffer
@@ -79,7 +93,7 @@ func runDataDriven(t *testing.T, file string, parallelism bool) {
 			var meta *WriterMetadata
 			var err error
 			meta, r, err = runBuildCmd(td, &WriterOptions{
-				TableFormat: formatVersion,
+				TableFormat: tableFormat,
 				Parallelism: parallelism,
 			}, 0)
 			if err != nil {
@@ -95,7 +109,7 @@ func runDataDriven(t *testing.T, file string, parallelism bool) {
 			var meta *WriterMetadata
 			var err error
 			meta, r, err = runBuildRawCmd(td, &WriterOptions{
-				TableFormat: formatVersion,
+				TableFormat: tableFormat,
 			})
 			if err != nil {
 				return err.Error()
@@ -181,7 +195,7 @@ func runDataDriven(t *testing.T, file string, parallelism bool) {
 			var meta *WriterMetadata
 			var err error
 			meta, r, err = runRewriteCmd(td, r, WriterOptions{
-				TableFormat: formatVersion,
+				TableFormat: tableFormat,
 			})
 			if err != nil {
 				return err.Error()
@@ -197,9 +211,6 @@ func runDataDriven(t *testing.T, file string, parallelism bool) {
 	})
 }
 
-// TODO(sumeer):
-// - test layout after Reader.Layout is implemented.
-// - test rewriteKeySuffixesInBlocks.
 func TestWriterWithValueBlocks(t *testing.T) {
 	var r *Reader
 	defer func() {
@@ -252,7 +263,6 @@ func TestWriterWithValueBlocks(t *testing.T) {
 				Comparer:                  testkeys.Comparer,
 				TableFormat:               formatVersion,
 				Parallelism:               parallelism,
-				EnableValueBlocks:         true,
 				RequiredInPlaceValueBound: inPlaceValueBound,
 				ShortAttributeExtractor:   attributeExtractor,
 			}, 0)
@@ -261,14 +271,34 @@ func TestWriterWithValueBlocks(t *testing.T) {
 			}
 			return formatMeta(meta)
 
+		case "layout":
+			l, err := r.Layout()
+			if err != nil {
+				return err.Error()
+			}
+			var buf bytes.Buffer
+			l.Describe(&buf, true, r, func(key *base.InternalKey, value []byte) {
+				fmt.Fprintf(&buf, "%s:%s", key.String(), string(value))
+			})
+			return buf.String()
+
 		case "scan-raw":
-			// Raw scan does not fetch from value blocks, since we have not written
-			// the read path yet.
-			// TODO(sumeer): add a real scan.
+			// Raw scan does not fetch from value blocks.
 			origIter, err := r.NewIter(nil /* lower */, nil /* upper */)
 			if err != nil {
 				return err.Error()
 			}
+			forceIgnoreValueBlocks := func(i *singleLevelIterator) {
+				i.vbReader = nil
+				i.data.lazyValueHandling.vbr = nil
+				i.data.lazyValueHandling.hasValuePrefix = false
+			}
+			switch i := origIter.(type) {
+			case *twoLevelIterator:
+				forceIgnoreValueBlocks(&i.singleLevelIterator)
+			case *singleLevelIterator:
+				forceIgnoreValueBlocks(i)
+			}
 			iter := newIterAdapter(origIter)
 			defer iter.Close()
 
@@ -280,8 +310,7 @@ func TestWriterWithValueBlocks(t *testing.T) {
 					setWithSamePrefix := setHasSamePrefix(prefix)
 					if isValueHandle(prefix) {
 						attribute := getShortAttribute(prefix)
-						vh, err := decodeValueHandle(v[1:])
-						require.NoError(t, err)
+						vh := decodeValueHandle(v[1:])
 						fmt.Fprintf(&buf, "%s:value-handle len %d block %d offset %d, att %d, same-pre %t\n",
 							iter.Key(), vh.valueLen, vh.blockNum, vh.offsetInBlock, attribute, setWithSamePrefix)
 					} else {
@@ -293,6 +322,62 @@ func TestWriterWithValueBlocks(t *testing.T) {
 			}
 			return buf.String()
 
+		case "scan":
+			origIter, err := r.NewIter(nil /* lower */, nil /* upper */)
+			if err != nil {
+				return err.Error()
+			}
+			iter := newIterAdapter(origIter)
+			defer iter.Close()
+			var buf bytes.Buffer
+			for valid := iter.First(); valid; valid = iter.Next() {
+				fmt.Fprintf(&buf, "%s:%s\n", iter.Key(), iter.Value())
+			}
+			return buf.String()
+
+		case "scan-cloned-lazy-values":
+			iter, err := r.NewIter(nil /* lower */, nil /* upper */)
+			if err != nil {
+				return err.Error()
+			}
+			var fetchers [100]base.LazyFetcher
+			var values []base.LazyValue
+			n := 0
+			var b []byte
+			for k, lv := iter.First(); k != nil; k, lv = iter.Next() {
+				var lvClone base.LazyValue
+				lvClone, b = lv.Clone(b, &fetchers[n])
+				if lv.Fetcher != nil {
+					_, callerOwned, err := lv.Value(nil)
+					require.False(t, callerOwned)
+					require.NoError(t, err)
+				}
+				n++
+				values = append(values, lvClone)
+			}
+			require.NoError(t, iter.Error())
+			iter.Close()
+			var buf bytes.Buffer
+			for i := range values {
+				fmt.Fprintf(&buf, "%d", i)
+				v, callerOwned, err := values[i].Value(nil)
+				require.NoError(t, err)
+				if values[i].Fetcher != nil {
+					require.True(t, callerOwned)
+					fmt.Fprintf(&buf, "(lazy: len %d, attr: %d): %s\n",
+						values[i].Len(), values[i].Fetcher.Attribute.ShortAttribute, string(v))
+					v2, callerOwned, err := values[i].Value(nil)
+					require.NoError(t, err)
+					require.True(t, callerOwned)
+					require.Equal(t, &v[0], &v2[0])
+
+				} else {
+					require.False(t, callerOwned)
+					fmt.Fprintf(&buf, "(in-place: len %d): %s\n", values[i].Len(), string(v))
+				}
+			}
+			return buf.String()
+
 		default:
 			return fmt.Sprintf("unknown command: %s", td.Cmd)
 		}
@@ -455,15 +540,20 @@ func TestSizeEstimate(t *testing.T) {
 		})
 }
 
-// TODO(sumeer): update to test value blocks after read path (including Layout
-// changes) is implemented.
 func TestWriterClearCache(t *testing.T) {
 	// Verify that Writer clears the cache of blocks that it writes.
 	mem := vfs.NewMem()
-	opts := ReaderOptions{Cache: cache.New(64 << 20)}
+	opts := ReaderOptions{
+		Cache:    cache.New(64 << 20),
+		Comparer: testkeys.Comparer,
+	}
 	defer opts.Cache.Unref()
 
-	writerOpts := WriterOptions{Cache: opts.Cache}
+	writerOpts := WriterOptions{
+		Cache:       opts.Cache,
+		Comparer:    testkeys.Comparer,
+		TableFormat: TableFormatPebblev3,
+	}
 	cacheOpts := &cacheOpts{cacheID: 1, fileNum: 1}
 	invalidData := func() *cache.Value {
 		invalid := []byte("invalid data")
@@ -478,6 +568,8 @@ func TestWriterClearCache(t *testing.T) {
 
 		w := NewWriter(f, writerOpts, cacheOpts)
 		require.NoError(t, w.Set([]byte("hello"), []byte("world")))
+		require.NoError(t, w.Set([]byte("hello@42"), []byte("world@42")))
+		require.NoError(t, w.Set([]byte("hello@5"), []byte("world@5")))
 		require.NoError(t, w.Close())
 	}
 
@@ -504,6 +596,12 @@ func TestWriterClearCache(t *testing.T) {
 		f(layout.TopIndex)
 		f(layout.Filter)
 		f(layout.RangeDel)
+		for _, bh := range layout.ValueBlock {
+			f(bh)
+		}
+		if layout.ValueIndex.Length != 0 {
+			f(layout.ValueIndex)
+		}
 		f(layout.Properties)
 		f(layout.MetaIndex)
 	}
diff --git a/testdata/event_listener b/testdata/event_listener
index dd19a734..15d92e66 100644
--- a/testdata/event_listener
+++ b/testdata/event_listener
@@ -183,20 +183,20 @@ close: db/marker.manifest.000006.MANIFEST-000016
 sync: db
 [JOB 14] MANIFEST created 000016
 [JOB 14] MANIFEST deleted 000011
-[JOB 14] ingested L0:000015 (825 B)
+[JOB 14] ingested L0:000015 (826 B)
 
 metrics
 ----
 __level_____count____size___score______in__ingest(sz_cnt)____move(sz_cnt)___write(sz_cnt)____read___r-amp___w-amp
     WAL         1    27 B       -    48 B       -       -       -       -   108 B       -       -       -     2.2
-      0         2   1.6 K    0.40    81 B   825 B       1     0 B       0   2.3 K       3     0 B       2    28.5
+      0         2   1.6 K    0.40    81 B   826 B       1     0 B       0   2.3 K       3     0 B       2    28.5
       1         0     0 B    0.00     0 B     0 B       0     0 B       0     0 B       0     0 B       0     0.0
       2         0     0 B    0.00     0 B     0 B       0     0 B       0     0 B       0     0 B       0     0.0
       3         0     0 B    0.00     0 B     0 B       0     0 B       0     0 B       0     0 B       0     0.0
       4         0     0 B    0.00     0 B     0 B       0     0 B       0     0 B       0     0 B       0     0.0
       5         0     0 B    0.00     0 B     0 B       0     0 B       0     0 B       0     0 B       0     0.0
       6         1   770 B       -   1.5 K     0 B       0     0 B       0   770 B       1   1.5 K       1     0.5
-  total         3   2.3 K       -   933 B   825 B       1     0 B       0   3.9 K       4   1.5 K       3     4.3
+  total         3   2.3 K       -   934 B   826 B       1     0 B       0   3.9 K       4   1.5 K       3     4.3
   flush         3
 compact         1   2.3 K     0 B       0          (size == estimated-debt, score = in-progress-bytes, in = num-in-progress)
   ctype         1       0       0       0       0       0       0  (default, delete, elision, move, read, rewrite, multi-level)
@@ -204,7 +204,7 @@ compact         1   2.3 K     0 B       0          (size == estimated-debt, scor
 zmemtbl         0     0 B
    ztbl         0     0 B
  bcache         8   1.4 K   11.1%  (score == hit-rate)
- tcache         1   688 B   50.0%  (score == hit-rate)
+ tcache         1   712 B   50.0%  (score == hit-rate)
   snaps         0       -       0  (score == earliest seq num)
  titers         0
  filter         -       -    0.0%  (score == utility)
diff --git a/testdata/format_major_version_pebblev1_migration b/testdata/format_major_version_pebblev1_migration
index 8e53de67..5f698196 100644
--- a/testdata/format_major_version_pebblev1_migration
+++ b/testdata/format_major_version_pebblev1_migration
@@ -34,22 +34,22 @@ flush
 # Ingest an external table written at the max table format for the current
 # version (i.e. Pebblev2).
 
-ingest format=4
+ingest a format=pebblev2
 set pebblev2 pebblev2
 ----
 
 # Ingest some external table written at earlier versions (i.e. Pebblev1,
 # RocksDBv2, LevelDB).
 
-ingest format=3
+ingest b format=pebblev1
 set pebblev1 pebblev1
 ----
 
-ingest format=2
+ingest c format=rocksdbv2
 set rocksdbv2 rockdbv2
 ----
 
-ingest format=1
+ingest d format=leveldb
 set leveldb leveldb
 ----
 
@@ -92,7 +92,7 @@ max-table-format
 
 # Ingesting a table with a format prior to this version fails.
 
-ingest format=2
+ingest e format=rocksdbv2
 set rocksdbv2 rockdbv2
 ----
 pebble: table format (RocksDB,v2) is not within range supported at DB format major version 9, ((Pebble,v1),(Pebble,v2))
diff --git a/testdata/ingest b/testdata/ingest
index f4debbc7..967d99cc 100644
--- a/testdata/ingest
+++ b/testdata/ingest
@@ -16,7 +16,7 @@ ingest ext0
 lsm
 ----
 
-build ext0
+build ext0 format=pebblev2
 set a 1
 set b 2
 ----
@@ -48,7 +48,7 @@ compact         0     0 B     0 B       0          (size == estimated-debt, scor
 zmemtbl         0     0 B
    ztbl         0     0 B
  bcache         8   1.5 K   42.9%  (score == hit-rate)
- tcache         1   688 B   50.0%  (score == hit-rate)
+ tcache         1   712 B   50.0%  (score == hit-rate)
   snaps         0       -       0  (score == earliest seq num)
  titers         0
  filter         -       -    0.0%  (score == utility)
@@ -107,7 +107,7 @@ b
 a:3
 b: pebble: not found
 
-build ext2
+build ext2 format=pebblev2
 set a 4
 set b 5
 set c 6
@@ -294,7 +294,7 @@ n
 m:12
 n:13
 
-build ext7
+build ext7 format=pebblev2
 del-range a c
 del-range x z
 ----
diff --git a/testdata/metrics b/testdata/metrics
index 64ecda40..7b59541c 100644
--- a/testdata/metrics
+++ b/testdata/metrics
@@ -34,7 +34,7 @@ compact         0     0 B     0 B       0          (size == estimated-debt, scor
 zmemtbl         1   256 K
    ztbl         0     0 B
  bcache         4   698 B    0.0%  (score == hit-rate)
- tcache         1   688 B    0.0%  (score == hit-rate)
+ tcache         1   712 B    0.0%  (score == hit-rate)
   snaps         0       -       0  (score == earliest seq num)
  titers         1
  filter         -       -    0.0%  (score == utility)
@@ -82,7 +82,7 @@ compact         1     0 B     0 B       0          (size == estimated-debt, scor
 zmemtbl         2   512 K
    ztbl         2   1.5 K
  bcache         8   1.4 K   42.9%  (score == hit-rate)
- tcache         2   1.3 K   66.7%  (score == hit-rate)
+ tcache         2   1.4 K   66.7%  (score == hit-rate)
   snaps         0       -       0  (score == earliest seq num)
  titers         2
  filter         -       -    0.0%  (score == utility)
@@ -115,7 +115,7 @@ compact         1     0 B     0 B       0          (size == estimated-debt, scor
 zmemtbl         1   256 K
    ztbl         2   1.5 K
  bcache         8   1.4 K   42.9%  (score == hit-rate)
- tcache         2   1.3 K   66.7%  (score == hit-rate)
+ tcache         2   1.4 K   66.7%  (score == hit-rate)
   snaps         0       -       0  (score == earliest seq num)
  titers         2
  filter         -       -    0.0%  (score == utility)
@@ -145,7 +145,7 @@ compact         1     0 B     0 B       0          (size == estimated-debt, scor
 zmemtbl         1   256 K
    ztbl         1   771 B
  bcache         4   698 B   42.9%  (score == hit-rate)
- tcache         1   688 B   66.7%  (score == hit-rate)
+ tcache         1   712 B   66.7%  (score == hit-rate)
   snaps         0       -       0  (score == earliest seq num)
  titers         1
  filter         -       -    0.0%  (score == utility)