Remove shadow WAL iterator
This commit removes the shadow WAL iterator and replaces it with a fileWalSegmentIterator instead. This works since the shadow WAL now has the same structure as the replica WAL. This reduces duplicate code and will make it so read replication can be daisy chained in the future.
This commit is contained in:
98
db.go
98
db.go
@@ -1236,103 +1236,7 @@ func (db *DB) WALSegments(ctx context.Context, generation string) (WALSegmentIte
|
|||||||
|
|
||||||
sort.Ints(indexes)
|
sort.Ints(indexes)
|
||||||
|
|
||||||
return newShadowWALSegmentIterator(db, generation, indexes), nil
|
return newFileWALSegmentIterator(db.ShadowWALDir(generation), generation, indexes), nil
|
||||||
}
|
|
||||||
|
|
||||||
type shadowWALSegmentIterator struct {
|
|
||||||
db *DB
|
|
||||||
generation string
|
|
||||||
indexes []int
|
|
||||||
|
|
||||||
infos []WALSegmentInfo
|
|
||||||
err error
|
|
||||||
}
|
|
||||||
|
|
||||||
func newShadowWALSegmentIterator(db *DB, generation string, indexes []int) *shadowWALSegmentIterator {
|
|
||||||
return &shadowWALSegmentIterator{
|
|
||||||
db: db,
|
|
||||||
generation: generation,
|
|
||||||
indexes: indexes,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func (itr *shadowWALSegmentIterator) Close() (err error) {
|
|
||||||
return itr.err
|
|
||||||
}
|
|
||||||
|
|
||||||
func (itr *shadowWALSegmentIterator) Next() bool {
|
|
||||||
// Exit if an error has already occurred.
|
|
||||||
if itr.err != nil {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
|
|
||||||
for {
|
|
||||||
// Move to the next segment in cache, if available.
|
|
||||||
if len(itr.infos) > 1 {
|
|
||||||
itr.infos = itr.infos[1:]
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
itr.infos = itr.infos[:0] // otherwise clear infos
|
|
||||||
|
|
||||||
// If no indexes remain, stop iteration.
|
|
||||||
if len(itr.indexes) == 0 {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
|
|
||||||
// Read segments into a cache for the current index.
|
|
||||||
index := itr.indexes[0]
|
|
||||||
itr.indexes = itr.indexes[1:]
|
|
||||||
f, err := os.Open(filepath.Join(itr.db.ShadowWALDir(itr.generation), FormatIndex(index)))
|
|
||||||
if err != nil {
|
|
||||||
itr.err = err
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
defer func() { _ = f.Close() }()
|
|
||||||
|
|
||||||
fis, err := f.Readdir(-1)
|
|
||||||
if err != nil {
|
|
||||||
itr.err = err
|
|
||||||
return false
|
|
||||||
} else if err := f.Close(); err != nil {
|
|
||||||
itr.err = err
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
for _, fi := range fis {
|
|
||||||
filename := filepath.Base(fi.Name())
|
|
||||||
if fi.IsDir() {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
offset, err := ParseOffset(strings.TrimSuffix(filename, ".wal.lz4"))
|
|
||||||
if err != nil {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
itr.infos = append(itr.infos, WALSegmentInfo{
|
|
||||||
Generation: itr.generation,
|
|
||||||
Index: index,
|
|
||||||
Offset: offset,
|
|
||||||
Size: fi.Size(),
|
|
||||||
CreatedAt: fi.ModTime().UTC(),
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
// Ensure segments are sorted within index.
|
|
||||||
sort.Sort(WALSegmentInfoSlice(itr.infos))
|
|
||||||
|
|
||||||
if len(itr.infos) > 0 {
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func (itr *shadowWALSegmentIterator) Err() error { return itr.err }
|
|
||||||
|
|
||||||
func (itr *shadowWALSegmentIterator) WALSegment() WALSegmentInfo {
|
|
||||||
if len(itr.infos) == 0 {
|
|
||||||
return WALSegmentInfo{}
|
|
||||||
}
|
|
||||||
return itr.infos[0]
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// SQLite WAL constants
|
// SQLite WAL constants
|
||||||
|
|||||||
Reference in New Issue
Block a user