db: fix WAL offset in replayWAL errors

If an error occurs while replaying the WALs during Open, the error is annotated
with the log number and the offset within the log. Previously, the offset was
always zero. This also caused the queued flushables to be created with a zero
logSize. This didn't matter in practice because the queued flushables would be
flushed before Open returned.

This commit adds a PreviousFilesBytes total to the wal.Offset type to describe
the number of bytes replayed from previous WAL segement files.

Author: jbowens

open.go

@@ -766,7 +766,7 @@ func (d *DB) replayWAL(
 		buf             bytes.Buffer
 		mem             *memTable
 		entry           *flushableEntry
-		offset          int64 // byte offset in rr
+		offset          wal.Offset
 		lastFlushOffset int64
 		keysReplayed    int64 // number of keys replayed
 		batchesReplayed int64 // number of batches replayed
@@ -792,12 +792,13 @@ func (d *DB) replayWAL(
 			return
 		}
 		var logSize uint64
-		if offset >= lastFlushOffset {
-			logSize = uint64(offset - lastFlushOffset)
+		mergedOffset := offset.Physical + offset.PreviousFilesBytes
+		if mergedOffset >= lastFlushOffset {
+			logSize = uint64(mergedOffset - lastFlushOffset)
 		}
 		// Else, this was the initial memtable in the read-only case which must have
 		// been empty, but we need to flush it since we don't want to add to it later.
-		lastFlushOffset = offset
+		lastFlushOffset = mergedOffset
 		entry.logSize = logSize
 		if !d.opts.ReadOnly {
 			toFlush = append(toFlush, entry)
@@ -836,12 +837,14 @@ func (d *DB) replayWAL(
 	}
 	defer func() {
 		if err != nil {
-			err = errors.WithDetailf(err, "replaying wal %d, offset %d", ll.Num, offset)
+			err = errors.WithDetailf(err, "replaying wal %d, offset %s", ll.Num, offset)
 		}
 	}()
 
 	for {
-		r, offset, err := rr.NextRecord()
+		var r io.Reader
+		var err error
+		r, offset, err = rr.NextRecord()
 		if err == nil {
 			_, err = io.Copy(&buf, r)
 		}
@@ -1048,7 +1051,7 @@ func (d *DB) replayWAL(
 		buf.Reset()
 	}
 
-	d.opts.Logger.Infof("[JOB %d] WAL %s stopped reading at offset: %d; replayed %d keys in %d batches",
+	d.opts.Logger.Infof("[JOB %d] WAL %s stopped reading at offset: %s; replayed %d keys in %d batches",
 		jobID, base.DiskFileNum(ll.Num).String(), offset, keysReplayed, batchesReplayed)
 	flushMem()
 

wal/reader.go

@@ -360,6 +360,7 @@ func (r *virtualWALReader) nextFile() error {
 	}
 
 	fs, path := r.LogicalLog.SegmentLocation(r.currIndex)
+	r.off.PreviousFilesBytes += r.off.Physical
 	r.off.PhysicalFile = path
 	r.off.Physical = 0
 	var err error

wal/testdata/reader

@@ -52,13 +52,13 @@ r.NextRecord() = (rr, (000001.log: 1035), <nil>)
 r.NextRecord() = (rr, (000001.log: 1076), <nil>)
   io.ReadAll(rr) = ("1500000000000000320000004f091e9a83fdeae0ec55eb233a9b5394cb3c7856... <512000-byte record>", <nil>)
   BatchHeader: [seqNum=21,count=50]
-r.NextRecord() = (rr, (000001-001.log: 0), <nil>)
+r.NextRecord() = (rr, (000001-001.log: 0), 513252 from previous files, <nil>)
   io.ReadAll(rr) = ("16000000000000000200000038d0ccacfb33b57fb3d386cbe2b67a2fbdc82214... <412-byte record>", <nil>)
   BatchHeader: [seqNum=22,count=2]
-r.NextRecord() = (rr, (000001-001.log: 498), <nil>)
+r.NextRecord() = (rr, (000001-001.log: 498), 513252 from previous files, <nil>)
   io.ReadAll(rr) = ("180000000000000001000000ede8f156c48faf84dd55235d19a2df01d13021fc... <100-byte record>", <nil>)
   BatchHeader: [seqNum=24,count=1]
-r.NextRecord() = (rr, (000001-001.log: 609), EOF)
+r.NextRecord() = (rr, (000001-001.log: 609), 513252 from previous files, EOF)
 
 # Test a recycled log file. Recycle 000001.log as 000002.log. This time, do not
 # exit cleanly. This simulates a hard process exit (eg, during a fatal shutdown,
@@ -134,13 +134,13 @@ r.NextRecord() = (rr, (000003.log: 111), <nil>)
 r.NextRecord() = (rr, (000003.log: 272), <nil>)
   io.ReadAll(rr) = ("2a000000000000000100000019458dc5400169e5", <nil>)
   BatchHeader: [seqNum=42,count=1]
-r.NextRecord() = (rr, (000003-001.log: 192), <nil>)
+r.NextRecord() = (rr, (000003-001.log: 192), 303 from previous files, <nil>)
   io.ReadAll(rr) = ("2b00000000000000030000009cbf29476e797bac2db8bfea65bda29ea50ddbe4... <80-byte record>", <nil>)
   BatchHeader: [seqNum=43,count=3]
-r.NextRecord() = (rr, (000003-001.log: 283), <nil>)
+r.NextRecord() = (rr, (000003-001.log: 283), 303 from previous files, <nil>)
   io.ReadAll(rr) = ("2e000000000000000900000027337fa5bd626044dc5d9d08085bf4ce13bc8d00... <2055-byte record>", <nil>)
   BatchHeader: [seqNum=46,count=9]
-r.NextRecord() = (rr, (000003-001.log: 2349), EOF)
+r.NextRecord() = (rr, (000003-001.log: 2349), 303 from previous files, EOF)
 
 # Extend logical log file 000003 with another log file, the result of failing
 # back to the original the device. This time do an "unclean" close.
@@ -166,16 +166,16 @@ r.NextRecord() = (rr, (000003.log: 111), <nil>)
 r.NextRecord() = (rr, (000003.log: 272), <nil>)
   io.ReadAll(rr) = ("2a000000000000000100000019458dc5400169e5", <nil>)
   BatchHeader: [seqNum=42,count=1]
-r.NextRecord() = (rr, (000003-001.log: 192), <nil>)
+r.NextRecord() = (rr, (000003-001.log: 192), 303 from previous files, <nil>)
   io.ReadAll(rr) = ("2b00000000000000030000009cbf29476e797bac2db8bfea65bda29ea50ddbe4... <80-byte record>", <nil>)
   BatchHeader: [seqNum=43,count=3]
-r.NextRecord() = (rr, (000003-001.log: 283), <nil>)
+r.NextRecord() = (rr, (000003-001.log: 283), 303 from previous files, <nil>)
   io.ReadAll(rr) = ("2e000000000000000900000027337fa5bd626044dc5d9d08085bf4ce13bc8d00... <2055-byte record>", <nil>)
   BatchHeader: [seqNum=46,count=9]
-r.NextRecord() = (rr, (000003-002.log: 2157), <nil>)
+r.NextRecord() = (rr, (000003-002.log: 2157), 2652 from previous files, <nil>)
   io.ReadAll(rr) = ("370000000000000002000000ff0710201f4008e679428b4994708a1af8507303... <205-byte record>", <nil>)
   BatchHeader: [seqNum=55,count=2]
-r.NextRecord() = (rr, (000003-002.log: 2373), EOF)
+r.NextRecord() = (rr, (000003-002.log: 2373), 2652 from previous files, EOF)
 
 # Test reading a log file that does not exist.
 
@@ -243,16 +243,16 @@ r.NextRecord() = (rr, (000005.log: 909), <nil>)
 r.NextRecord() = (rr, (000005.log: 3445), <nil>)
   io.ReadAll(rr) = ("0b7401000000000000010000907cd29c9a6deaf239e76e3374f6e9eef047f57f... <2566-byte record>", <nil>)
   BatchHeader: [seqNum=95243,count=256]
-r.NextRecord() = (rr, (000005-001.log: 0), <nil>)
+r.NextRecord() = (rr, (000005-001.log: 0), 6022 from previous files, <nil>)
   io.ReadAll(rr) = ("0b75010000000000020000006cad8a0a1461d1ec53bb834b47c6853e040ae9ce... <44-byte record>", <nil>)
   BatchHeader: [seqNum=95499,count=2]
-r.NextRecord() = (rr, (000005-001.log: 55), <nil>)
+r.NextRecord() = (rr, (000005-001.log: 55), 6022 from previous files, <nil>)
   io.ReadAll(rr) = ("0d7501000000000005000000c78be2f74d28753a03854ed63e6fd0f17113688d... <416-byte record>", <nil>)
   BatchHeader: [seqNum=95501,count=5]
-r.NextRecord() = (rr, (000005-001.log: 482), <nil>)
+r.NextRecord() = (rr, (000005-001.log: 482), 6022 from previous files, <nil>)
   io.ReadAll(rr) = ("12750100000000001d00000096cedf6103af61c008d9f850e63a1dfc7518b9a7... <199-byte record>", <nil>)
   BatchHeader: [seqNum=95506,count=29]
-r.NextRecord() = (rr, (000005-001.log: 692), pebble/record: invalid chunk)
+r.NextRecord() = (rr, (000005-001.log: 692), 6022 from previous files, pebble/record: invalid chunk)
 
 # Read again, this time pretending we found a third segment with the
 # logNameIndex=002. This helps exercise error conditions switching to a new
@@ -272,16 +272,16 @@ r.NextRecord() = (rr, (000005.log: 909), <nil>)
 r.NextRecord() = (rr, (000005.log: 3445), <nil>)
   io.ReadAll(rr) = ("0b7401000000000000010000907cd29c9a6deaf239e76e3374f6e9eef047f57f... <2566-byte record>", <nil>)
   BatchHeader: [seqNum=95243,count=256]
-r.NextRecord() = (rr, (000005-001.log: 0), <nil>)
+r.NextRecord() = (rr, (000005-001.log: 0), 6022 from previous files, <nil>)
   io.ReadAll(rr) = ("0b75010000000000020000006cad8a0a1461d1ec53bb834b47c6853e040ae9ce... <44-byte record>", <nil>)
   BatchHeader: [seqNum=95499,count=2]
-r.NextRecord() = (rr, (000005-001.log: 55), <nil>)
+r.NextRecord() = (rr, (000005-001.log: 55), 6022 from previous files, <nil>)
   io.ReadAll(rr) = ("0d7501000000000005000000c78be2f74d28753a03854ed63e6fd0f17113688d... <416-byte record>", <nil>)
   BatchHeader: [seqNum=95501,count=5]
-r.NextRecord() = (rr, (000005-001.log: 482), <nil>)
+r.NextRecord() = (rr, (000005-001.log: 482), 6022 from previous files, <nil>)
   io.ReadAll(rr) = ("12750100000000001d00000096cedf6103af61c008d9f850e63a1dfc7518b9a7... <199-byte record>", <nil>)
   BatchHeader: [seqNum=95506,count=29]
-r.NextRecord() = (rr, (000005-002.log: 0), opening WAL file segment "000005-002.log": open 000005-002.log: file does not exist)
+r.NextRecord() = (rr, (000005-002.log: 0), 6714 from previous files, opening WAL file segment "000005-002.log": open 000005-002.log: file does not exist)
 
 # Test a scenario where 4 unique batches are split across three physical log
 # files. The first log contains (b0, b1, b2), the second log (b1) and the third
@@ -325,10 +325,10 @@ r.NextRecord() = (rr, (000006.log: 406), <nil>)
 r.NextRecord() = (rr, (000006.log: 94105), <nil>)
   io.ReadAll(rr) = ("1c0200000000000001000000404841433f5369713ee90d8f86c50c5903fa38e9... <180-byte record>", <nil>)
   BatchHeader: [seqNum=540,count=1]
-r.NextRecord() = (rr, (000006-001.log: 93890), <nil>)
+r.NextRecord() = (rr, (000006-001.log: 93890), 94296 from previous files, <nil>)
   io.ReadAll(rr) = ("1d0200000000000005000000b68c7a260135dce1ce5c5498550793d15edfae62... <2055-byte record>", <nil>)
   BatchHeader: [seqNum=541,count=5]
-r.NextRecord() = (rr, (000006-001.log: 95956), EOF)
+r.NextRecord() = (rr, (000006-001.log: 95956), 94296 from previous files, EOF)
 
 # Test corrupting the tail of a batch that's large enough to be split into
 # multiple reads. Regression test for #3865.
@@ -365,10 +365,10 @@ r.NextRecord() = (rr, (000007.log: 55), <nil>)
 r.NextRecord() = (rr, (000007.log: 482), <nil>)
   io.ReadAll(rr) = ("12750100000000001d000000362bba27f0ed6f5433a12bc502873a27c67f256c... <199-byte record>", <nil>)
   BatchHeader: [seqNum=95506,count=29]
-r.NextRecord() = (rr, (000007-001.log: 0), <nil>)
+r.NextRecord() = (rr, (000007-001.log: 0), 692 from previous files, <nil>)
   io.ReadAll(rr) = ("2f75010000000000130000001ddd809cbb45782c44544a15a15dd52fb7b81a74... <45991-byte record>", <nil>)
   BatchHeader: [seqNum=95535,count=19]
-r.NextRecord() = (rr, (000007-001.log: 46013), <nil>)
+r.NextRecord() = (rr, (000007-001.log: 46013), 692 from previous files, <nil>)
   io.ReadAll(rr) = ("427501000000000013000000b30c11cf619ea65167511346cc55bb784a9af26f... <292-byte record>", <nil>)
   BatchHeader: [seqNum=95554,count=19]
-r.NextRecord() = (rr, (000007-001.log: 46316), EOF)
+r.NextRecord() = (rr, (000007-001.log: 46316), 692 from previous files, EOF)

wal/wal.go

@@ -423,10 +423,19 @@ type Offset struct {
 	// Physical indicates the file offset at which a record begins within
 	// the physical file named by PhysicalFile.
 	Physical int64
+	// PreviousFilesBytes is the bytes read from all the previous physical
+	// segment files that have been read up to the current log segment. If WAL
+	// failover is not in use, PreviousFileBytes will always be zero. Otherwise,
+	// it may be non-zero when replaying records from multiple segment files
+	// that make up a single logical WAL.
+	PreviousFilesBytes int64
 }
 
 // String implements fmt.Stringer, returning a string representation of the
 // offset.
 func (o Offset) String() string {
+	if o.PreviousFilesBytes > 0 {
+		return fmt.Sprintf("(%s: %d), %d from previous files", o.PhysicalFile, o.Physical, o.PreviousFilesBytes)
+	}
 	return fmt.Sprintf("(%s: %d)", o.PhysicalFile, o.Physical)
 }