Segment sequence ID (#34)

Author: akrylysov

compaction.go

@@ -87,7 +87,7 @@ func (db *DB) compact(f *segment) (CompactionResult, error) {
 }
 
 func (db *DB) pickForCompaction() ([]*segment, error) {
-	segments, err := db.datalog.segmentsByModification()
+	segments, err := db.datalog.segmentsBySequenceID()
 	if err != nil {
 		return nil, err
 	}

compaction_test.go

@@ -72,8 +72,8 @@ func TestCompaction(t *testing.T) {
 		assert.Nil(t, db.datalog.segments[0])
 		assert.Equal(t, &segmentMeta{PutRecords: 1}, db.datalog.segments[1].meta)
 		// Compacted file was removed.
-		assert.Equal(t, false, fileExists(filepath.Join(db.opts.path, segmentName(0))))
-		assert.Equal(t, false, fileExists(filepath.Join(db.opts.path, segmentMetaName(0))))
+		assert.Equal(t, false, fileExists(filepath.Join(db.opts.path, segmentName(0, 1))))
+		assert.Equal(t, false, fileExists(filepath.Join(db.opts.path, segmentMetaName(0, 1))))
 	})
 
 	run("compact entire segment", func(t *testing.T, db *DB) {

datalog.go

@@ -9,7 +9,6 @@ import (
 	"sort"
 	"strconv"
 	"strings"
-	"time"
 )
 
 const (
@@ -18,10 +17,10 @@ const (
 
 // datalog is a write-ahead log.
 type datalog struct {
-	opts     *Options
-	curSeg   *segment
-	segments [maxSegments]*segment
-	modTime  int64
+	opts          *Options
+	curSeg        *segment
+	segments      [maxSegments]*segment
+	maxSequenceID uint64
 }
 
 func openDatalog(opts *Options) (*datalog, error) {
@@ -31,8 +30,7 @@ func openDatalog(opts *Options) (*datalog, error) {
 	}
 
 	dl := &datalog{
-		opts:    opts,
-		modTime: time.Now().UnixNano(),
+		opts: opts,
 	}
 
 	for _, file := range files {
@@ -41,13 +39,12 @@ func openDatalog(opts *Options) (*datalog, error) {
 		if ext != segmentExt {
 			continue
 		}
-		id, err := strconv.ParseInt(strings.TrimSuffix(name, ext), 10, 16)
+		seg, err := dl.openSegment(name)
 		if err != nil {
 			return nil, err
 		}
-		_, err = dl.openSegment(filepath.Join(opts.path, name), uint16(id))
-		if err != nil {
-			return nil, err
+		if seg.sequenceID > dl.maxSequenceID {
+			dl.maxSequenceID = seg.sequenceID
 		}
 	}
 
@@ -58,80 +55,110 @@ func openDatalog(opts *Options) (*datalog, error) {
 	return dl, nil
 }
 
-func (dl *datalog) openSegment(path string, id uint16) (*segment, error) {
-	f, err := openFile(dl.opts.FileSystem, path, false)
+func parseSegmentName(name string) (uint16, uint64, error) {
+	parts := strings.SplitN(strings.TrimSuffix(name, segmentExt), "-", 2)
+	id, err := strconv.ParseUint(parts[0], 10, 16)
+	if err != nil {
+		return 0, 0, err
+	}
+	var seqID uint64
+	if len(parts) == 2 {
+		seqID, err = strconv.ParseUint(parts[1], 10, 64)
+		if err != nil {
+			return 0, 0, err
+		}
+	}
+	return uint16(id), seqID, nil
+}
+
+func (dl *datalog) openSegment(name string) (*segment, error) {
+	id, seqID, err := parseSegmentName(name)
+	if err != nil {
+		return nil, err
+	}
+	f, err := openFile(dl.opts.FileSystem, filepath.Join(dl.opts.path, name), false)
 	if err != nil {
 		return nil, err
 	}
-	var modTime int64
 	meta := &segmentMeta{}
 	if !f.empty() {
-		metaPath := filepath.Join(dl.opts.path, segmentMetaName(id))
+		metaPath := filepath.Join(dl.opts.path, name+metaExt)
 		if err := readGobFile(dl.opts.FileSystem, metaPath, &meta); err != nil {
 			logger.Printf("error reading segment meta %d: %v", id, err)
 			// TODO: rebuild meta?
 		}
-		stat, err := f.MmapFile.Stat()
-		if err != nil {
-			return nil, err
-		}
-		modTime = stat.ModTime().UnixNano()
 	} else {
-		dl.modTime++
-		modTime = dl.modTime
+		dl.maxSequenceID++
+		seqID = dl.maxSequenceID
 	}
 
-	df := &segment{file: f, id: id, meta: meta, modTime: modTime}
+	df := &segment{
+		file:       f,
+		id:         id,
+		sequenceID: seqID,
+		name:       name,
+		meta:       meta,
+	}
 	dl.segments[id] = df
 	return df, nil
 }
 
-func (dl *datalog) nextWritableSegmentID() (uint16, error) {
-	for i, file := range dl.segments {
-		if file == nil || !file.meta.Full {
-			return uint16(i), nil
+func (dl *datalog) nextWritableSegmentID() (uint16, uint64, error) {
+	for id, seg := range dl.segments {
+		// Pick unfilled segment.
+		if seg != nil && !seg.meta.Full {
+			dl.maxSequenceID++
+			return uint16(id), dl.maxSequenceID, nil
+		}
+	}
+	for id, seg := range dl.segments {
+		// Pick empty segment.
+		if seg == nil {
+			dl.maxSequenceID++
+			return uint16(id), dl.maxSequenceID, nil
 		}
 	}
-	return 0, fmt.Errorf("number of segments exceeds %d", maxSegments)
+	return 0, 0, fmt.Errorf("number of segments exceeds %d", maxSegments)
 }
 
 func (dl *datalog) swapSegment() error {
-	id, err := dl.nextWritableSegmentID()
+	id, seqID, err := dl.nextWritableSegmentID()
 	if err != nil {
 		return err
 	}
-	var f *segment
+	var seg *segment
 	if dl.segments[id] != nil {
-		f = dl.segments[id]
+		seg = dl.segments[id]
 	} else {
-		name := segmentName(id)
-		f, err = dl.openSegment(filepath.Join(dl.opts.path, name), id)
+		name := segmentName(id, seqID)
+		seg, err = dl.openSegment(name)
 		if err != nil {
 			return err
 		}
 	}
-	dl.curSeg = f
+	dl.curSeg = seg
 	return nil
 }
 
-func (dl *datalog) removeSegment(f *segment) error {
-	dl.segments[f.id] = nil
+func (dl *datalog) removeSegment(seg *segment) error {
+	dl.segments[seg.id] = nil
 
-	if err := f.Close(); err != nil {
+	if err := seg.Close(); err != nil {
 		return err
 	}
 
-	// Remove segment.
-	filePath := filepath.Join(dl.opts.path, segmentName(f.id))
-	if err := os.Remove(filePath); err != nil {
+	// Remove segment meta from FS.
+	metaPath := filepath.Join(dl.opts.path, seg.name+segmentExt)
+	if err := dl.opts.FileSystem.Remove(metaPath); err != nil && !os.IsNotExist(err) {
 		return err
 	}
 
-	// Remove segment meta.
-	metaPath := filepath.Join(dl.opts.path, segmentMetaName(f.id))
-	if err := os.Remove(metaPath); err != nil && !os.IsNotExist(err) {
+	// Remove segment from FS.
+	filePath := filepath.Join(dl.opts.path, seg.name)
+	if err := dl.opts.FileSystem.Remove(filePath); err != nil {
 		return err
 	}
+
 	return nil
 }
 
@@ -208,23 +235,23 @@ func (dl *datalog) sync() error {
 }
 
 func (dl *datalog) close() error {
-	for id, f := range dl.segments {
-		if f == nil {
+	for _, seg := range dl.segments {
+		if seg == nil {
 			continue
 		}
-		if err := f.Close(); err != nil {
+		if err := seg.Close(); err != nil {
 			return err
 		}
-		metaPath := filepath.Join(dl.opts.path, segmentMetaName(uint16(id)))
-		if err := writeGobFile(dl.opts.FileSystem, metaPath, f.meta); err != nil {
+		metaPath := filepath.Join(dl.opts.path, seg.name+metaExt)
+		if err := writeGobFile(dl.opts.FileSystem, metaPath, seg.meta); err != nil {
 			return err
 		}
 	}
 	return nil
 }
 
-func (dl *datalog) segmentsByModification() ([]*segment, error) {
-	// Sort segments in ascending order by last modified time.
+func (dl *datalog) segmentsBySequenceID() ([]*segment, error) {
+	// Sort segments in ascending order by sequence ID.
 	var segments []*segment
 
 	for _, f := range dl.segments {
@@ -235,7 +262,7 @@ func (dl *datalog) segmentsByModification() ([]*segment, error) {
 	}
 
 	sort.SliceStable(segments, func(i, j int) bool {
-		return segments[i].modTime < segments[j].modTime
+		return segments[i].sequenceID < segments[j].sequenceID
 	})
 
 	return segments, nil

datalog_test.go

@@ -15,7 +15,7 @@ func TestDatalog(t *testing.T) {
 	assert.Equal(t, &segmentMeta{PutRecords: 1}, db.datalog.segments[0].meta)
 	assert.Nil(t, db.datalog.segments[1])
 
-	sm, err := db.datalog.segmentsByModification()
+	sm, err := db.datalog.segmentsBySequenceID()
 	assert.Nil(t, err)
 	assert.Equal(t, []*segment{db.datalog.segments[0]}, sm)
 
@@ -26,7 +26,7 @@ func TestDatalog(t *testing.T) {
 	assert.Equal(t, &segmentMeta{PutRecords: 1, Full: true}, db.datalog.segments[0].meta)
 	assert.Equal(t, &segmentMeta{PutRecords: 1}, db.datalog.segments[1].meta)
 
-	sm, err = db.datalog.segmentsByModification()
+	sm, err = db.datalog.segmentsBySequenceID()
 	assert.Nil(t, err)
 	assert.Equal(t, []*segment{db.datalog.segments[0], db.datalog.segments[1]}, sm)
 

db_test.go

@@ -139,7 +139,7 @@ func TestSimple(t *testing.T) {
 
 	// Simulate crash.
 	assert.Nil(t, touchFile(filepath.Join("test.db", lockName)))
-	assert.Nil(t, os.Remove(filepath.Join("test.db", segmentMetaName(0))))
+	assert.Nil(t, os.Remove(filepath.Join("test.db", segmentMetaName(0, 1))))
 	assert.Nil(t, os.Remove(filepath.Join("test.db", indexMetaName)))
 
 	// Open and check again

recovery.go

@@ -68,7 +68,7 @@ type recoveryIterator struct {
 }
 
 func newRecoveryIterator(dl *datalog) (*recoveryIterator, error) {
-	files, err := dl.segmentsByModification()
+	files, err := dl.segmentsBySequenceID()
 	if err != nil {
 		return nil, err
 	}

recovery_test.go

@@ -9,60 +9,60 @@ import (
 )
 
 func TestRecovery(t *testing.T) {
-	dfPath := filepath.Join("test.db", segmentName(0))
+	segPath := filepath.Join("test.db", segmentName(0, 1))
 	testCases := []struct {
 		name string
 		fn   func() error
 	}{
 		{
 			name: "all zeroes",
 			fn: func() error {
-				return appendFile(dfPath, make([]byte, 128))
+				return appendFile(segPath, make([]byte, 128))
 			},
 		},
 		{
 			name: "partial kv size",
 			fn: func() error {
-				return appendFile(dfPath, []byte{1})
+				return appendFile(segPath, []byte{1})
 			},
 		},
 		{
 			name: "only kv size",
 			fn: func() error {
-				return appendFile(dfPath, []byte{1, 0, 1, 0, 0, 0})
+				return appendFile(segPath, []byte{1, 0, 1, 0, 0, 0})
 			},
 		},
 		{
 			name: "kv size and key",
 			fn: func() error {
-				return appendFile(dfPath, []byte{1, 0, 1, 0, 0, 0, 1})
+				return appendFile(segPath, []byte{1, 0, 1, 0, 0, 0, 1})
 			},
 		},
 		{
 			name: "kv size, key, value",
 			fn: func() error {
-				return appendFile(dfPath, []byte{1, 0, 1, 0, 0, 0, 1, 1})
+				return appendFile(segPath, []byte{1, 0, 1, 0, 0, 0, 1, 1})
 			},
 		},
 		{
 			name: "kv size, key, value, partial crc32",
 			fn: func() error {
-				return appendFile(dfPath, []byte{1, 0, 1, 0, 0, 0, 1, 1, 40})
+				return appendFile(segPath, []byte{1, 0, 1, 0, 0, 0, 1, 1, 40})
 			},
 		},
 		{
 			name: "kv size, key, value, invalid crc32",
 			fn: func() error {
-				return appendFile(dfPath, []byte{1, 0, 1, 0, 0, 0, 1, 1, 40, 19, 197, 0})
+				return appendFile(segPath, []byte{1, 0, 1, 0, 0, 0, 1, 1, 40, 19, 197, 0})
 			},
 		},
 		{
 			name: "corrupted and not corrupted record",
 			fn: func() error {
-				if err := appendFile(dfPath, []byte{1, 0, 1, 0, 0, 0, 1, 1, 40, 19, 197, 0}); err != nil {
+				if err := appendFile(segPath, []byte{1, 0, 1, 0, 0, 0, 1, 1, 40, 19, 197, 0}); err != nil {
 					return err
 				}
-				return appendFile(dfPath, []byte{1, 0, 1, 0, 0, 0, 1, 1, 133, 13, 200, 12})
+				return appendFile(segPath, []byte{1, 0, 1, 0, 0, 0, 1, 1, 133, 13, 200, 12})
 			},
 		},
 	}
@@ -71,7 +71,7 @@ func TestRecovery(t *testing.T) {
 		t.Run(fmt.Sprintf("case %s", testCase.name), func(t *testing.T) {
 			db, err := createTestDB(nil)
 			assert.Nil(t, err)
-			// Fill file 0.
+			// Fill segment 0.
 			var i uint8
 			for i = 0; i < 128; i++ {
 				assert.Nil(t, db.Put([]byte{i}, []byte{i}))

segment.go

@@ -21,13 +21,14 @@ const (
 // It consists of a sequence of binary-encoded variable length records.
 type segment struct {
 	*file
-	id      uint16
-	meta    *segmentMeta
-	modTime int64
+	id         uint16 // Physical segment identifier.
+	sequenceID uint64 // Logical monotonically increasing segment identifier.
+	name       string
+	meta       *segmentMeta
 }
 
-func segmentName(id uint16) string {
-	return fmt.Sprintf("%05d%s", id, segmentExt)
+func segmentName(id uint16, sequenceID uint64) string {
+	return fmt.Sprintf("%05d-%d%s", id, sequenceID, segmentExt)
 }
 
 type segmentMeta struct {
@@ -38,8 +39,8 @@ type segmentMeta struct {
 	DeletedBytes  uint32
 }
 
-func segmentMetaName(id uint16) string {
-	return fmt.Sprintf("%05d%s%s", id, segmentExt, metaExt)
+func segmentMetaName(id uint16, sequenceID uint64) string {
+	return segmentName(id, sequenceID) + metaExt
 }
 
 // Binary representation of a segment record: