New: Delete Expired Keys

db.go

@@ -1,6 +1,8 @@
 package rosedb
 
 import (
+	"bytes"
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -49,6 +51,7 @@ type DB struct {
 	batchPool    sync.Pool
 	watchCh      chan *Event // user consume channel for watch events
 	watcher      *Watcher
+	cursorKey    []byte // the location to which DeleteExpiredKeys executes.
 }
 
 // Stat represents the statistics of the database.
@@ -489,8 +492,6 @@ func (db *DB) checkValue(chunk []byte) ([]byte, error) {
 	record := decodeLogRecord(chunk)
 	now := time.Now().UnixNano()
 	if record.Type == LogRecordDeleted || record.IsExpired(now) {
-		// delete from index if the key is expired.
-		db.index.Delete(record.Key)
 		return nil, ErrKeyNotFound
 	}
 	return record.Value, nil
@@ -576,3 +577,58 @@ func (db *DB) loadIndexFromWAL() error {
 	}
 	return nil
 }
+
+// DeleteExpiredKeys scan the entire index in ascending order to delete expired keys.
+// It is a time-consuming operation, so we need to specify a timeout
+// to prevent the DB from being unavailable for a long time.
+func (db *DB) DeleteExpiredKeys(timeout time.Duration) {
+	db.mu.Lock()
+	defer db.mu.Unlock()
+
+	// set expiration time
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	now := time.Now().UnixNano()
+	go func(ctx context.Context) {
+		for {
+			// get 100 key's positions
+			positions := make([]*wal.ChunkPosition, 0, 100)
+			db.index.AscendGreaterOrEqual(db.cursorKey, func(k []byte, pos *wal.ChunkPosition) (bool, error) {
+				// filter processed key
+				if bytes.Compare(k, db.cursorKey) == 0 {
+					return true, nil
+				}
+				positions = append(positions, pos)
+				if len(positions) >= 100 {
+					return false, nil
+				}
+				return true, nil
+			})
+
+			// If keys in the db.index has been traversed, len(positions) will be 0.
+			if len(positions) == 0 {
+				db.cursorKey = []byte{}
+				return
+			}
+
+			// delete from index if the key is expired.
+			for _, pos := range positions {
+				chunk, err := db.dataFiles.Read(pos)
+				if err != nil {
+					continue
+				}
+				record := decodeLogRecord(chunk)
+				if record.IsExpired(now) {
+					db.index.Delete(record.Key)
+				}
+				db.cursorKey = record.Key
+			}
+		}
+	}(ctx)
+
+	select {
+	case <-ctx.Done():
+		return
+	}
+}

db_test.go

@@ -563,3 +563,25 @@ func TestDB_Expire2(t *testing.T) {
 	err = db2.Expire(utils.GetTestKey(2), time.Second)
 	assert.Equal(t, err, ErrKeyNotFound)
 }
+
+func TestDB_DeleteExpiredKeys(t *testing.T) {
+	options := DefaultOptions
+	db, err := Open(options)
+	assert.Nil(t, err)
+	defer destroyDB(db)
+
+	for i := 0; i < 10000; i++ {
+		err = db.Put(utils.GetTestKey(i), utils.RandomValue(10))
+		assert.Nil(t, err)
+	}
+	for i := 10000; i < 100001; i++ {
+		err = db.PutWithTTL(utils.GetTestKey(i), utils.RandomValue(10), time.Second*1)
+		assert.Nil(t, err)
+	}
+
+	// wait for key to expire
+	time.Sleep(time.Second * 2)
+
+	db.DeleteExpiredKeys(time.Second * 2)
+	assert.Equal(t, 10000, db.Stat().KeysNum)
+}