Repository: petermattis/pebble
Pull request #228 (2019-08-17 00:12:02)

Implement `wal dump` tool
* Add Reader.Offset() for retrieving the record offset
* Move filename building and parsing to internal/base
* Export Batch.{SetNum,Count}
* Remove unused reverse fields

petermattis: 

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

petermattis: FYI, the first 4 commits are restructuring and minor changes. The last commit is the one with all the meat.

Patch:
diff --git a/batch.go b/batch.go
index de01ff1..9073369 100644
--- a/batch.go
+++ b/batch.go
@@ -286,7 +286,7 @@ func (b *Batch) Apply(batch *Batch, _ *WriteOptions) error {
 	b.storage.data = append(b.storage.data, batch.storage.data[batchHeaderLen:]...)
 
 	count := binary.LittleEndian.Uint32(batch.storage.data[8:12])
-	b.setCount(b.count() + count)
+	b.setCount(b.Count() + count)
 
 	for iter := BatchReader(b.storage.data[offset:]); len(iter) > 0; {
 		offset := uintptr(unsafe.Pointer(&iter[0])) - uintptr(unsafe.Pointer(&b.storage.data[0]))
@@ -459,7 +459,7 @@ func (b *Batch) LogData(data []byte, _ *WriteOptions) error {
 	pos := len(b.storage.data)
 	b.grow(1 + maxVarintLen32 + len(data))
 	b.storage.data[pos] = byte(InternalKeyKindLogData)
-	pos, varlen1 := b.copyStr(pos+1, data)
+	_, varlen1 := b.copyStr(pos+1, data)
 	b.storage.data = b.storage.data[:len(b.storage.data)-(maxVarintLen32-varlen1)]
 	return nil
 }
@@ -470,6 +470,17 @@ func (b *Batch) Repr() []byte {
 	return b.storage.data
 }
 
+// SetRepr sets the underlying batch representation. The batch takes ownership
+// of the supplied slice. It is not safe to modify it afterwards until the
+// Batch is no longer in use.
+func (b *Batch) SetRepr(data []byte) error {
+	if len(data) < batchHeaderLen {
+		return fmt.Errorf("invalid batch")
+	}
+	b.storage.data = data
+	return nil
+}
+
 // NewIter returns an iterator that is unpositioned (Iterator.Valid() will
 // return false). The iterator can be positioned via a call to SeekGE,
 // SeekPrefixGE, SeekLT, First or Last. Only indexed batches support iterators.
@@ -637,7 +648,10 @@ func (b *Batch) setSeqNum(seqNum uint64) {
 	binary.LittleEndian.PutUint64(b.seqNumData(), seqNum)
 }
 
-func (b *Batch) seqNum() uint64 {
+// SeqNum returns the batch sequence number which is applied to the first
+// record in the batch. The sequence number is incremented for each subsequent
+// record.
+func (b *Batch) SeqNum() uint64 {
 	return binary.LittleEndian.Uint64(b.seqNumData())
 }
 
@@ -645,7 +659,8 @@ func (b *Batch) setCount(v uint32) {
 	binary.LittleEndian.PutUint32(b.countData(), v)
 }
 
-func (b *Batch) count() uint32 {
+// Count returns the number of records in the batch.
+func (b *Batch) Count() uint32 {
 	return binary.LittleEndian.Uint32(b.countData())
 }
 
@@ -741,11 +756,10 @@ func (r *BatchReader) nextStr() (s []byte, ok bool) {
 // Note: batchIter mirrors the implementation of flushableBatchIter. Keep the
 // two in sync.
 type batchIter struct {
-	cmp     Compare
-	batch   *Batch
-	reverse bool
-	iter    batchskl.Iterator
-	err     error
+	cmp   Compare
+	batch *Batch
+	iter  batchskl.Iterator
+	err   error
 }
 
 // batchIter implements the internalIterator interface.
@@ -873,7 +887,7 @@ func newFlushableBatch(batch *Batch, comparer *Comparer) *flushableBatch {
 	b := &flushableBatch{
 		data:      batch.storage.data,
 		cmp:       comparer.Compare,
-		offsets:   make([]flushableBatchEntry, 0, batch.count()),
+		offsets:   make([]flushableBatchEntry, 0, batch.Count()),
 		flushedCh: make(chan struct{}),
 	}
 
@@ -1016,7 +1030,6 @@ type flushableBatchIter struct {
 	data    []byte
 	offsets []flushableBatchEntry
 	cmp     Compare
-	reverse bool
 	index   int
 	key     InternalKey
 	err     error
diff --git a/commit.go b/commit.go
index 32fcfd5..bd953da 100644
--- a/commit.go
+++ b/commit.go
@@ -317,7 +317,7 @@ func (p *commitPipeline) AllocateSeqNum(count int, prepare func(), apply func(se
 	p.mu.Unlock()
 
 	// Invoke the apply callback.
-	apply(b.seqNum())
+	apply(b.SeqNum())
 
 	// Publish the sequence number.
 	p.publish(b)
@@ -326,7 +326,7 @@ func (p *commitPipeline) AllocateSeqNum(count int, prepare func(), apply func(se
 }
 
 func (p *commitPipeline) prepare(b *Batch, syncWAL bool) (*memTable, error) {
-	n := uint64(b.count())
+	n := uint64(b.Count())
 	if n == invalidBatchCount {
 		return nil, ErrInvalidBatch
 	}
@@ -390,7 +390,7 @@ func (p *commitPipeline) publish(b *Batch) {
 		// that the sequence number ratchets up.
 		for {
 			curSeqNum := atomic.LoadUint64(p.env.visibleSeqNum)
-			newSeqNum := t.seqNum() + uint64(t.count())
+			newSeqNum := t.SeqNum() + uint64(t.Count())
 			if newSeqNum <= curSeqNum {
 				// t's sequence number has already been published.
 				break
diff --git a/compaction.go b/compaction.go
index a7354b4..0cc9d12 100644
--- a/compaction.go
+++ b/compaction.go
@@ -954,7 +954,7 @@ func (d *DB) runCompaction(jobID int, c *compaction, pacer pacer) (
 		pendingOutputs = append(pendingOutputs, fileNum)
 		d.mu.Unlock()
 
-		filename := dbFilename(d.dirname, fileTypeTable, fileNum)
+		filename := base.MakeFilename(d.dirname, fileTypeTable, fileNum)
 		file, err := d.opts.FS.Create(filename)
 		if err != nil {
 			return err
@@ -1137,7 +1137,7 @@ func (d *DB) scanObsoleteFiles(list []string) {
 	var obsoleteOptions []uint64
 
 	for _, filename := range list {
-		fileType, fileNum, ok := parseDBFilename(filename)
+		fileType, fileNum, ok := base.ParseFilename(filename)
 		if !ok {
 			continue
 		}
@@ -1242,7 +1242,7 @@ func (d *DB) deleteObsoleteFiles(jobID int) {
 				d.tableCache.evict(fileNum)
 			}
 
-			path := dbFilename(d.dirname, f.fileType, fileNum)
+			path := base.MakeFilename(d.dirname, f.fileType, fileNum)
 			err := d.opts.FS.Remove(path)
 			if err == os.ErrNotExist {
 				continue
diff --git a/db.go b/db.go
index 8228fd2..91475c1 100644
--- a/db.go
+++ b/db.go
@@ -404,7 +404,7 @@ func (d *DB) commitApply(b *Batch, mem *memTable) error {
 		// This is a large batch which was already added to the immutable queue.
 		return nil
 	}
-	err := mem.apply(b, b.seqNum())
+	err := mem.apply(b, b.SeqNum())
 	if err != nil {
 		return err
 	}
@@ -420,7 +420,7 @@ func (d *DB) commitWrite(b *Batch, wg *sync.WaitGroup) (*memTable, error) {
 	d.mu.Lock()
 
 	if b.flushable != nil {
-		b.flushable.seqNum = b.seqNum()
+		b.flushable.seqNum = b.SeqNum()
 	}
 
 	// Switch out the memtable if there was not enough room to store the batch.
@@ -920,7 +920,7 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 			d.mu.mem.switching = true
 			d.mu.Unlock()
 
-			newLogName := dbFilename(d.walDirname, fileTypeLog, newLogNumber)
+			newLogName := base.MakeFilename(d.walDirname, fileTypeLog, newLogNumber)
 
 			// Try to use a recycled log file. Recycling log files is an important
 			// performance optimization as it is faster to sync a file that has
@@ -930,7 +930,7 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 			// preallocation is performed (e.g. fallocate).
 			recycleLogNumber := d.logRecycler.peek()
 			if recycleLogNumber > 0 {
-				recycleLogName := dbFilename(d.walDirname, fileTypeLog, recycleLogNumber)
+				recycleLogName := base.MakeFilename(d.walDirname, fileTypeLog, recycleLogNumber)
 				err = d.opts.FS.Rename(recycleLogName, newLogName)
 			}
 
diff --git a/filenames.go b/filenames.go
index 7467bc9..8fe5048 100644
--- a/filenames.go
+++ b/filenames.go
@@ -6,86 +6,24 @@ package pebble
 
 import (
 	"fmt"
-	"os"
-	"path/filepath"
-	"strconv"
-	"strings"
 
+	"github.com/petermattis/pebble/internal/base"
 	"github.com/petermattis/pebble/vfs"
 )
 
-type fileType int
+type fileType = base.FileType
 
 const (
-	fileTypeLog fileType = iota
-	fileTypeLock
-	fileTypeTable
-	fileTypeManifest
-	fileTypeCurrent
-	fileTypeOptions
+	fileTypeLog      = base.FileTypeLog
+	fileTypeLock     = base.FileTypeLock
+	fileTypeTable    = base.FileTypeTable
+	fileTypeManifest = base.FileTypeManifest
+	fileTypeCurrent  = base.FileTypeCurrent
+	fileTypeOptions  = base.FileTypeOptions
 )
 
-func dbFilename(dirname string, fileType fileType, fileNum uint64) string {
-	for len(dirname) > 0 && dirname[len(dirname)-1] == os.PathSeparator {
-		dirname = dirname[:len(dirname)-1]
-	}
-	switch fileType {
-	case fileTypeLog:
-		return fmt.Sprintf("%s%c%06d.log", dirname, os.PathSeparator, fileNum)
-	case fileTypeLock:
-		return fmt.Sprintf("%s%cLOCK", dirname, os.PathSeparator)
-	case fileTypeTable:
-		return fmt.Sprintf("%s%c%06d.sst", dirname, os.PathSeparator, fileNum)
-	case fileTypeManifest:
-		return fmt.Sprintf("%s%cMANIFEST-%06d", dirname, os.PathSeparator, fileNum)
-	case fileTypeCurrent:
-		return fmt.Sprintf("%s%cCURRENT", dirname, os.PathSeparator)
-	case fileTypeOptions:
-		return fmt.Sprintf("%s%cOPTIONS-%06d", dirname, os.PathSeparator, fileNum)
-	}
-	panic("unreachable")
-}
-
-func parseDBFilename(filename string) (fileType fileType, fileNum uint64, ok bool) {
-	filename = filepath.Base(filename)
-	switch {
-	case filename == "CURRENT":
-		return fileTypeCurrent, 0, true
-	case filename == "LOCK":
-		return fileTypeLock, 0, true
-	case strings.HasPrefix(filename, "MANIFEST-"):
-		u, err := strconv.ParseUint(filename[len("MANIFEST-"):], 10, 64)
-		if err != nil {
-			break
-		}
-		return fileTypeManifest, u, true
-	case strings.HasPrefix(filename, "OPTIONS-"):
-		u, err := strconv.ParseUint(filename[len("OPTIONS-"):], 10, 64)
-		if err != nil {
-			break
-		}
-		return fileTypeOptions, u, true
-	default:
-		i := strings.IndexByte(filename, '.')
-		if i < 0 {
-			break
-		}
-		u, err := strconv.ParseUint(filename[:i], 10, 64)
-		if err != nil {
-			break
-		}
-		switch filename[i+1:] {
-		case "sst":
-			return fileTypeTable, u, true
-		case "log":
-			return fileTypeLog, u, true
-		}
-	}
-	return 0, 0, false
-}
-
 func setCurrentFile(dirname string, fs vfs.FS, fileNum uint64) error {
-	newFilename := dbFilename(dirname, fileTypeCurrent, fileNum)
+	newFilename := base.MakeFilename(dirname, fileTypeCurrent, fileNum)
 	oldFilename := fmt.Sprintf("%s.%06d.dbtmp", newFilename, fileNum)
 	fs.Remove(oldFilename)
 	f, err := fs.Create(oldFilename)
diff --git a/ingest.go b/ingest.go
index 2f16523..149f693 100644
--- a/ingest.go
+++ b/ingest.go
@@ -122,7 +122,7 @@ func ingestSortAndVerify(cmp Compare, meta []*fileMetadata) error {
 func ingestCleanup(fs vfs.FS, dirname string, meta []*fileMetadata) error {
 	var firstErr error
 	for i := range meta {
-		target := dbFilename(dirname, fileTypeTable, meta[i].fileNum)
+		target := base.MakeFilename(dirname, fileTypeTable, meta[i].fileNum)
 		if err := fs.Remove(target); err != nil {
 			if firstErr != nil {
 				firstErr = err
@@ -134,7 +134,7 @@ func ingestCleanup(fs vfs.FS, dirname string, meta []*fileMetadata) error {
 
 func ingestLink(opts *Options, dirname string, paths []string, meta []*fileMetadata) error {
 	for i := range paths {
-		target := dbFilename(dirname, fileTypeTable, meta[i].fileNum)
+		target := base.MakeFilename(dirname, fileTypeTable, meta[i].fileNum)
 		err := opts.FS.Link(paths[i], target)
 		if err != nil {
 			if err2 := ingestCleanup(opts.FS, dirname, meta[:i]); err2 != nil {
diff --git a/internal/base/filenames.go b/internal/base/filenames.go
new file mode 100644
index 0000000..cd7bbb8
--- /dev/null
+++ b/internal/base/filenames.go
@@ -0,0 +1,87 @@
+// Copyright 2012 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package base
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+)
+
+// FileType enumerates the types of files found in a DB.
+type FileType int
+
+// The FileType enumeration.
+const (
+	FileTypeLog FileType = iota
+	FileTypeLock
+	FileTypeTable
+	FileTypeManifest
+	FileTypeCurrent
+	FileTypeOptions
+)
+
+// MakeFilename builds a filename from components.
+func MakeFilename(dirname string, fileType FileType, fileNum uint64) string {
+	for len(dirname) > 0 && dirname[len(dirname)-1] == os.PathSeparator {
+		dirname = dirname[:len(dirname)-1]
+	}
+	switch fileType {
+	case FileTypeLog:
+		return fmt.Sprintf("%s%c%06d.log", dirname, os.PathSeparator, fileNum)
+	case FileTypeLock:
+		return fmt.Sprintf("%s%cLOCK", dirname, os.PathSeparator)
+	case FileTypeTable:
+		return fmt.Sprintf("%s%c%06d.sst", dirname, os.PathSeparator, fileNum)
+	case FileTypeManifest:
+		return fmt.Sprintf("%s%cMANIFEST-%06d", dirname, os.PathSeparator, fileNum)
+	case FileTypeCurrent:
+		return fmt.Sprintf("%s%cCURRENT", dirname, os.PathSeparator)
+	case FileTypeOptions:
+		return fmt.Sprintf("%s%cOPTIONS-%06d", dirname, os.PathSeparator, fileNum)
+	}
+	panic("unreachable")
+}
+
+// ParseFilename parses the components from a filename.
+func ParseFilename(filename string) (fileType FileType, fileNum uint64, ok bool) {
+	filename = filepath.Base(filename)
+	switch {
+	case filename == "CURRENT":
+		return FileTypeCurrent, 0, true
+	case filename == "LOCK":
+		return FileTypeLock, 0, true
+	case strings.HasPrefix(filename, "MANIFEST-"):
+		u, err := strconv.ParseUint(filename[len("MANIFEST-"):], 10, 64)
+		if err != nil {
+			break
+		}
+		return FileTypeManifest, u, true
+	case strings.HasPrefix(filename, "OPTIONS-"):
+		u, err := strconv.ParseUint(filename[len("OPTIONS-"):], 10, 64)
+		if err != nil {
+			break
+		}
+		return FileTypeOptions, u, true
+	default:
+		i := strings.IndexByte(filename, '.')
+		if i < 0 {
+			break
+		}
+		u, err := strconv.ParseUint(filename[:i], 10, 64)
+		if err != nil {
+			break
+		}
+		switch filename[i+1:] {
+		case "sst":
+			return FileTypeTable, u, true
+		case "log":
+			return FileTypeLog, u, true
+		}
+	}
+	return 0, 0, false
+}
diff --git a/internal/record/record.go b/internal/record/record.go
index 0ba81b8..3d18e73 100644
--- a/internal/record/record.go
+++ b/internal/record/record.go
@@ -155,6 +155,8 @@ type Reader struct {
 	// logNum is the low 32-bits of the log's file number. May be zero when used
 	// with log files that do not have a file number (e.g. the MANIFEST).
 	logNum uint32
+	// blockNum is the zero based block number currently held in buf.
+	blockNum int64
 	// seq is the sequence number of the current record.
 	seq int
 	// buf[begin:end] is the unread portion of the current chunk's payload. The
@@ -180,8 +182,9 @@ type Reader struct {
 // match the specifed logNum.
 func NewReader(r io.Reader, logNum uint64) *Reader {
 	return &Reader{
-		r:      r,
-		logNum: uint32(logNum),
+		r:        r,
+		logNum:   uint32(logNum),
+		blockNum: -1,
 	}
 }
 
@@ -265,6 +268,7 @@ func (r *Reader) nextChunk(wantFirst bool) error {
 			return err
 		}
 		r.begin, r.end, r.n = 0, 0, n
+		r.blockNum++
 	}
 }
 
@@ -285,6 +289,15 @@ func (r *Reader) Next() (io.Reader, error) {
 	return singleReader{r, r.seq}, nil
 }
 
+// Offset returns the current offset within the file. If called immediately
+// before a call to Next(), Offset() will return the record offset.
+func (r *Reader) Offset() int64 {
+	if r.blockNum < 0 {
+		return 0
+	}
+	return int64(r.blockNum)*blockSize + int64(r.end)
+}
+
 // recover clears any errors read so far, so that calling Next will start
 // reading from the next good 32KiB block. If there are no such blocks, Next
 // will return io.EOF. recover also marks the current reader, the one most
diff --git a/mem_table.go b/mem_table.go
index b4ce7d2..aac52df 100644
--- a/mem_table.go
+++ b/mem_table.go
@@ -163,7 +163,7 @@ func (m *memTable) apply(batch *Batch, seqNum uint64) error {
 			return err
 		}
 	}
-	if seqNum != startSeqNum+uint64(batch.count()) {
+	if seqNum != startSeqNum+uint64(batch.Count()) {
 		panic("pebble: inconsistent batch count")
 	}
 	if tombstoneCount != 0 {
diff --git a/open.go b/open.go
index 80018fa..5590e7c 100644
--- a/open.go
+++ b/open.go
@@ -15,6 +15,7 @@ import (
 	"sync"
 
 	"github.com/petermattis/pebble/internal/arenaskl"
+	"github.com/petermattis/pebble/internal/base"
 	"github.com/petermattis/pebble/internal/rate"
 	"github.com/petermattis/pebble/internal/record"
 	"github.com/petermattis/pebble/vfs"
@@ -39,7 +40,7 @@ func createDB(dirname string, opts *Options) (retErr error) {
 		comparatorName: opts.Comparer.Name,
 		nextFileNumber: manifestFileNum + 1,
 	}
-	manifestFilename := dbFilename(dirname, fileTypeManifest, manifestFileNum)
+	manifestFilename := base.MakeFilename(dirname, fileTypeManifest, manifestFileNum)
 	f, err := opts.FS.Create(manifestFilename)
 	if err != nil {
 		return fmt.Errorf("pebble: could not create %q: %v", manifestFilename, err)
@@ -122,7 +123,7 @@ func Open(dirname string, opts *Options) (*DB, error) {
 			return nil, err
 		}
 	}
-	fileLock, err := opts.FS.Lock(dbFilename(dirname, fileTypeLock, 0))
+	fileLock, err := opts.FS.Lock(base.MakeFilename(dirname, fileTypeLock, 0))
 	if err != nil {
 		return nil, err
 	}
@@ -151,7 +152,7 @@ func Open(dirname string, opts *Options) (*DB, error) {
 		d.walDir, err = opts.FS.OpenDir(d.walDirname)
 	}
 
-	if _, err := opts.FS.Stat(dbFilename(dirname, fileTypeCurrent, 0)); os.IsNotExist(err) && !d.opts.ReadOnly {
+	if _, err := opts.FS.Stat(base.MakeFilename(dirname, fileTypeCurrent, 0)); os.IsNotExist(err) && !d.opts.ReadOnly {
 		// Create the DB if it did not already exist.
 		if err := createDB(dirname, opts); err != nil {
 			return nil, err
@@ -190,7 +191,7 @@ func Open(dirname string, opts *Options) (*DB, error) {
 	}
 	var logFiles []fileNumAndName
 	for _, filename := range ls {
-		ft, fn, ok := parseDBFilename(filename)
+		ft, fn, ok := base.ParseFilename(filename)
 		if !ok {
 			continue
 		}
@@ -229,7 +230,7 @@ func Open(dirname string, opts *Options) (*DB, error) {
 		// Create an empty .log file.
 		ve.logNumber = d.mu.versions.nextFileNum()
 		d.mu.log.queue = append(d.mu.log.queue, ve.logNumber)
-		logFile, err := opts.FS.Create(dbFilename(d.walDirname, fileTypeLog, ve.logNumber))
+		logFile, err := opts.FS.Create(base.MakeFilename(d.walDirname, fileTypeLog, ve.logNumber))
 		if err != nil {
 			return nil, err
 		}
@@ -253,7 +254,7 @@ func Open(dirname string, opts *Options) (*DB, error) {
 	if !d.opts.ReadOnly {
 		// Write the current options to disk.
 		d.optionsFileNum = d.mu.versions.nextFileNum()
-		optionsFile, err := opts.FS.Create(dbFilename(dirname, fileTypeOptions, d.optionsFileNum))
+		optionsFile, err := opts.FS.Create(base.MakeFilename(dirname, fileTypeOptions, d.optionsFileNum))
 		if err != nil {
 			return nil, err
 		}
@@ -330,8 +331,8 @@ func (d *DB) replayWAL(
 		b = Batch{}
 		b.storage.data = buf.Bytes()
 		b.refreshMemTableSize()
-		seqNum := b.seqNum()
-		maxSeqNum = seqNum + uint64(b.count())
+		seqNum := b.SeqNum()
+		maxSeqNum = seqNum + uint64(b.Count())
 
 		if mem == nil {
 			mem = newMemTable(d.opts)
diff --git a/table_cache.go b/table_cache.go
index 65f1e9f..9bd4f1b 100644
--- a/table_cache.go
+++ b/table_cache.go
@@ -13,6 +13,7 @@ import (
 	"sync"
 	"sync/atomic"
 
+	"github.com/petermattis/pebble/internal/base"
 	"github.com/petermattis/pebble/sstable"
 	"github.com/petermattis/pebble/vfs"
 )
@@ -327,7 +328,8 @@ type tableCacheNode struct {
 
 func (n *tableCacheNode) load(c *tableCacheShard) {
 	// Try opening the fileTypeTable first.
-	f, err := c.fs.Open(dbFilename(c.dirname, fileTypeTable, n.meta.fileNum), vfs.RandomReadsOption)
+	f, err := c.fs.Open(base.MakeFilename(c.dirname, fileTypeTable, n.meta.fileNum),
+		vfs.RandomReadsOption)
 	if err != nil {
 		n.err = err
 		close(n.loaded)
diff --git a/tool/sstable.go b/tool/sstable.go
index 90119d0..fcc872e 100644
--- a/tool/sstable.go
+++ b/tool/sstable.go
@@ -47,7 +47,7 @@ func newSSTable(
 		mergers:   mergers,
 	}
 	s.fmtKey.mustSet("quoted")
-	s.fmtValue.mustSet("hex")
+	s.fmtValue.mustSet("[%x]")
 
 	s.Root = &cobra.Command{
 		Use:   "sstable",
diff --git a/tool/util.go b/tool/util.go
index d0400fd..b104731 100644
--- a/tool/util.go
+++ b/tool/util.go
@@ -64,12 +64,12 @@ func (f *formatter) Type() string {
 func (f *formatter) Set(spec string) error {
 	f.spec = spec
 	switch spec {
-	case "hex":
-		f.fn = formatHex
 	case "null":
 		f.fn = formatNull
 	case "quoted":
 		f.fn = formatQuoted
+	case "size":
+		f.fn = formatSize
 	default:
 		if strings.Count(spec, "%") != 1 {
 			return fmt.Errorf("unknown formatter: %q", spec)
@@ -87,10 +87,6 @@ func (f *formatter) mustSet(spec string) {
 	}
 }
 
-func formatHex(w io.Writer, v []byte) {
-	fmt.Fprintf(w, "[% x]", v)
-}
-
 func formatNull(w io.Writer, v []byte) {
 }
 
@@ -100,6 +96,10 @@ func formatQuoted(w io.Writer, v []byte) {
 	w.Write(q)
 }
 
+func formatSize(w io.Writer, v []byte) {
+	fmt.Fprintf(w, "<%d>", len(v))
+}
+
 func formatKeyValue(
 	w io.Writer, fmtKey formatter, fmtValue formatter, key *base.InternalKey, value []byte,
 ) {
diff --git a/tool/wal.go b/tool/wal.go
index 3578ceb..437730f 100644
--- a/tool/wal.go
+++ b/tool/wal.go
@@ -5,9 +5,14 @@
 package tool
 
 import (
+	"bytes"
 	"fmt"
+	"io"
 
+	"github.com/petermattis/pebble"
 	"github.com/petermattis/pebble/internal/base"
+	"github.com/petermattis/pebble/internal/record"
+	"github.com/petermattis/pebble/vfs"
 	"github.com/spf13/cobra"
 )
 
@@ -16,10 +21,15 @@ import (
 type walT struct {
 	Root *cobra.Command
 	Dump *cobra.Command
+
+	fmtKey   formatter
+	fmtValue formatter
 }
 
 func newWAL(opts *base.Options) *walT {
 	w := &walT{}
+	w.fmtKey.mustSet("quoted")
+	w.fmtValue.mustSet("size")
 
 	w.Root = &cobra.Command{
 		Use:   "wal",
@@ -36,10 +46,95 @@ Print the contents of the WAL files.
 	}
 
 	w.Root.AddCommand(w.Dump)
+
+	w.Dump.Flags().Var(
+		&w.fmtKey, "key", "key formatter")
+	w.Dump.Flags().Var(
+		&w.fmtValue, "value", "value formatter")
 	return w
 }
 
 func (w *walT) runDump(cmd *cobra.Command, args []string) {
-	fmt.Fprintf(stderr, "TODO(peter): \"wal dump\" unimplemented\n")
-	osExit(1)
+	for _, arg := range args {
+		func() {
+			// Parse the filename in order to extract the file number. This is
+			// necessary in case WAL recycling was used (which it is usually is). If
+			// we can't parse the filename or it isn't a log file, we'll plow ahead
+			// anyways (which will likely fail when we try to read the file).
+			_, fileNum, ok := base.ParseFilename(arg)
+			if !ok {
+				fileNum = 0
+			}
+
+			f, err := vfs.Default.Open(arg)
+			if err != nil {
+				fmt.Fprintf(stderr, "%s\n", err)
+				return
+			}
+			defer f.Close()
+
+			fmt.Fprintf(stdout, "%s\n", arg)
+
+			var b pebble.Batch
+			var buf bytes.Buffer
+			rr := record.NewReader(f, fileNum)
+			for {
+				offset := rr.Offset()
+				r, err := rr.Next()
+				if err == nil {
+					buf.Reset()
+					_, err = io.Copy(&buf, r)
+				}
+				if err != nil {
+					// It is common to encounter a zeroed or invalid chunk due to WAL
+					// preallocation and WAL recycling. We need to distinguish these
+					// errors from EOF in order to recognize that the record was
+					// truncated, but want to otherwise treat them like EOF.
+					switch err {
+					case record.ErrZeroedChunk:
+						fmt.Fprintf(stdout, "EOF [%s] (may be due to WAL preallocation)\n", err)
+					case record.ErrInvalidChunk:
+						fmt.Fprintf(stdout, "EOF [%s] (may be due to WAL recycling)\n", err)
+					default:
+						fmt.Fprintf(stdout, "%s\n", err)
+					}
+					return
+				}
+
+				b = pebble.Batch{}
+				if err := b.SetRepr(buf.Bytes()); err != nil {
+					fmt.Fprintf(stdout, "corrupt log file %q: %v", arg, err)
+					return
+				}
+				fmt.Fprintf(stdout, "%d(%d) seq=%d count=%d\n",
+					offset, len(b.Repr()), b.SeqNum(), b.Count())
+				for r := b.Reader(); ; {
+					kind, ukey, value, ok := r.Next()
+					if !ok {
+						break
+					}
+					fmt.Fprintf(stdout, "    %s(", kind)
+					switch kind {
+					case base.InternalKeyKindDelete:
+						w.fmtKey.fn(stdout, ukey)
+					case base.InternalKeyKindSet:
+						w.fmtKey.fn(stdout, ukey)
+						stdout.Write([]byte{','})
+						w.fmtValue.fn(stdout, value)
+					case base.InternalKeyKindMerge:
+						w.fmtKey.fn(stdout, ukey)
+						stdout.Write([]byte{','})
+						w.fmtValue.fn(stdout, value)
+					case base.InternalKeyKindLogData:
+						w.fmtValue.fn(stdout, ukey)
+					case base.InternalKeyKindRangeDelete:
+						w.fmtKey.fn(stdout, ukey)
+						stdout.Write([]byte{','})
+						w.fmtKey.fn(stdout, value)
+					}
+					fmt.Fprintf(stdout, ")\n")
+				}
+			}
+		}()
+	}
 }
diff --git a/version.go b/version.go
index bf59fed..8561292 100644
--- a/version.go
+++ b/version.go
@@ -43,7 +43,7 @@ func (m *fileMetadata) String() string {
 
 func (m *fileMetadata) tableInfo(dirname string) TableInfo {
 	return TableInfo{
-		Path:           dbFilename(dirname, fileTypeTable, m.fileNum),
+		Path:           base.MakeFilename(dirname, fileTypeTable, m.fileNum),
 		FileNum:        m.fileNum,
 		Size:           m.size,
 		Smallest:       m.smallest,
diff --git a/version_set.go b/version_set.go
index 335dcfe..c8c3b1e 100644
--- a/version_set.go
+++ b/version_set.go
@@ -11,6 +11,7 @@ import (
 	"sync"
 	"sync/atomic"
 
+	"github.com/petermattis/pebble/internal/base"
 	"github.com/petermattis/pebble/internal/record"
 	"github.com/petermattis/pebble/vfs"
 )
@@ -72,7 +73,7 @@ func (vs *versionSet) load(dirname string, opts *Options, mu *sync.Mutex) error
 	vs.nextFileNumber = 2
 
 	// Read the CURRENT file to find the current manifest file.
-	current, err := vs.fs.Open(dbFilename(dirname, fileTypeCurrent, 0))
+	current, err := vs.fs.Open(base.MakeFilename(dirname, fileTypeCurrent, 0))
 	if err != nil {
 		return fmt.Errorf("pebble: could not open CURRENT file for DB %q: %v", dirname, err)
 	}
@@ -210,7 +211,7 @@ func (vs *versionSet) logAndApply(jobID int, ve *versionEdit, dir vfs.File) erro
 				if vs.opts.EventListener.ManifestCreated != nil {
 					vs.opts.EventListener.ManifestCreated(ManifestCreateInfo{
 						JobID:   jobID,
-						Path:    dbFilename(vs.dirname, fileTypeManifest, newManifestFileNumber),
+						Path:    base.MakeFilename(vs.dirname, fileTypeManifest, newManifestFileNumber),
 						FileNum: newManifestFileNumber,
 						Err:     err,
 					})
@@ -252,7 +253,7 @@ func (vs *versionSet) logAndApply(jobID int, ve *versionEdit, dir vfs.File) erro
 			if vs.opts.EventListener.ManifestCreated != nil {
 				vs.opts.EventListener.ManifestCreated(ManifestCreateInfo{
 					JobID:   jobID,
-					Path:    dbFilename(vs.dirname, fileTypeManifest, newManifestFileNumber),
+					Path:    base.MakeFilename(vs.dirname, fileTypeManifest, newManifestFileNumber),
 					FileNum: newManifestFileNumber,
 				})
 			}
@@ -298,7 +299,7 @@ func (vs *versionSet) logAndApply(jobID int, ve *versionEdit, dir vfs.File) erro
 // createManifest creates a manifest file that contains a snapshot of vs.
 func (vs *versionSet) createManifest(dirname string, fileNum uint64) (err error) {
 	var (
-		filename     = dbFilename(dirname, fileTypeManifest, fileNum)
+		filename     = base.MakeFilename(dirname, fileTypeManifest, fileNum)
 		manifestFile vfs.File
 		manifest     *record.Writer
 	)


Test patch:
diff --git a/commit_test.go b/commit_test.go
index 5d0a71d..149fdc0 100644
--- a/commit_test.go
+++ b/commit_test.go
@@ -40,7 +40,7 @@ func (e *testCommitEnv) env() commitEnv {
 
 func (e *testCommitEnv) apply(b *Batch, mem *memTable) error {
 	e.applyBuf.Lock()
-	e.applyBuf.buf = append(e.applyBuf.buf, b.seqNum())
+	e.applyBuf.buf = append(e.applyBuf.buf, b.SeqNum())
 	e.applyBuf.Unlock()
 	return nil
 }
@@ -157,7 +157,7 @@ func BenchmarkCommitPipeline(b *testing.B) {
 				logSeqNum:     new(uint64),
 				visibleSeqNum: new(uint64),
 				apply: func(b *Batch, mem *memTable) error {
-					err := mem.apply(b, b.seqNum())
+					err := mem.apply(b, b.SeqNum())
 					if err != nil {
 						return err
 					}
diff --git a/compaction_test.go b/compaction_test.go
index 47a2499..3416180 100644
--- a/compaction_test.go
+++ b/compaction_test.go
@@ -622,7 +622,7 @@ func TestCompaction(t *testing.T) {
 		v := d.mu.versions.currentVersion()
 		for _, files := range v.files {
 			for _, meta := range files {
-				f, err := mem.Open(dbFilename("", fileTypeTable, meta.fileNum))
+				f, err := mem.Open(base.MakeFilename("", fileTypeTable, meta.fileNum))
 				if err != nil {
 					return "", "", fmt.Errorf("Open: %v", err)
 				}
diff --git a/db_test.go b/db_test.go
index b48e271..6bb6fb7 100644
--- a/db_test.go
+++ b/db_test.go
@@ -17,6 +17,7 @@ import (
 	"time"
 
 	"github.com/petermattis/pebble/cache"
+	"github.com/petermattis/pebble/internal/base"
 	"github.com/petermattis/pebble/vfs"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/exp/rand"
@@ -632,7 +633,7 @@ func TestRollManifest(t *testing.T) {
 	}
 
 	current := func() string {
-		f, err := d.opts.FS.Open(dbFilename(d.dirname, fileTypeCurrent, 0))
+		f, err := d.opts.FS.Open(base.MakeFilename(d.dirname, fileTypeCurrent, 0))
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -675,7 +676,7 @@ func TestRollManifest(t *testing.T) {
 	}
 	var manifests []string
 	for _, filename := range files {
-		fileType, _, ok := parseDBFilename(filename)
+		fileType, _, ok := base.ParseFilename(filename)
 		if !ok {
 			continue
 		}
diff --git a/ingest_test.go b/ingest_test.go
index f98d551..f937070 100644
--- a/ingest_test.go
+++ b/ingest_test.go
@@ -279,7 +279,7 @@ func TestIngestLink(t *testing.T) {
 					t.Fatalf("expected %d files, but found:\n%s", count, strings.Join(files, "\n"))
 				}
 				for j := range files {
-					ftype, fileNum, ok := parseDBFilename(files[j])
+					ftype, fileNum, ok := base.ParseFilename(files[j])
 					if !ok {
 						t.Fatalf("unable to parse filename: %s", files[j])
 					}
diff --git a/filenames_test.go b/internal/base/filenames_test.go
similarity index 80%
rename from filenames_test.go
rename to internal/base/filenames_test.go
index b6af4c6..eb8beb8 100644
--- a/filenames_test.go
+++ b/internal/base/filenames_test.go
@@ -2,14 +2,14 @@
 // of this source code is governed by a BSD-style license that can be found in
 // the LICENSE file.
 
-package pebble
+package base
 
 import (
 	"path/filepath"
 	"testing"
 )
 
-func TestParseDBFilename(t *testing.T) {
+func TestParseFilename(t *testing.T) {
 	testCases := map[string]bool{
 		"000000.log":          true,
 		"000000.log.zip":      false,
@@ -35,7 +35,7 @@ func TestParseDBFilename(t *testing.T) {
 		"OPTIONS-123456.doc":  false,
 	}
 	for tc, want := range testCases {
-		_, _, got := parseDBFilename(filepath.Join("foo", tc))
+		_, _, got := ParseFilename(filepath.Join("foo", tc))
 		if got != want {
 			t.Errorf("%q: got %v, want %v", tc, got, want)
 		}
@@ -43,15 +43,15 @@ func TestParseDBFilename(t *testing.T) {
 }
 
 func TestFilenameRoundTrip(t *testing.T) {
-	testCases := map[fileType]bool{
+	testCases := map[FileType]bool{
 		// CURRENT and LOCK files aren't numbered.
-		fileTypeCurrent: false,
-		fileTypeLock:    false,
+		FileTypeCurrent: false,
+		FileTypeLock:    false,
 		// The remaining file types are numbered.
-		fileTypeLog:      true,
-		fileTypeManifest: true,
-		fileTypeTable:    true,
-		fileTypeOptions:  true,
+		FileTypeLog:      true,
+		FileTypeManifest: true,
+		FileTypeTable:    true,
+		FileTypeOptions:  true,
 	}
 	for fileType, numbered := range testCases {
 		fileNums := []uint64{0}
@@ -59,8 +59,8 @@ func TestFilenameRoundTrip(t *testing.T) {
 			fileNums = []uint64{0, 1, 2, 3, 10, 42, 99, 1001}
 		}
 		for _, fileNum := range fileNums {
-			filename := dbFilename("foo", fileType, fileNum)
-			gotFT, gotFN, gotOK := parseDBFilename(filename)
+			filename := MakeFilename("foo", fileType, fileNum)
+			gotFT, gotFN, gotOK := ParseFilename(filename)
 			if !gotOK {
 				t.Errorf("could not parse %q", filename)
 				continue
diff --git a/internal/record/record_test.go b/internal/record/record_test.go
index 80297f8..990cb67 100644
--- a/internal/record/record_test.go
+++ b/internal/record/record_test.go
@@ -635,6 +635,37 @@ func TestRecoverLastCompleteBlock(t *testing.T) {
 	}
 }
 
+func TestReaderOffset(t *testing.T) {
+	recs, err := makeTestRecords(
+		blockSize*2,
+		400,
+		500,
+		600,
+		700,
+		800,
+		9000,
+		1000,
+	)
+	if err != nil {
+		t.Fatalf("makeTestRecords: %v", err)
+	}
+
+	// The first record should fail, but only when we read deeper beyond the first block.
+	r := NewReader(bytes.NewReader(recs.buf), 0 /* logNum */)
+	for i, offset := range recs.offsets {
+		if offset != r.Offset() {
+			t.Fatalf("%d: expected offset %d, but found %d", i, offset, r.Offset())
+		}
+		rec, err := r.Next()
+		if err != nil {
+			t.Fatalf("Next: %v", err)
+		}
+		if _, err = ioutil.ReadAll(rec); err != nil {
+			t.Fatalf("ReadAll: %v", err)
+		}
+	}
+}
+
 func TestSeekRecord(t *testing.T) {
 	recs, err := makeTestRecords(
 		// The first record will consume 3 entire blocks but a fraction of the 4th.
diff --git a/mem_table_test.go b/mem_table_test.go
index ff98721..d849ac1 100644
--- a/mem_table_test.go
+++ b/mem_table_test.go
@@ -282,7 +282,7 @@ func TestMemTableDeleteRange(t *testing.T) {
 			if err := mem.apply(b, seqNum); err != nil {
 				return err.Error()
 			}
-			seqNum += uint64(b.count())
+			seqNum += uint64(b.Count())
 			return ""
 
 		case "scan":
diff --git a/open_test.go b/open_test.go
index 60926d4..c3d1714 100644
--- a/open_test.go
+++ b/open_test.go
@@ -15,6 +15,7 @@ import (
 	"testing"
 
 	"github.com/kr/pretty"
+	"github.com/petermattis/pebble/internal/base"
 	"github.com/petermattis/pebble/vfs"
 	"github.com/stretchr/testify/require"
 )
@@ -155,7 +156,7 @@ func testOpenCloseOpenClose(t *testing.T, fs vfs.FS, root string) {
 					}
 					var optionsCount int
 					for _, s := range got {
-						if t, _, ok := parseDBFilename(s); ok && t == fileTypeOptions {
+						if t, _, ok := base.ParseFilename(s); ok && t == fileTypeOptions {
 							optionsCount++
 						}
 					}
diff --git a/table_cache_test.go b/table_cache_test.go
index 46d86e2..40cb55c 100644
--- a/table_cache_test.go
+++ b/table_cache_test.go
@@ -79,7 +79,7 @@ func (fs *tableCacheTestFS) validateOpenTables(f func(i, gotO, gotC int) error)
 
 		numStillOpen := 0
 		for i := 0; i < tableCacheTestNumTables; i++ {
-			filename := dbFilename("", fileTypeTable, uint64(i))
+			filename := base.MakeFilename("", fileTypeTable, uint64(i))
 			gotO, gotC := fs.openCounts[filename], fs.closeCounts[filename]
 			if gotO > gotC {
 				numStillOpen++
@@ -109,7 +109,7 @@ func (fs *tableCacheTestFS) validateNoneStillOpen() error {
 		defer fs.mu.Unlock()
 
 		for i := 0; i < tableCacheTestNumTables; i++ {
-			filename := dbFilename("", fileTypeTable, uint64(i))
+			filename := base.MakeFilename("", fileTypeTable, uint64(i))
 			gotO, gotC := fs.openCounts[filename], fs.closeCounts[filename]
 			if gotO != gotC {
 				return fmt.Errorf("i=%d: opened %d times, closed %d times", i, gotO, gotC)
@@ -131,7 +131,7 @@ func newTableCache() (*tableCache, *tableCacheTestFS, error) {
 		FS: vfs.NewMem(),
 	}
 	for i := 0; i < tableCacheTestNumTables; i++ {
-		f, err := fs.Create(dbFilename("", fileTypeTable, uint64(i)))
+		f, err := fs.Create(base.MakeFilename("", fileTypeTable, uint64(i)))
 		if err != nil {
 			return nil, nil, fmt.Errorf("fs.Create: %v", err)
 		}
diff --git a/tool/testdata/sstable_scan b/tool/testdata/sstable_scan
index 20b5e51..27684b8 100644
--- a/tool/testdata/sstable_scan
+++ b/tool/testdata/sstable_scan
@@ -13,14 +13,14 @@ art#0,1 [36]
 artery#0,1 [31]
 article#0,1 [31]
 articles#0,1 [31]
-as#0,1 [35 36]
+as#0,1 [3536]
 
 sstable scan
 --end=abused
 ../sstable/testdata/h.sst
 ----
 ../sstable/testdata/h.sst
-a#0,1 [39 37]
+a#0,1 [3937]
 aboard#0,1 [32]
 about#0,1 [32]
 above#0,1 [31]
@@ -32,9 +32,9 @@ sstable scan
 ../sstable/testdata/h.sst
 ----
 ../sstable/testdata/h.sst
-you#0,1 [31 31 30]
+you#0,1 [313130]
 young#0,1 [36]
-your#0,1 [34 39]
+your#0,1 [3439]
 yourself#0,1 [37]
 youth#0,1 [35]
 
diff --git a/tool/testdata/wal_dump b/tool/testdata/wal_dump
new file mode 100644
index 0000000..3e8798e
--- /dev/null
+++ b/tool/testdata/wal_dump
@@ -0,0 +1,58 @@
+wal dump
+----
+requires at least 1 arg(s), only received 0
+
+wal dump
+../testdata/db-stage-2/000003.log
+----
+../testdata/db-stage-2/000003.log
+0(21) seq=1 count=1
+    SET(foo,<3>)
+28(21) seq=2 count=1
+    SET(bar,<3>)
+56(23) seq=3 count=1
+    SET(baz,<5>)
+86(22) seq=4 count=1
+    SET(foo,<4>)
+115(17) seq=5 count=1
+    DEL(bar)
+EOF
+
+wal dump
+../testdata/db-stage-4/000006.log
+----
+../testdata/db-stage-4/000006.log
+0(22) seq=6 count=1
+    SET(foo,<4>)
+29(22) seq=7 count=1
+    SET(quux,<3>)
+58(17) seq=8 count=1
+    DEL(baz)
+EOF
+
+wal dump
+../testdata/db-stage-4/000006.log
+--key=%x
+--value=%x
+----
+../testdata/db-stage-4/000006.log
+0(22) seq=6 count=1
+    SET(666f6f,66697665)
+29(22) seq=7 count=1
+    SET(71757578,736978)
+58(17) seq=8 count=1
+    DEL(62617a)
+EOF
+
+wal dump
+../testdata/db-stage-4/000006.log
+--value=quoted
+----
+../testdata/db-stage-4/000006.log
+0(22) seq=6 count=1
+    SET(foo,five)
+29(22) seq=7 count=1
+    SET(quux,six)
+58(17) seq=8 count=1
+    DEL(baz)
+EOF
diff --git a/tool/wal_test.go b/tool/wal_test.go
new file mode 100644
index 0000000..3255393
--- /dev/null
+++ b/tool/wal_test.go
@@ -0,0 +1,13 @@
+// Copyright 2019 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package tool
+
+import (
+	"testing"
+)
+
+func TestWAL(t *testing.T) {
+	runTests(t, "testdata/wal_*")
+}