Repository: petermattis/pebble
Pull request #80 (2019-04-18 19:22:29)

Add support for the "recyclable" log record format
The recyclable log record format associated a log file number with each
record in a file which allows a log file to be reused safely without
truncating the file. This is an important performance optimization in
itself and a prerequisite for writing the WAL using direct IO.

Changed `record.LogWriter` to always use the recyclable log record
format which in turn changes Pebble to use the recyclable format for the
WAL. `record.Writer` (used to write the `MANIFEST`) continues to use the
legacy log record format. `record.Reader` can now read either format and
will return an error when a record with a log number does not match the
supplied log number.

Took the opportunity to unexport some methods that were not being used
outside of the `internal/record` package such as `Reader.SeekRecord` and
`LogWriter.Flush`.

Actual reuse of WAL files will be implemented in a future change.

See #41

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/80)
<!-- Reviewable:end -->


Patch:
diff --git a/db.go b/db.go
index 616f070..6973ca6 100644
--- a/db.go
+++ b/db.go
@@ -764,7 +764,7 @@ func (d *DB) makeRoomForWrite(b *Batch) error {
 		// have been applied.
 		if !d.opts.DisableWAL {
 			d.mu.log.number = newLogNumber
-			d.mu.log.LogWriter = record.NewLogWriter(newLogFile)
+			d.mu.log.LogWriter = record.NewLogWriter(newLogFile, newLogNumber)
 		}
 		imm := d.mu.mem.mutable
 		if imm.empty() {
diff --git a/internal/record/log_writer.go b/internal/record/log_writer.go
index 22cc703..d67911f 100644
--- a/internal/record/log_writer.go
+++ b/internal/record/log_writer.go
@@ -22,7 +22,18 @@ type block struct {
 	buf     [blockSize]byte
 }
 
-// LogWriter writes records to an underlying io.Writer.
+type flusher interface {
+	Flush() error
+}
+
+type syncer interface {
+	Sync() error
+}
+
+// LogWriter writes records to an underlying io.Writer. In order to support WAL
+// file reuse, a LogWriter's records are tagged with the WAL's file
+// number. When reading a log file a record from a previous incarnation of the
+// file will return the error ErrInvalidLogNum.
 type LogWriter struct {
 	// w is the underlying writer.
 	w io.Writer
@@ -32,8 +43,10 @@ type LogWriter struct {
 	f flusher
 	// s is w as a syncer.
 	s syncer
-	// blockNumber is the zero based block number for the current block.
-	blockNumber int64
+	// logNum is the low 32-bits of the log's file number.
+	logNum uint32
+	// blockNum is the zero based block number for the current block.
+	blockNum int64
 	// err is any accumulated error. TODO(peter): This needs to be protected in
 	// some fashion. Perhaps using atomic.Value.
 	err error
@@ -62,16 +75,21 @@ type LogWriter struct {
 }
 
 // NewLogWriter returns a new LogWriter.
-func NewLogWriter(w io.Writer) *LogWriter {
+func NewLogWriter(w io.Writer, logNum uint64) *LogWriter {
 	c, _ := w.(io.Closer)
 	f, _ := w.(flusher)
 	s, _ := w.(syncer)
 	r := &LogWriter{
-		w:    w,
-		c:    c,
-		f:    f,
-		s:    s,
-		free: make(chan *block, 4),
+		w: w,
+		c: c,
+		f: f,
+		s: s,
+		// NB: we truncate the 64-bit log number to 32-bits. This is ok because a)
+		// we are very unlikely to reach a file number of 4 billion and b) the log
+		// number is used as a validation check and using only the low 32-bits is
+		// sufficient for that purpose.
+		logNum: uint32(logNum),
+		free:   make(chan *block, 4),
 	}
 	for i := 0; i < cap(r.free); i++ {
 		r.free <- &block{}
@@ -152,7 +170,7 @@ func (w *LogWriter) queueBlock() {
 	w.err = w.flusher.err
 	f.Unlock()
 
-	w.blockNumber++
+	w.blockNum++
 }
 
 // Close flushes any unwritten data and closes the writer.
@@ -181,14 +199,6 @@ func (w *LogWriter) closed() bool {
 	return closed
 }
 
-// Flush flushes unwritten data. May be called concurrently with Write, Sync
-// and itself.
-func (w *LogWriter) Flush() error {
-	w.flushMu.Lock()
-	defer w.flushMu.Unlock()
-	return w.flushLocked()
-}
-
 func (w *LogWriter) flushLocked() error {
 	if w.err != nil {
 		if w.closed() {
@@ -277,7 +287,7 @@ func (w *LogWriter) WriteRecord(p []byte) (int64, error) {
 		p = w.emitFragment(i, p)
 	}
 
-	offset := w.blockNumber*blockSize + int64(w.block.written)
+	offset := w.blockNum*blockSize + int64(w.block.written)
 	return offset, w.err
 }
 
@@ -285,29 +295,31 @@ func (w *LogWriter) emitFragment(n int, p []byte) []byte {
 	b := w.block
 	i := b.written
 	first := n == 0
-	last := blockSize-i-headerSize >= int32(len(p))
+	last := blockSize-i-recyclableHeaderSize >= int32(len(p))
 
 	if last {
 		if first {
-			b.buf[i+6] = fullChunkType
+			b.buf[i+6] = recyclableFullChunkType
 		} else {
-			b.buf[i+6] = lastChunkType
+			b.buf[i+6] = recyclableLastChunkType
 		}
 	} else {
 		if first {
-			b.buf[i+6] = firstChunkType
+			b.buf[i+6] = recyclableFirstChunkType
 		} else {
-			b.buf[i+6] = middleChunkType
+			b.buf[i+6] = recyclableMiddleChunkType
 		}
 	}
 
-	r := copy(b.buf[i+headerSize:], p)
-	j := i + int32(headerSize+r)
+	binary.LittleEndian.PutUint32(b.buf[i+7:i+11], w.logNum)
+
+	r := copy(b.buf[i+recyclableHeaderSize:], p)
+	j := i + int32(recyclableHeaderSize+r)
 	binary.LittleEndian.PutUint32(b.buf[i+0:i+4], crc.New(b.buf[i+6:j]).Value())
 	binary.LittleEndian.PutUint16(b.buf[i+4:i+6], uint16(r))
 	atomic.StoreInt32(&b.written, j)
 
-	if blockSize-b.written < headerSize {
+	if blockSize-b.written < recyclableHeaderSize {
 		// There is no room for another fragment in the block, so fill the
 		// remaining bytes with zeros and queue the block for flushing.
 		for i := b.written; i < blockSize; i++ {
diff --git a/internal/record/record.go b/internal/record/record.go
index c6b4c0d..1bd6970 100644
--- a/internal/record/record.go
+++ b/internal/record/record.go
@@ -62,14 +62,37 @@
 // boundaries. The last block may be shorter than 32 KiB. Any unused bytes in a
 // block must be zero.
 //
-// A record maps to one or more chunks. Each chunk has a 7 byte header (a 4
-// byte checksum, a 2 byte little-endian uint16 length, and a 1 byte chunk type)
-// followed by a payload. The checksum is over the chunk type and the payload.
+// A record maps to one or more chunks. There are two chunk formats: legacy and
+// recyclable. The legacy chunk format:
+//
+//   +----------+-----------+-----------+--- ... ---+
+//   | CRC (4B) | Size (2B) | Type (1B) | Payload   |
+//   +----------+-----------+-----------+--- ... ---+
+//
+// CRC is computed over the type and payload
+// Size is the length of the payload in bytes
+// Type is the chunk type
 //
 // There are four chunk types: whether the chunk is the full record, or the
 // first, middle or last chunk of a multi-chunk record. A multi-chunk record
 // has one first chunk, zero or more middle chunks, and one last chunk.
 //
+// The recyclyable chunk format is similar to the legacy format, but extends
+// the chunk header with an additional log number field. This allows reuse
+// (recycling) of log files which can provide significantly better performance
+// when syncing frequently as it avoids needing to update the file
+// metadata. Additionally, recycling log files is a prequisite for using direct
+// IO with log writing. The recyclyable format is:
+//
+//   +----------+-----------+-----------+----------------+--- ... ---+
+//   | CRC (4B) | Size (2B) | Type (1B) | Log number (4B)| Payload   |
+//   +----------+-----------+-----------+----------------+--- ... ---+
+//
+// Recyclable chunks are distinguished from legacy chunks by the addition of 4
+// extra "recyclable" chunk types that map directly to the legacy chunk types
+// (i.e. full, first, middle, last). The CRC is computed over the type, log
+// number, and payload.
+//
 // The wire format allows for limited recovery in the face of data corruption:
 // on a format error (such as a checksum mismatch), the reader moves to the
 // next block and looks for the next full or first chunk.
@@ -94,12 +117,18 @@ const (
 	firstChunkType  = 2
 	middleChunkType = 3
 	lastChunkType   = 4
+
+	recyclableFullChunkType   = 5
+	recyclableFirstChunkType  = 6
+	recyclableMiddleChunkType = 7
+	recyclableLastChunkType   = 8
 )
 
 const (
-	blockSize     = 32 * 1024
-	blockSizeMask = blockSize - 1
-	headerSize    = 7
+	blockSize            = 32 * 1024
+	blockSizeMask        = blockSize - 1
+	legacyHeaderSize     = 7
+	recyclableHeaderSize = legacyHeaderSize + 4
 )
 
 var (
@@ -108,25 +137,24 @@ var (
 
 	// ErrNoLastRecord is returned if LastRecordOffset is called and there is no previous record.
 	ErrNoLastRecord = errors.New("pebble/record: no last record exists")
-)
 
-type flusher interface {
-	Flush() error
-}
-
-type syncer interface {
-	Sync() error
-}
+	// ErrInvalidLogNum is returned if the log number in a record differs from
+	// the one expected.
+	ErrInvalidLogNum = errors.New("pebble/record: invalid log number")
+)
 
 // Reader reads records from an underlying io.Reader.
 type Reader struct {
 	// r is the underlying reader.
 	r io.Reader
+	// 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
 	// seq is the sequence number of the current record.
 	seq int
-	// buf[i:j] is the unread portion of the current chunk's payload.
-	// The low bound, i, excludes the chunk header.
-	i, j int
+	// buf[begin:end] is the unread portion of the current chunk's payload. The
+	// low bound, begin, excludes the chunk header.
+	begin, end int
 	// n is the number of bytes of buf that are valid. Once reading has started,
 	// only the final block can have n < blockSize.
 	n int
@@ -142,10 +170,13 @@ type Reader struct {
 	buf [blockSize]byte
 }
 
-// NewReader returns a new reader.
-func NewReader(r io.Reader) *Reader {
+// NewReader returns a new reader. If the file contains records encoded using
+// the recyclable record format, then the log number in those records must
+// match the specifed logNum.
+func NewReader(r io.Reader, logNum uint64) *Reader {
 	return &Reader{
-		r: r,
+		r:      r,
+		logNum: uint32(logNum),
 	}
 }
 
@@ -153,40 +184,60 @@ func NewReader(r io.Reader) *Reader {
 // next block into the buffer if necessary.
 func (r *Reader) nextChunk(wantFirst bool) error {
 	for {
-		if r.j+headerSize <= r.n {
-			checksum := binary.LittleEndian.Uint32(r.buf[r.j+0 : r.j+4])
-			length := binary.LittleEndian.Uint16(r.buf[r.j+4 : r.j+6])
-			chunkType := r.buf[r.j+6]
+		if r.end+legacyHeaderSize <= r.n {
+			checksum := binary.LittleEndian.Uint32(r.buf[r.end+0 : r.end+4])
+			length := binary.LittleEndian.Uint16(r.buf[r.end+4 : r.end+6])
+			chunkType := r.buf[r.end+6]
 
 			if checksum == 0 && length == 0 && chunkType == 0 {
+				if r.end+recyclableHeaderSize > r.n {
+					// Skip the rest of the block if the recyclable header size does not
+					// fit within it.
+					r.end = r.n
+					continue
+				}
 				if wantFirst || r.recovering {
 					// Skip the rest of the block, if it looks like it is all
-					// zeroes. This is common if the record file was created
-					// via mmap.
+					// zeroes. This is common if the record file was created via mmap.
 					//
 					// Set r.err to be an error so r.Recover actually recovers.
 					r.err = errors.New("pebble/record: block appears to be zeroed")
 					if !r.recovering {
 						return r.err
 					}
-					r.Recover()
+					r.recover()
 					continue
 				}
 				return errors.New("pebble/record: invalid chunk")
 			}
 
-			r.i = r.j + headerSize
-			r.j = r.j + headerSize + int(length)
-			if r.j > r.n {
+			headerSize := legacyHeaderSize
+			if chunkType >= recyclableFullChunkType && chunkType <= recyclableLastChunkType {
+				headerSize = recyclableHeaderSize
+				if r.end+headerSize > r.n {
+					return errors.New("pebble/record: invalid chunk (header overflows block)")
+				}
+
+				logNum := binary.LittleEndian.Uint32(r.buf[r.end+7 : r.end+11])
+				if logNum != r.logNum {
+					return ErrInvalidLogNum
+				}
+
+				chunkType -= (recyclableFullChunkType - 1)
+			}
+
+			r.begin = r.end + headerSize
+			r.end = r.begin + int(length)
+			if r.end > r.n {
 				if r.recovering {
-					r.Recover()
+					r.recover()
 					continue
 				}
 				return errors.New("pebble/record: invalid chunk (length overflows block)")
 			}
-			if checksum != crc.New(r.buf[r.i-1:r.j]).Value() {
+			if checksum != crc.New(r.buf[r.begin-headerSize+6:r.end]).Value() {
 				if r.recovering {
-					r.Recover()
+					r.recover()
 					continue
 				}
 				return errors.New("pebble/record: invalid chunk (checksum mismatch)")
@@ -201,7 +252,7 @@ func (r *Reader) nextChunk(wantFirst bool) error {
 			return nil
 		}
 		if r.n < blockSize && r.started {
-			if r.j != r.n {
+			if r.end != r.n {
 				return io.ErrUnexpectedEOF
 			}
 			return io.EOF
@@ -210,7 +261,7 @@ func (r *Reader) nextChunk(wantFirst bool) error {
 		if err != nil && err != io.ErrUnexpectedEOF {
 			return err
 		}
-		r.i, r.j, r.n = 0, 0, n
+		r.begin, r.end, r.n = 0, 0, n
 	}
 }
 
@@ -222,7 +273,7 @@ func (r *Reader) Next() (io.Reader, error) {
 	if r.err != nil {
 		return nil, r.err
 	}
-	r.i = r.j
+	r.begin = r.end
 	r.err = r.nextChunk(true)
 	if r.err != nil {
 		return nil, r.err
@@ -231,24 +282,24 @@ func (r *Reader) Next() (io.Reader, error) {
 	return singleReader{r, r.seq}, nil
 }
 
-// Recover clears any errors read so far, so that calling Next will start
+// 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
-// recently returned by Next, as stale. If Recover is called without any
-// prior error, then Recover is a no-op.
-func (r *Reader) Recover() {
+// will return io.EOF. recover also marks the current reader, the one most
+// recently returned by Next, as stale. If recover is called without any
+// prior error, then recover is a no-op.
+func (r *Reader) recover() {
 	if r.err == nil {
 		return
 	}
 	r.recovering = true
 	r.err = nil
 	// Discard the rest of the current block.
-	r.i, r.j, r.last = r.n, r.n, false
+	r.begin, r.end, r.last = r.n, r.n, false
 	// Invalidate any outstanding singleReader.
 	r.seq++
 }
 
-// SeekRecord seeks in the underlying io.Reader such that calling r.Next
+// seekRecord seeks in the underlying io.Reader such that calling r.Next
 // returns the record whose first chunk header starts at the provided offset.
 // Its behavior is undefined if the argument given is not such an offset, as
 // the bytes at that offset may coincidentally appear to be a valid header.
@@ -256,17 +307,17 @@ func (r *Reader) Recover() {
 // It returns ErrNotAnIOSeeker if the underlying io.Reader does not implement
 // io.Seeker.
 //
-// SeekRecord will fail and return an error if the Reader previously
+// seekRecord will fail and return an error if the Reader previously
 // encountered an error, including io.EOF. Such errors can be cleared by
-// calling Recover. Calling SeekRecord after Recover will make calling Next
+// calling Recover. Calling seekRecord after Recover will make calling Next
 // return the record at the given offset, instead of the record at the next
-// good 32KiB block as Recover normally would. Calling SeekRecord before
+// good 32KiB block as Recover normally would. Calling seekRecord before
 // Recover has no effect on Recover's semantics other than changing the
 // starting point for determining the next good 32KiB block.
 //
 // The offset is always relative to the start of the underlying io.Reader, so
 // negative values will result in an error as per io.Seeker.
-func (r *Reader) SeekRecord(offset int64) error {
+func (r *Reader) seekRecord(offset int64) error {
 	r.seq++
 	if r.err != nil {
 		return r.err
@@ -284,7 +335,7 @@ func (r *Reader) SeekRecord(offset int64) error {
 	}
 
 	// Clear the state of the internal reader.
-	r.i, r.j, r.n = 0, 0, 0
+	r.begin, r.end, r.n = 0, 0, 0
 	r.started, r.recovering, r.last = false, false, false
 	if r.err = r.nextChunk(false); r.err != nil {
 		return r.err
@@ -292,7 +343,7 @@ func (r *Reader) SeekRecord(offset int64) error {
 
 	// Now skip to the offset requested within the block. A subsequent
 	// call to Next will return the block at the requested offset.
-	r.i, r.j = c, c
+	r.begin, r.end = c, c
 
 	return nil
 }
@@ -310,7 +361,7 @@ func (x singleReader) Read(p []byte) (int, error) {
 	if r.err != nil {
 		return 0, r.err
 	}
-	for r.i == r.j {
+	for r.begin == r.end {
 		if r.last {
 			return 0, io.EOF
 		}
@@ -318,8 +369,8 @@ func (x singleReader) Read(p []byte) (int, error) {
 			return 0, r.err
 		}
 	}
-	n := copy(p, r.buf[r.i:r.j])
-	r.i += n
+	n := copy(p, r.buf[r.begin:r.end])
+	r.begin += n
 	return n, nil
 }
 
@@ -378,7 +429,7 @@ func NewWriter(w io.Writer) *Writer {
 
 // fillHeader fills in the header for the pending chunk.
 func (w *Writer) fillHeader(last bool) {
-	if w.i+headerSize > w.j || w.j > blockSize {
+	if w.i+legacyHeaderSize > w.j || w.j > blockSize {
 		panic("pebble/record: bad writer state")
 	}
 	if last {
@@ -395,7 +446,7 @@ func (w *Writer) fillHeader(last bool) {
 		}
 	}
 	binary.LittleEndian.PutUint32(w.buf[w.i+0:w.i+4], crc.New(w.buf[w.i+6:w.j]).Value())
-	binary.LittleEndian.PutUint16(w.buf[w.i+4:w.i+6], uint16(w.j-w.i-headerSize))
+	binary.LittleEndian.PutUint16(w.buf[w.i+4:w.i+6], uint16(w.j-w.i-legacyHeaderSize))
 }
 
 // writeBlock writes the buffered block to the underlying writer, and reserves
@@ -403,7 +454,7 @@ func (w *Writer) fillHeader(last bool) {
 func (w *Writer) writeBlock() {
 	_, w.err = w.w.Write(w.buf[w.written:])
 	w.i = 0
-	w.j = headerSize
+	w.j = legacyHeaderSize
 	w.written = 0
 	w.blockNumber++
 }
@@ -459,7 +510,7 @@ func (w *Writer) Next() (io.Writer, error) {
 		w.fillHeader(true)
 	}
 	w.i = w.j
-	w.j = w.j + headerSize
+	w.j = w.j + legacyHeaderSize
 	// Check if there is room in the block for the header.
 	if w.j > blockSize {
 		// Fill in the rest of the block with zeroes.
diff --git a/open.go b/open.go
index 57b6f13..5e41e83 100644
--- a/open.go
+++ b/open.go
@@ -146,7 +146,7 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 		return logFiles[i].num < logFiles[j].num
 	})
 	for _, lf := range logFiles {
-		maxSeqNum, err := d.replayWAL(&ve, fs, filepath.Join(dirname, lf.name))
+		maxSeqNum, err := d.replayWAL(&ve, fs, filepath.Join(dirname, lf.name), lf.num)
 		if err != nil {
 			return nil, err
 		}
@@ -168,7 +168,7 @@ func Open(dirname string, opts *db.Options) (*DB, error) {
 		BytesPerSync:    d.opts.BytesPerSync,
 		PreallocateSize: d.walPreallocateSize(),
 	})
-	d.mu.log.LogWriter = record.NewLogWriter(logFile)
+	d.mu.log.LogWriter = record.NewLogWriter(logFile, d.mu.log.number)
 
 	// Write a new manifest to disk.
 	if err := d.mu.versions.logAndApply(&ve); err != nil {
@@ -204,6 +204,7 @@ func (d *DB) replayWAL(
 	ve *versionEdit,
 	fs storage.Storage,
 	filename string,
+	logNum uint64,
 ) (maxSeqNum uint64, err error) {
 	file, err := fs.Open(filename)
 	if err != nil {
@@ -215,7 +216,7 @@ func (d *DB) replayWAL(
 		b   Batch
 		buf bytes.Buffer
 		mem *memTable
-		rr  = record.NewReader(file)
+		rr  = record.NewReader(file, logNum)
 	)
 	for {
 		r, err := rr.Next()
diff --git a/version_set.go b/version_set.go
index ca0a6b8..c18262a 100644
--- a/version_set.go
+++ b/version_set.go
@@ -96,7 +96,7 @@ func (vs *versionSet) load(dirname string, opts *db.Options, mu *sync.Mutex) err
 		return fmt.Errorf("pebble: could not open manifest file %q for DB %q: %v", b, dirname, err)
 	}
 	defer manifest.Close()
-	rr := record.NewReader(manifest)
+	rr := record.NewReader(manifest, 0 /* logNum */)
 	for {
 		r, err := rr.Next()
 		if err == io.EOF {


Test patch:
diff --git a/commit_test.go b/commit_test.go
index 3ed3476..e4cb382 100644
--- a/commit_test.go
+++ b/commit_test.go
@@ -131,7 +131,7 @@ func BenchmarkCommitPipeline(b *testing.B) {
 		b.Run(fmt.Sprintf("parallel=%d", parallelism), func(b *testing.B) {
 			b.SetParallelism(parallelism)
 			mem := newMemTable(nil)
-			wal := record.NewLogWriter(ioutil.Discard)
+			wal := record.NewLogWriter(ioutil.Discard, 0 /* logNum */)
 
 			nullCommitEnv := commitEnv{
 				logSeqNum:     new(uint64),
diff --git a/internal/record/record_test.go b/internal/record/record_test.go
index a9e9879..b41fc28 100644
--- a/internal/record/record_test.go
+++ b/internal/record/record_test.go
@@ -37,7 +37,7 @@ func testGeneratorWriter(
 	t *testing.T,
 	reset func(),
 	gen func() (string, bool),
-	newWriter func (io.Writer) recordWriter,
+	newWriter func(io.Writer) recordWriter,
 ) {
 	buf := new(bytes.Buffer)
 
@@ -55,9 +55,8 @@ func testGeneratorWriter(
 	if err := w.Close(); err != nil {
 		t.Fatalf("Close: %v", err)
 	}
-
 	reset()
-	r := NewReader(buf)
+	r := NewReader(buf, 0 /* logNum */)
 	for {
 		s, ok := gen()
 		if !ok {
@@ -80,16 +79,16 @@ func testGeneratorWriter(
 	}
 }
 
-func testGenerator( t *testing.T, reset func(), gen func() (string, bool)) {
-	t.Run("Writer", func (t *testing.T) {
-		testGeneratorWriter(t, reset, gen, func (w io.Writer) recordWriter {
+func testGenerator(t *testing.T, reset func(), gen func() (string, bool)) {
+	t.Run("Writer", func(t *testing.T) {
+		testGeneratorWriter(t, reset, gen, func(w io.Writer) recordWriter {
 			return NewWriter(w)
 		})
 	})
 
-	t.Run("LogWriter", func (t *testing.T) {
-		testGeneratorWriter(t, reset, gen, func (w io.Writer) recordWriter {
-			return NewLogWriter(w)
+	t.Run("LogWriter", func(t *testing.T) {
+		testGeneratorWriter(t, reset, gen, func(w io.Writer) recordWriter {
+			return NewLogWriter(w, 0 /* logNum */)
 		})
 	})
 }
@@ -217,7 +216,7 @@ func TestFlush(t *testing.T) {
 		t.Fatalf("buffer length #5: got %d want %d", got, want)
 	}
 	// Check that reading those records give the right lengths.
-	r := NewReader(buf)
+	r := NewReader(buf, 0 /* logNum */)
 	wants := []int64{1, 2, 10000, 40000}
 	for i, want := range wants {
 		rr, _ := r.Next()
@@ -247,7 +246,7 @@ func TestNonExhaustiveRead(t *testing.T) {
 		t.Fatalf("Close: %v", err)
 	}
 
-	r := NewReader(buf)
+	r := NewReader(buf, 0 /* logNum */)
 	for i := 0; i < n; i++ {
 		rr, _ := r.Next()
 		_, err := io.ReadFull(rr, p)
@@ -275,7 +274,7 @@ func TestStaleReader(t *testing.T) {
 		t.Fatalf("Close: %v\n", err)
 	}
 
-	r := NewReader(buf)
+	r := NewReader(buf, 0 /* logNum */)
 	r0, err := r.Next()
 	if err != nil {
 		t.Fatalf("reader.Next: %v", err)
@@ -366,36 +365,36 @@ func corruptBlock(buf []byte, blockNum int) {
 
 func TestRecoverNoOp(t *testing.T) {
 	recs, err := makeTestRecords(
-		blockSize-headerSize,
-		blockSize-headerSize,
-		blockSize-headerSize,
+		blockSize-legacyHeaderSize,
+		blockSize-legacyHeaderSize,
+		blockSize-legacyHeaderSize,
 	)
 	if err != nil {
 		t.Fatalf("makeTestRecords: %v", err)
 	}
 
-	r := NewReader(bytes.NewReader(recs.buf))
+	r := NewReader(bytes.NewReader(recs.buf), 0 /* logNum */)
 	_, err = r.Next()
 	if err != nil || r.err != nil {
 		t.Fatalf("reader.Next: %v reader.err: %v", err, r.err)
 	}
 
-	seq, i, j, n := r.seq, r.i, r.j, r.n
+	seq, begin, end, n := r.seq, r.begin, r.end, r.n
 
 	// Should be a no-op since r.err == nil.
-	r.Recover()
+	r.recover()
 
 	// r.err was nil, nothing should have changed.
-	if seq != r.seq || i != r.i || j != r.j || n != r.n {
+	if seq != r.seq || begin != r.begin || end != r.end || n != r.n {
 		t.Fatal("reader.Recover when no error existed, was not a no-op")
 	}
 }
 
 func TestBasicRecover(t *testing.T) {
 	recs, err := makeTestRecords(
-		blockSize-headerSize,
-		blockSize-headerSize,
-		blockSize-headerSize,
+		blockSize-legacyHeaderSize,
+		blockSize-legacyHeaderSize,
+		blockSize-legacyHeaderSize,
 	)
 	if err != nil {
 		t.Fatalf("makeTestRecords: %v", err)
@@ -405,7 +404,7 @@ func TestBasicRecover(t *testing.T) {
 	corruptBlock(recs.buf, 1)
 
 	underlyingReader := bytes.NewReader(recs.buf)
-	r := NewReader(underlyingReader)
+	r := NewReader(underlyingReader, 0 /* logNum */)
 
 	// The first record r0 should be read just fine.
 	r0, err := r.Next()
@@ -430,7 +429,7 @@ func TestBasicRecover(t *testing.T) {
 	}
 
 	// Recover from that checksum mismatch.
-	r.Recover()
+	r.recover()
 	currentOffset, err := underlyingReader.Seek(0, os.SEEK_CUR)
 	if err != nil {
 		t.Fatalf("current offset: %v", err)
@@ -458,7 +457,7 @@ func TestRecoverSingleBlock(t *testing.T) {
 	// a 7 byte header, the first record will roll over into 4 blocks.
 	recs, err := makeTestRecords(
 		blockSize*3,
-		blockSize-headerSize,
+		blockSize-legacyHeaderSize,
 		blockSize/2,
 	)
 	if err != nil {
@@ -471,7 +470,7 @@ func TestRecoverSingleBlock(t *testing.T) {
 
 	// The first record should fail, but only when we read deeper beyond the
 	// first block.
-	r := NewReader(bytes.NewReader(recs.buf))
+	r := NewReader(bytes.NewReader(recs.buf), 0 /* logNum */)
 	r0, err := r.Next()
 	if err != nil {
 		t.Fatalf("Next: %v", err)
@@ -487,7 +486,7 @@ func TestRecoverSingleBlock(t *testing.T) {
 	}
 
 	// Recover from that checksum mismatch.
-	r.Recover()
+	r.recover()
 
 	// All of the data in the second record r1 is lost because the first record
 	// r0 shared a partial block with it. The second record also overlapped
@@ -509,11 +508,11 @@ func TestRecoverMultipleBlocks(t *testing.T) {
 		// The first record will consume 3 entire blocks but a fraction of the 4th.
 		blockSize*3,
 		// The second record will completely fill the remainder of the 4th block.
-		3*(blockSize-headerSize)-2*blockSize-2*headerSize,
+		3*(blockSize-legacyHeaderSize)-2*blockSize-2*legacyHeaderSize,
 		// Consume the entirety of the 5th block.
-		blockSize-headerSize,
+		blockSize-legacyHeaderSize,
 		// Consume the entirety of the 6th block.
-		blockSize-headerSize,
+		blockSize-legacyHeaderSize,
 		// Consume roughly half of the 7th block.
 		blockSize/2,
 	)
@@ -529,7 +528,7 @@ func TestRecoverMultipleBlocks(t *testing.T) {
 	corruptBlock(recs.buf, 5)
 
 	// The first record should fail, but only when we read deeper beyond the first block.
-	r := NewReader(bytes.NewReader(recs.buf))
+	r := NewReader(bytes.NewReader(recs.buf), 0 /* logNum */)
 	r0, err := r.Next()
 	if err != nil {
 		t.Fatalf("Next: %v", err)
@@ -545,7 +544,7 @@ func TestRecoverMultipleBlocks(t *testing.T) {
 	}
 
 	// Recover from that checksum mismatch.
-	r.Recover()
+	r.recover()
 
 	// All of the data in the second record is lost because the first
 	// record shared a partial block with it. The following two records
@@ -566,7 +565,7 @@ func TestRecoverMultipleBlocks(t *testing.T) {
 // last record will be corrupted. It will then try Recover and verify that EOF
 // is returned.
 func verifyLastBlockRecover(recs *testRecords) error {
-	r := NewReader(bytes.NewReader(recs.buf))
+	r := NewReader(bytes.NewReader(recs.buf), 0 /* logNum */)
 	// Loop to one element larger than the number of records to verify EOF.
 	for i := 0; i < len(recs.records)+1; i++ {
 		_, err := r.Next()
@@ -575,7 +574,7 @@ func verifyLastBlockRecover(recs *testRecords) error {
 			if err == nil {
 				return errors.New("Expected a checksum mismatch error, got nil")
 			}
-			r.Recover()
+			r.recover()
 		case len(recs.records):
 			if err != io.EOF {
 				return fmt.Errorf("Expected io.EOF, got %v", err)
@@ -594,7 +593,7 @@ func TestRecoverLastPartialBlock(t *testing.T) {
 		// The first record will consume 3 entire blocks but a fraction of the 4th.
 		blockSize*3,
 		// The second record will completely fill the remainder of the 4th block.
-		3*(blockSize-headerSize)-2*blockSize-2*headerSize,
+		3*(blockSize-legacyHeaderSize)-2*blockSize-2*legacyHeaderSize,
 		// Consume roughly half of the 5th block.
 		blockSize/2,
 	)
@@ -616,9 +615,9 @@ func TestRecoverLastCompleteBlock(t *testing.T) {
 		// The first record will consume 3 entire blocks but a fraction of the 4th.
 		blockSize*3,
 		// The second record will completely fill the remainder of the 4th block.
-		3*(blockSize-headerSize)-2*blockSize-2*headerSize,
+		3*(blockSize-legacyHeaderSize)-2*blockSize-2*legacyHeaderSize,
 		// Consume the entire 5th block.
-		blockSize-headerSize,
+		blockSize-legacyHeaderSize,
 	)
 	if err != nil {
 		t.Fatalf("makeTestRecords: %v", err)
@@ -638,11 +637,11 @@ func TestSeekRecord(t *testing.T) {
 		// The first record will consume 3 entire blocks but a fraction of the 4th.
 		blockSize*3,
 		// The second record will completely fill the remainder of the 4th block.
-		3*(blockSize-headerSize)-2*blockSize-2*headerSize,
+		3*(blockSize-legacyHeaderSize)-2*blockSize-2*legacyHeaderSize,
 		// Consume the entirety of the 5th block.
-		blockSize-headerSize,
+		blockSize-legacyHeaderSize,
 		// Consume the entirety of the 6th block.
-		blockSize-headerSize,
+		blockSize-legacyHeaderSize,
 		// Consume roughly half of the 7th block.
 		blockSize/2,
 	)
@@ -650,10 +649,10 @@ func TestSeekRecord(t *testing.T) {
 		t.Fatalf("makeTestRecords: %v", err)
 	}
 
-	r := NewReader(bytes.NewReader(recs.buf))
+	r := NewReader(bytes.NewReader(recs.buf), 0 /* logNum */)
 	// Seek to a valid block offset, but within a multiblock record. This should cause the next call to
 	// Next after SeekRecord to return the next valid FIRST/FULL chunk of the subsequent record.
-	err = r.SeekRecord(blockSize)
+	err = r.seekRecord(blockSize)
 	if err != nil {
 		t.Fatalf("SeekRecord: %v", err)
 	}
@@ -668,17 +667,17 @@ func TestSeekRecord(t *testing.T) {
 
 	// Seek 3 bytes into the second block, which is still in the middle of the first record, but not
 	// at a valid chunk boundary. Should result in an error upon calling r.Next.
-	err = r.SeekRecord(blockSize + 3)
+	err = r.seekRecord(blockSize + 3)
 	if err != nil {
 		t.Fatalf("SeekRecord: %v", err)
 	}
 	if _, err = r.Next(); err == nil {
 		t.Fatalf("Expected an error seeking to an invalid chunk boundary")
 	}
-	r.Recover()
+	r.recover()
 
 	// Seek to the fifth block and verify all records can be read as appropriate.
-	err = r.SeekRecord(blockSize * 4)
+	err = r.seekRecord(blockSize * 4)
 	if err != nil {
 		t.Fatalf("SeekRecord: %v", err)
 	}
@@ -699,24 +698,24 @@ func TestSeekRecord(t *testing.T) {
 	check(2)
 
 	// Seek back to the fourth block, and read all subsequent records and verify them.
-	err = r.SeekRecord(blockSize * 3)
+	err = r.seekRecord(blockSize * 3)
 	if err != nil {
 		t.Fatalf("SeekRecord: %v", err)
 	}
 	check(1)
 
 	// Now seek past the end of the file and verify it causes an error.
-	err = r.SeekRecord(1 << 20)
+	err = r.seekRecord(1 << 20)
 	if err == nil {
 		t.Fatalf("Seek past the end of a file didn't cause an error")
 	}
 	if err != io.EOF {
 		t.Fatalf("Seeking past EOF raised unexpected error: %v", err)
 	}
-	r.Recover() // Verify recovery works.
+	r.recover() // Verify recovery works.
 
 	// Validate the current records are returned after seeking to a valid offset.
-	err = r.SeekRecord(blockSize * 4)
+	err = r.seekRecord(blockSize * 4)
 	if err != nil {
 		t.Fatalf("SeekRecord: %v", err)
 	}
@@ -728,11 +727,11 @@ func TestLastRecordOffset(t *testing.T) {
 		// The first record will consume 3 entire blocks but a fraction of the 4th.
 		blockSize*3,
 		// The second record will completely fill the remainder of the 4th block.
-		3*(blockSize-headerSize)-2*blockSize-2*headerSize,
+		3*(blockSize-legacyHeaderSize)-2*blockSize-2*legacyHeaderSize,
 		// Consume the entirety of the 5th block.
-		blockSize-headerSize,
+		blockSize-legacyHeaderSize,
 		// Consume the entirety of the 6th block.
-		blockSize-headerSize,
+		blockSize-legacyHeaderSize,
 		// Consume roughly half of the 7th block.
 		blockSize/2,
 	)
@@ -776,10 +775,52 @@ func TestNoLastRecordOffset(t *testing.T) {
 	}
 }
 
+func TestInvalidLogNum(t *testing.T) {
+	var buf bytes.Buffer
+	w := NewLogWriter(&buf, 1)
+	for i := 0; i < 10; i++ {
+		s := fmt.Sprintf("%04d\n", i)
+		if _, err := w.WriteRecord([]byte(s)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := w.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	{
+		r := NewReader(bytes.NewReader(buf.Bytes()), 1)
+		for i := 0; i < 10; i++ {
+			rr, err := r.Next()
+			if err != nil {
+				t.Fatal(err)
+			}
+			x, err := ioutil.ReadAll(rr)
+			if err != nil {
+				t.Fatal(err)
+			}
+			s := fmt.Sprintf("%04d\n", i)
+			if s != string(x) {
+				t.Fatalf("expected %s, but found %s", s, x)
+			}
+		}
+		if _, err := r.Next(); err != io.EOF {
+			t.Fatalf("expected EOF, but found %s", err)
+		}
+	}
+
+	{
+		r := NewReader(bytes.NewReader(buf.Bytes()), 2)
+		if _, err := r.Next(); err != ErrInvalidLogNum {
+			t.Fatalf("expected %s, but found %s\n", ErrInvalidLogNum, err)
+		}
+	}
+}
+
 func BenchmarkRecordWrite(b *testing.B) {
 	for _, size := range []int{8, 16, 32, 64, 128} {
 		b.Run(fmt.Sprintf("size=%d", size), func(b *testing.B) {
-			w := NewLogWriter(ioutil.Discard)
+			w := NewLogWriter(ioutil.Discard, 0 /* logNum */)
 			defer w.Close()
 			buf := make([]byte, size)
 
diff --git a/version_edit_test.go b/version_edit_test.go
index 32f7eb3..aedd412 100644
--- a/version_edit_test.go
+++ b/version_edit_test.go
@@ -150,7 +150,7 @@ func TestVersionEditDecode(t *testing.T) {
 				t.Fatalf("filename=%q: open error: %v", tc.filename, err)
 			}
 			defer f.Close()
-			i, r := 0, record.NewReader(f)
+			i, r := 0, record.NewReader(f, 0 /* logNum */)
 			for {
 				rr, err := r.Next()
 				if err == io.EOF {