diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml index 0065213..8bef7a6 100644 --- a/.github/workflows/go.yml +++ b/.github/workflows/go.yml @@ -1,28 +1,29 @@ -# This workflow will build a golang project -# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-go +# Run build and all unit tests on every push and pull request. +# See: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-go name: Go on: push: - branches: [ "master" ] + branches: [main, master] pull_request: - branches: [ "master" ] + branches: [main, master] + workflow_dispatch: jobs: - - build: + test: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v3 + - uses: actions/checkout@v4 - - name: Set up Go - uses: actions/setup-go@v4 - with: - go-version: '1.20' + - name: Set up Go + uses: actions/setup-go@v5 + with: + go-version-file: go.mod + cache-dependency-path: go.sum - - name: Build - run: go build -v ./... + - name: Build + run: go build -v ./... - - name: Test - run: go test -v ./... + - name: Test + run: go test -v ./... diff --git a/README.md b/README.md index 470f9c6..6b2dffd 100644 --- a/README.md +++ b/README.md @@ -1,27 +1,77 @@ -## tiny-bitcask +# tiny-bitcask -tiny-bitcask是对[bitcask论文](https://riak.com/assets/bitcask-intro.pdf)的简单实现，旨在提供用Go实现简单kv存储引擎的参考。后来我想把这个项目作为实现我各种想法的试验田。所以会持续迭代这个项目，迭代的结果就是master分支的代码。如果想看简单版本的实现可以通过以下方式拉取代码，并且切换到demo分支。 +[tiny-bitcask](https://github.com/elliotchenzichang/tiny-bitcask) is a small [Bitcask](https://riak.com/assets/bitcask-intro.pdf)-style key/value store in Go: one active append-only data file per directory, an in-memory **keydir** (hash map) pointing at the latest record per key, and optional **merge** to drop stale records and reclaim space. -````shell +This repo is an educational reference and a playground for experiments; `master` carries ongoing work. For a smaller, teaching-oriented snapshot, use the **`demo`** branch: + +```shell git clone git@github.com:elliotchenzichang/tiny-bitcask.git cd tiny-bitcask -git chckout demo -```` +git checkout demo +``` + +--- + +## Design (aligned with the paper) + +| Idea | Role in this project | +|------|----------------------| +| Append-only active file | Writes go to `ActiveFile`; when size exceeds `SegmentSize`, the file is sealed and a new active file is opened (`storage/datafiles.go`). | +| Hint files | After rotation, each sealed `fid.dat` can have a compact `fid.hint` for faster recovery; invalid or missing hints fall back to scanning the data file (`storage/hint.go`, `db.go` recovery). | +| Keydir | `index.KeyDir` maps string key → `DataPosition` (file id, offset, key/value sizes, timestamp). | +| Read path | One hash lookup + one `ReadAt` by `(fid, offset, length)`; optional **CRC32** verification on read (`Options.VerifyCRC`, default `true`). | +| Merge / compaction | Scans **immutable** files and rewrites entries that are still the live version into the active file, then deletes merged files (`DB.Merge`). Live vs. stale is decided by comparing the keydir’s `(fid, offset)` to the **start** offset of each record while scanning. | +| Tombstone delete | Deletes append a record with `DeleteFlag`; the key is written in the record for recovery; the key is removed from the keydir (`DB.Delete`). | + +--- + +## What is implemented + +- **Open / create**: `NewDB` — empty directory creates a new store; existing directory **recovers** the keydir by scanning `*.dat` files in order (or hints for sealed segments). **`Options`**: `VerifyCRC` (default on), `ReadOnly` (open existing store read-only), `ExclusiveLock` (Unix advisory `flock` on `.tiny-bitcask.lock`; shared lock when `ReadOnly`). +- **Hint files**: On **segment rotation**, a compact **`fid.hint`** is written next to the sealed **`fid.dat`** (atomic write). Hint entries omit values; tombstone records are skipped. When **merge** removes an old segment, the matching **`.hint`** is removed with it. +- **Put / Get / Delete**: basic APIs with a process-wide `RWMutex`. +- **ListKeys / Fold**: `ListKeys` returns keys sorted lexicographically; `Fold` walks keys in that order and reads each value (read lock held for the scan). +- **Sync / Close**: `DB.Sync` fsyncs the active segment; `DB.Close` syncs then closes segment files and releases the lock file handle. +- **Segment rotation**: configurable `Options.SegmentSize` (default 256 MiB). +- **On-disk record layout**: fixed meta (CRC32, timestamps, sizes, flag) + key + value (`entity/entry.go`). Tombstone records store the key with `ValueSize` 0. +- **Merge**: rewrites live entries from old segments and removes merged files; tombstone records in old files are skipped during merge. +- **CRC on read**: Enabled by default; disable with `Options.VerifyCRC = false` if needed. +- **Recovery**: Full segment scans apply tombstones in order (remove key from keydir) and populate `DataPosition.Timestamp` from record meta; hint recovery skips tombstone rows. +- **Tests**: `db_test.go` covers CRUD, rotation, merge, delete+merge, hint recovery, merge after reopen, tombstone recovery, CRC failure, ListKeys/Fold, read-only open; `storage/hint_test.go` and `entity/entry_test.go` cover hint encoding and CRC/tombstones (requires `github.com/stretchr/testify`). + +--- + +## Gaps vs. the Bitcask paper and production-grade behavior -另外我想实现做的实验和实现的想法会记录在项目TODO中。并切换分支进行相关迭代，有效果的部分会合并进入master分支。实践的相关文章会列在文章列表上。感谢各位的关注，希望各位都能从中学到一些东西。欢迎star，欢迎提PR。 +Some items from [bitcask-intro.pdf](https://riak.com/assets/bitcask-intro.pdf) and typical production engines are still out of scope or partial: -## Todo +1. **Merge + hint generation** — Merge rewrites live data into the active file and deletes merged segments (and their hints); a new hint is written only when that segment is **rotated** (normal append-only behavior). +2. **Portability** — Advisory locking is implemented on Unix (`flock`). On other platforms the lock is a no-op; use a single process or external coordination. +3. **API breadth** — No key prefix / range iterator beyond sorted `ListKeys` + `Fold`. No snapshot or MVCC reads. +4. **Durability policy** — `Sync` is explicit; there is no `Sync` after every write (call `Sync` when you need durability beyond process crash). -- [ ] 实现HintFile -- [ ] 探究对map的优化 -- [ ] 实现version control +--- -## 文章list +## Code map -1. [基于Bitcask实现简单的kv存储详细讲解](https://mp.weixin.qq.com/s?__biz=Mzg5MzU5NzQxMA==&mid=2247483844&idx=1&sn=2fc13cf8ce7c465dbd08690c56eaba69&chksm=c02d2249f75aab5f0955377c6ed29f8529c4a5f18bd53f27ab4cd88b85b3792af370f8a378ab#rd) +| Path | Purpose | +|------|---------| +| `db.go` | `NewDB`, `Get` / `Set` / `Delete`, `Merge`, `ListKeys`, `Fold`, `Sync`, `Close`, `recovery` | +| `lock_unix.go`, `lock_other.go` | Optional advisory DB lock | +| `index/index.go` | Keydir (`map` + `DataPosition`) | +| `storage/datafiles.go` | Active/old files, rotation, read/write entries, CRC, `Sync`/`Close` | +| `storage/hint.go` | Hint file format, write on rotation, read/remove with segments | +| `entity/entry.go` | Binary encoding of records, tombstones, `VerifyRecordCRC` | +| `options.go` | `Dir`, `SegmentSize`, `VerifyCRC`, `ReadOnly`, `ExclusiveLock` | -## 个人 +--- -下面是本人微信公众号，欢迎关注 +## Todo (paper-aligned and robustness) -![image](https://user-images.githubusercontent.com/92676541/226180799-973944bd-5c75-4a9b-8226-6c7e6e465d19.png) +- [x] **Hint files**: format, write on rotation, load hint instead of full scan on open when present. +- [x] **Recovery**: populate old-file ID list from on-disk segments so `Merge` works after restart; timestamps from full scan recovery. +- [x] **Merge**: compare keydir `(fid, offset)` to the **start** offset of each scanned entry; test `TestDB_Merge_LiveKeyOnlyInOldSegment`. +- [x] **Delete / tombstones**: valid meta + key for tombstone records; recovery applies tombstones; merge skips tombstone bodies. +- [x] **CRC verification** on read (`Options.VerifyCRC`, default `true`). +- [x] **`Sync` / `Close`**: flush and fsync active file; close FDs and lock. +- [x] **API**: `ListKeys`, `Fold`, read-only open (`Options.ReadOnly`), Unix advisory lock (`Options.ExclusiveLock`). diff --git a/db.go b/db.go index 8da01a3..72f33c2 100644 --- a/db.go +++ b/db.go @@ -2,7 +2,9 @@ package tiny_bitcask import ( "errors" + "fmt" "io" + "os" "sort" "sync" "tiny-bitcask/entity" @@ -13,13 +15,15 @@ import ( var ( KeyNotFoundErr = errors.New("key not found") NoNeedToMergeErr = errors.New("no need to merge") + ReadOnlyDBErr = errors.New("read-only database") ) type DB struct { - rw sync.RWMutex - kd *index.KeyDir - storage *storage.DataFiles - opt *Options + rw sync.RWMutex + kd *index.KeyDir + storage *storage.DataFiles + opt *Options + lockFile *os.File } // NewDB create a new DB instance with Options @@ -27,24 +31,124 @@ func NewDB(opt *Options) (db *DB, err error) { db = &DB{} db.kd = index.NewKD() db.opt = opt - if isExist, _ := isDirExist(opt.Dir); isExist { + + exists, err := isDirExist(opt.Dir) + if err != nil { + return nil, err + } + if opt.ReadOnly && !exists { + return nil, fmt.Errorf("tiny-bitcask: read-only open: %w", os.ErrNotExist) + } + + if exists { + lf, err := acquireDBLock(opt.Dir, opt.ReadOnly, opt.ExclusiveLock) + if err != nil { + return nil, err + } + db.lockFile = lf if err := db.recovery(opt); err != nil { + _ = db.closeStorageAndLock() return nil, err } return db, nil } + var fileSize = getSegmentSize(opt.SegmentSize) - db.storage, err = storage.NewDataFiles(opt.Dir, fileSize) + db.storage, err = storage.NewDataFiles(opt.Dir, fileSize, opt.VerifyCRC) + if err != nil { + return nil, err + } + lf, err := acquireDBLock(opt.Dir, false, opt.ExclusiveLock) if err != nil { + _ = db.storage.Close() return nil, err } - return db, err + db.lockFile = lf + return db, nil +} + +func (db *DB) closeStorageAndLock() error { + var first error + if db.storage != nil { + if err := db.storage.Close(); err != nil && first == nil { + first = err + } + db.storage = nil + } + if db.lockFile != nil { + _ = db.lockFile.Close() + db.lockFile = nil + } + return first +} + +// Sync flushes the active data file (fsync). Thread-safe. +func (db *DB) Sync() error { + db.rw.Lock() + defer db.rw.Unlock() + if db.storage == nil { + return nil + } + return db.storage.Sync() +} + +// Close syncs and releases file descriptors and the advisory lock. +func (db *DB) Close() error { + db.rw.Lock() + defer db.rw.Unlock() + if err := db.storageSyncBestEffort(); err != nil { + _ = db.closeStorageAndLock() + return err + } + return db.closeStorageAndLock() +} + +func (db *DB) storageSyncBestEffort() error { + if db.storage == nil { + return nil + } + return db.storage.Sync() +} + +// ListKeys returns all keys in lexicographic order (snapshot under read lock). +func (db *DB) ListKeys() [][]byte { + db.rw.RLock() + defer db.rw.RUnlock() + keys := db.kd.SortedKeys() + out := make([][]byte, len(keys)) + for i, k := range keys { + out[i] = []byte(k) + } + return out +} + +// Fold visits every key in sorted order and calls fn with the current value. Holds one read lock for the scan. +func (db *DB) Fold(fn func(key, value []byte) error) error { + db.rw.RLock() + defer db.rw.RUnlock() + for _, k := range db.kd.SortedKeys() { + dp := db.kd.Find(k) + if dp == nil { + continue + } + entry, err := db.storage.ReadEntry(dp) + if err != nil { + return err + } + if err := fn([]byte(k), entry.Value); err != nil { + return err + } + } + return nil } // Set sets a key-value pairs into DB func (db *DB) Set(key []byte, value []byte) error { db.rw.Lock() defer db.rw.Unlock() + if db.opt.ReadOnly { + return ReadOnlyDBErr + } entry := entity.NewEntryWithData(key, value) h, err := db.storage.WriterEntity(entry) if err != nil { @@ -73,13 +177,15 @@ func (db *DB) Get(key []byte) (value []byte, err error) { func (db *DB) Delete(key []byte) error { db.rw.Lock() defer db.rw.Unlock() + if db.opt.ReadOnly { + return ReadOnlyDBErr + } keyStr := string(key) index := db.kd.Find(keyStr) if index == nil { return KeyNotFoundErr } - e := entity.NewEntry() - e.Meta.Flag = entity.DeleteFlag + e := entity.NewTombstoneEntry(key) _, err := db.storage.WriterEntity(e) if err != nil { return err @@ -88,76 +194,135 @@ func (db *DB) Delete(key []byte) error { return nil } -// Merge clean the useless data +// Merge compacts old segments: copies live records still stored only in mergeable +// files into the active file, then removes those segment files. func (db *DB) Merge() error { db.rw.Lock() defer db.rw.Unlock() + if db.opt.ReadOnly { + return ReadOnlyDBErr + } fids := db.storage.GetOldFiles() if len(fids) < 2 { return NoNeedToMergeErr } - sort.Ints(fids) - for _, fid := range fids[:len(fids)-1] { - var off int64 = 0 - reader := db.storage.GetOldFile(fid) - for { - entry, err := reader.ReadEntityWithOutLength(off) - if err == nil { - key := string(entry.Key) - off += entry.Size() - oldIndex := db.kd.Find(key) - if oldIndex == nil { - continue - } - if oldIndex.IsEqualPos(fid, off) { - h, err := db.storage.WriterEntity(entry) - if err != nil { - return err - } - db.kd.AddIndexByData(h, entry) - } - } else { - if err == io.EOF { - break - } - return err + toMerge := append([]int(nil), fids...) + sort.Ints(toMerge) + for _, fid := range toMerge[:len(toMerge)-1] { + if err := db.mergeOldFile(fid); err != nil { + return err + } + } + return nil +} + +func (db *DB) mergeOldFile(fid int) error { + reader := db.storage.GetOldFile(fid) + var off int64 + for { + entry, err := reader.ReadEntityWithOutLength(off) + if err != nil { + if err == io.EOF { + return db.storage.RemoveFile(fid) } + return err } - err := db.storage.RemoveFile(fid) + if entry.Meta.Flag == entity.DeleteFlag { + off += entry.Size() + continue + } + // entryOff is the record start offset; keydir stores the same (see IsEqualPos). + entryOff := off + off += entry.Size() + + idx := db.kd.Find(string(entry.Key)) + if idx == nil || !idx.IsEqualPos(fid, entryOff) { + continue + } + h, err := db.storage.WriterEntity(entry) if err != nil { return err } + db.kd.AddIndexByData(h, entry) } - return nil } // recovery will rebuild a db from existing dir func (db *DB) recovery(opt *Options) (err error) { var fileSize = getSegmentSize(opt.SegmentSize) - db.storage, err = storage.NewDataFileWithFiles(opt.Dir, fileSize) + db.storage, err = storage.NewDataFileWithFiles(opt.Dir, fileSize, opt.VerifyCRC, opt.ReadOnly) if err != nil { return err } - fids := db.storage.GetOldFiles() - sort.Ints(fids) - for _, fid := range fids { - var off int64 = 0 - reader := db.storage.GetOldFile(fid) - for { - entry, err := reader.ReadEntityWithOutLength(off) - if err == nil { - db.kd.AddIndexByRawInfo(fid, off, entry.Key, entry.Value) - off += entry.Size() + fids, err := storage.ListDataFileIDs(opt.Dir) + if err != nil { + return err + } + for i, fid := range fids { + isActive := i == len(fids)-1 + if err := db.recoverSegment(fid, opt.Dir, isActive, opt.VerifyCRC); err != nil { + return err + } + } + return nil +} + +func (db *DB) recoverFromHint(fid int, dir string) error { + recs, err := storage.ReadHintFile(dir, fid) + if err != nil { + return err + } + datPath := storage.DataFilePath(dir, fid) + st, err := os.Stat(datPath) + if err != nil { + return err + } + datSize := st.Size() + for _, r := range recs { + if r.Flag == entity.DeleteFlag { + continue + } + if int(r.KeySize) != len(r.Key) { + return errors.New("hint key length mismatch") + } + recLen := int64(entity.MetaSize + r.KeySize + r.ValueSize) + if r.RecordOffset < 0 || r.RecordOffset+recLen > datSize { + return errors.New("hint record out of range for data file") + } + db.kd.AddIndexBySizes(fid, r.RecordOffset, r.Key, int(r.KeySize), int(r.ValueSize), r.Timestamp) + } + return nil +} + +func (db *DB) recoverSegment(fid int, dir string, isActive bool, verifyCRC bool) error { + if !isActive && storage.HintFileExists(dir, fid) { + if err := db.recoverFromHint(fid, dir); err == nil { + return nil + } + } + + path := storage.DataFilePath(dir, fid) + of, err := storage.NewOldFile(path, verifyCRC) + if err != nil { + return err + } + defer of.Close() + var off int64 + for { + entry, err := of.ReadEntityWithOutLength(off) + if err == nil { + if entry.Meta.Flag == entity.DeleteFlag { + db.kd.Delete(string(entry.Key)) } else { - if err == storage.DeleteEntryErr { - continue - } - if err == io.EOF { - break - } - return err + db.kd.AddIndexByRawInfo(fid, off, entry.Key, entry.Value, entry.Meta.TimeStamp) } + off += entry.Size() + } else { + if err == io.EOF { + break + } + return err } } - return err + return nil } diff --git a/db_test.go b/db_test.go index 208f497..0581f92 100644 --- a/db_test.go +++ b/db_test.go @@ -2,116 +2,446 @@ package tiny_bitcask import ( "fmt" - "github.com/stretchr/testify/assert" "os" + "path/filepath" "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" "tiny-bitcask/storage" ) -func BitCaskTest(t *testing.T, opt *Options, test func(db *DB)) { - if opt == nil { - opt = DefaultOptions +// newTestDB opens a DB under an isolated t.TempDir() with optional option overrides. +func newTestDB(t *testing.T, customize func(*Options)) *DB { + t.Helper() + opt := *DefaultOptions + // Use a non-existent subdir: t.TempDir() exists but is empty, which would + // incorrectly trigger recovery in NewDB on an empty store. + opt.Dir = filepath.Join(t.TempDir(), "db") + if customize != nil { + customize(&opt) } - db, err := NewDB(opt) + db, err := NewDB(&opt) assert.NoError(t, err) - test(db) - os.RemoveAll(opt.Dir) + return db } -func TestDB_Base(t *testing.T) { - var test = func(db *DB) { - err := db.Set([]byte("test_key"), []byte("test_value")) - assert.NoError(t, err) - value, err := db.Get([]byte("test_key")) - assert.NoError(t, err) - assert.Equal(t, "test_value", string(value)) - - err = db.Set([]byte("test_key"), []byte("test_value_2")) - assert.NoError(t, err) - - value, err = db.Get([]byte("test_key")) - assert.NoError(t, err) - assert.Equal(t, "test_value_2", string(value)) +func TestDB_SetGetUpdate(t *testing.T) { + tests := []struct { + name string + run func(t *testing.T, db *DB) + }{ + { + name: "read_after_write", + run: func(t *testing.T, db *DB) { + assert.NoError(t, db.Set([]byte("test_key"), []byte("test_value"))) + value, err := db.Get([]byte("test_key")) + assert.NoError(t, err) + assert.Equal(t, "test_value", string(value)) + }, + }, + { + name: "update_overwrites", + run: func(t *testing.T, db *DB) { + assert.NoError(t, db.Set([]byte("test_key"), []byte("test_value"))) + assert.NoError(t, db.Set([]byte("test_key"), []byte("test_value_2"))) + value, err := db.Get([]byte("test_key")) + assert.NoError(t, err) + assert.Equal(t, "test_value_2", string(value)) + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db := newTestDB(t, nil) + tt.run(t, db) + }) } - BitCaskTest(t, nil, test) } func TestDB_SegmentSize(t *testing.T) { - opt := &Options{ - Dir: "db", - SegmentSize: 4 * storage.KB, - } - var test = func(db *DB) { - for i := 0; i < 1000; i++ { - key := fmt.Sprintf("test_key_%d", i) - value := fmt.Sprintf("test_value_%d", i) - err := db.Set([]byte(key), []byte(value)) - assert.NoError(t, err) - } + tests := []struct { + name string + keys int + segmentSize int64 + }{ + { + name: "many_keys_small_segment", + keys: 1000, + segmentSize: 4 * storage.KB, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db := newTestDB(t, func(o *Options) { + o.SegmentSize = tt.segmentSize + }) + for i := 0; i < tt.keys; i++ { + key := fmt.Sprintf("test_key_%d", i) + value := fmt.Sprintf("test_value_%d", i) + assert.NoError(t, db.Set([]byte(key), []byte(value))) + } + }) } - BitCaskTest(t, opt, test) } func TestDB_Merge(t *testing.T) { - opt := &Options{ - Dir: "db", - SegmentSize: 4 * storage.KB, - } - var test = func(db *DB) { - key := "test_key" - for i := 0; i < 1000; i++ { - value := fmt.Sprintf("test_value_%d", i) - err := db.Set([]byte(key), []byte(value)) + tests := []struct { + name string + updates int + segmentSize int64 + want string + }{ + { + name: "repeated_key_compacted_to_latest", + updates: 1000, + segmentSize: 4 * storage.KB, + want: "test_value_999", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db := newTestDB(t, func(o *Options) { + o.SegmentSize = tt.segmentSize + }) + key := []byte("test_key") + for i := 0; i < tt.updates; i++ { + value := fmt.Sprintf("test_value_%d", i) + assert.NoError(t, db.Set(key, []byte(value))) + } + assert.NoError(t, db.Merge()) + value, err := db.Get(key) assert.NoError(t, err) - } - err := db.Merge() - assert.NoError(t, err) - value, err := db.Get([]byte("test_key")) - assert.NoError(t, err) - assert.Equal(t, "test_value_999", string(value)) - } - BitCaskTest(t, opt, test) + assert.Equal(t, tt.want, string(value)) + }) + } } -func TestDB_Delete(t *testing.T) { - var test = func(db *DB) { - err := db.Set([]byte("test_key"), []byte("test_value")) - assert.NoError(t, err) - value, err := db.Get([]byte("test_key")) +// TestDB_Merge_LiveKeyOnlyInOldSegment checks that merge copies a record whose +// keydir entry still points at an old segment (not the active file) before +// that segment is removed. +func TestDB_Merge_LiveKeyOnlyInOldSegment(t *testing.T) { + db := newTestDB(t, func(o *Options) { + o.SegmentSize = 4 * storage.KB + }) + staleKey := []byte("written_once_then_idle") + assert.NoError(t, db.Set(staleKey, []byte("keep_me"))) + + busyKey := []byte("busy") + for i := 0; i < 800; i++ { + assert.NoError(t, db.Set(busyKey, []byte(fmt.Sprintf("v_%d", i)))) + } + + assert.NoError(t, db.Merge()) + + got, err := db.Get(staleKey) + assert.NoError(t, err) + assert.Equal(t, "keep_me", string(got)) +} + +func TestDB_Recovery(t *testing.T) { + tests := []struct { + name string + }{ + {name: "reopen_loads_existing_data"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + dataDir := filepath.Join(t.TempDir(), "data") + opt := *DefaultOptions + opt.Dir = dataDir + + db1, err := NewDB(&opt) + assert.NoError(t, err) + assert.NoError(t, db1.Set([]byte("k"), []byte("v"))) + assert.NoError(t, db1.Close()) + + db2, err := NewDB(&opt) + assert.NoError(t, err) + v, err := db2.Get([]byte("k")) + assert.NoError(t, err) + assert.Equal(t, "v", string(v)) + }) + } +} + +// TestDB_Recovery_UsesHintFile checks sealed segments get a .hint on rotation and +// recovery rebuilds the keydir from hints (active segment still full-scanned). +func TestDB_Recovery_UsesHintFile(t *testing.T) { + dataDir := filepath.Join(t.TempDir(), "hintdb") + opt := *DefaultOptions + opt.Dir = dataDir + opt.SegmentSize = 4 * storage.KB + + db1, err := NewDB(&opt) + require.NoError(t, err) + // Fill segment 1 then rotate into segment 2. + busy := []byte("busy") + for i := 0; i < 800; i++ { + require.NoError(t, db1.Set(busy, []byte(fmt.Sprintf("v_%d", i)))) + } + staleKey := []byte("only_in_segment_1") + require.NoError(t, db1.Set(staleKey, []byte("pinned"))) + + assert.True(t, storage.HintFileExists(dataDir, 1), "rotation should write 1.hint") + + require.NoError(t, db1.Close()) + + db2, err := NewDB(&opt) + require.NoError(t, err) + got, err := db2.Get(staleKey) + require.NoError(t, err) + assert.Equal(t, "pinned", string(got)) +} + +// TestDB_Recovery_InvalidHintFallsBackToScan ensures a corrupt hint does not brick recovery. +// TestDB_Recovery_OldFileIDsEnableMerge checks that after reopening a store with +// multiple segment files, GetOldFiles is populated so Merge does not return +// NoNeedToMergeErr until new rotations (gap fixed: recovery lists sealed IDs). +func TestDB_Recovery_OldFileIDsEnableMerge(t *testing.T) { + tests := []struct { + name string + writes int + segmentSize int64 + wantValue string + }{ + { + name: "merge_runs_after_reopen_multi_segment", + writes: 2500, + segmentSize: 4 * storage.KB, + wantValue: "v_2499", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + dataDir := filepath.Join(t.TempDir(), "merge_recovery") + opt := *DefaultOptions + opt.Dir = dataDir + opt.SegmentSize = tt.segmentSize + + db1, err := NewDB(&opt) + require.NoError(t, err) + key := []byte("merge_key") + for i := 0; i < tt.writes; i++ { + require.NoError(t, db1.Set(key, []byte(fmt.Sprintf("v_%d", i)))) + } + + require.NoError(t, db1.Close()) + + db2, err := NewDB(&opt) + require.NoError(t, err) + err = db2.Merge() + require.NoError(t, err, "merge should run after recovery when multiple sealed segments exist") - assert.NoError(t, err) - assert.Equal(t, "test_value", string(value)) + got, err := db2.Get(key) + require.NoError(t, err) + assert.Equal(t, tt.wantValue, string(got)) + }) + } +} - err = db.Delete([]byte("test_key")) - assert.NoError(t, err) +func TestDB_Recovery_InvalidHintFallsBackToScan(t *testing.T) { + dataDir := filepath.Join(t.TempDir(), "badhint") + opt := *DefaultOptions + opt.Dir = dataDir + opt.SegmentSize = 4 * storage.KB - value, err = db.Get([]byte("test_key")) - assert.Nil(t, value) - assert.ErrorAs(t, KeyNotFoundErr, &err) + db1, err := NewDB(&opt) + require.NoError(t, err) + busy := []byte("busy") + for i := 0; i < 800; i++ { + require.NoError(t, db1.Set(busy, []byte(fmt.Sprintf("v_%d", i)))) + } + require.NoError(t, db1.Set([]byte("keep"), []byte("me"))) + + require.NoError(t, os.WriteFile(filepath.Join(dataDir, "1.hint"), []byte("TBHK\x01\x00\x00\x00\x00trunc"), 0o644)) + + require.NoError(t, db1.Close()) + + db2, err := NewDB(&opt) + require.NoError(t, err) + got, err := db2.Get([]byte("keep")) + require.NoError(t, err) + assert.Equal(t, "me", string(got)) +} + +func TestDB_Delete(t *testing.T) { + tests := []struct { + name string + run func(t *testing.T, db *DB) + }{ + { + name: "get_returns_not_found_after_delete", + run: func(t *testing.T, db *DB) { + assert.NoError(t, db.Set([]byte("test_key"), []byte("test_value"))) + value, err := db.Get([]byte("test_key")) + assert.NoError(t, err) + assert.Equal(t, "test_value", string(value)) + + assert.NoError(t, db.Delete([]byte("test_key"))) + + value, err = db.Get([]byte("test_key")) + assert.Nil(t, value) + assert.ErrorIs(t, err, KeyNotFoundErr) + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db := newTestDB(t, nil) + tt.run(t, db) + }) } - BitCaskTest(t, nil, test) } func TestDB_Delete_Merge(t *testing.T) { - opt := &Options{ - Dir: "db", - SegmentSize: 4 * storage.KB, - } - var test = func(db *DB) { - key := "test_key" - for i := 0; i < 1000; i++ { - value := fmt.Sprintf("test_value_%d", i) - err := db.Set([]byte(key), []byte(value)) - assert.NoError(t, err) - } - err := db.Delete([]byte("test_key")) - assert.NoError(t, err) - err = db.Merge() - assert.NoError(t, err) + tests := []struct { + name string + updates int + segmentSize int64 + }{ + { + name: "deleted_key_stays_absent_after_merge", + updates: 1000, + segmentSize: 4 * storage.KB, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db := newTestDB(t, func(o *Options) { + o.SegmentSize = tt.segmentSize + }) + key := []byte("test_key") + for i := 0; i < tt.updates; i++ { + value := fmt.Sprintf("test_value_%d", i) + assert.NoError(t, db.Set(key, []byte(value))) + } + assert.NoError(t, db.Delete(key)) + assert.NoError(t, db.Merge()) + + value, err := db.Get(key) + assert.Nil(t, value) + assert.ErrorIs(t, err, KeyNotFoundErr) + }) + } +} + +func TestDB_Recovery_TombstoneRemovesKey(t *testing.T) { + tests := []struct { + name string + }{ + {name: "delete_then_reopen"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + dataDir := filepath.Join(t.TempDir(), "tombdb") + opt := *DefaultOptions + opt.Dir = dataDir + db1, err := NewDB(&opt) + require.NoError(t, err) + require.NoError(t, db1.Set([]byte("k"), []byte("v"))) + require.NoError(t, db1.Delete([]byte("k"))) + require.NoError(t, db1.Close()) + + db2, err := NewDB(&opt) + require.NoError(t, err) + defer db2.Close() + _, err = db2.Get([]byte("k")) + assert.ErrorIs(t, err, KeyNotFoundErr) + }) + } +} + +func TestDB_Get_CRCCorrupt(t *testing.T) { + tests := []struct { + name string + }{ + {name: "read_returns_crc_error"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + dataDir := filepath.Join(t.TempDir(), "crcdb") + opt := *DefaultOptions + opt.Dir = dataDir + db, err := NewDB(&opt) + require.NoError(t, err) + require.NoError(t, db.Set([]byte("k"), []byte("v"))) + _, err = db.Get([]byte("k")) + require.NoError(t, err) + + dat := filepath.Join(dataDir, "1.dat") + b, err := os.ReadFile(dat) + require.NoError(t, err) + require.NotEmpty(t, b) + b[len(b)-1] ^= 0xFF + require.NoError(t, os.WriteFile(dat, b, 0o644)) + + _, err = db.Get([]byte("k")) + assert.ErrorIs(t, err, storage.CrcErr) + require.NoError(t, db.Close()) + }) + } +} + +func TestDB_ListKeys_Fold(t *testing.T) { + tests := []struct { + name string + }{ + {name: "sorted_keys_and_fold"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + db := newTestDB(t, nil) + require.NoError(t, db.Set([]byte("b"), []byte("2"))) + require.NoError(t, db.Set([]byte("a"), []byte("1"))) + + keys := db.ListKeys() + require.Len(t, keys, 2) + assert.Equal(t, "a", string(keys[0])) + assert.Equal(t, "b", string(keys[1])) + + var sum string + err := db.Fold(func(k, v []byte) error { + sum += string(k) + "=" + string(v) + ";" + return nil + }) + require.NoError(t, err) + assert.Contains(t, sum, "a=1") + assert.Contains(t, sum, "b=2") + }) + } +} + +func TestDB_ReadOnly(t *testing.T) { + tests := []struct { + name string + }{ + {name: "get_ok_writes_rejected"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + dataDir := filepath.Join(t.TempDir(), "rodb") + opt := *DefaultOptions + opt.Dir = dataDir + db1, err := NewDB(&opt) + require.NoError(t, err) + require.NoError(t, db1.Set([]byte("k"), []byte("v"))) + require.NoError(t, db1.Close()) + + ro := *DefaultOptions + ro.Dir = dataDir + ro.ReadOnly = true + db2, err := NewDB(&ro) + require.NoError(t, err) + defer db2.Close() + + v, err := db2.Get([]byte("k")) + require.NoError(t, err) + assert.Equal(t, "v", string(v)) - value, err := db.Get([]byte("test_key")) - assert.Nil(t, value) - assert.Equal(t, KeyNotFoundErr, err) + assert.ErrorIs(t, db2.Set([]byte("x"), []byte("y")), ReadOnlyDBErr) + assert.ErrorIs(t, db2.Delete([]byte("k")), ReadOnlyDBErr) + assert.ErrorIs(t, db2.Merge(), ReadOnlyDBErr) + }) } - BitCaskTest(t, opt, test) } diff --git a/entity/entry.go b/entity/entry.go index 70eb8df..d7cce02 100644 --- a/entity/entry.go +++ b/entity/entry.go @@ -6,6 +6,16 @@ import ( "time" ) +// VerifyRecordCRC checks the stored CRC32 against meta+payload bytes (matches Encode). +func VerifyRecordCRC(buf []byte) bool { + if len(buf) < MetaSize { + return false + } + got := binary.LittleEndian.Uint32(buf[0:4]) + want := crc32.ChecksumIEEE(buf[4:]) + return got == want +} + const ( MetaSize = 29 DeleteFlag = 1 @@ -38,6 +48,17 @@ func NewEntryWithData(key []byte, value []byte) *Entry { return e } +// NewTombstoneEntry builds a delete record: key in payload, empty value, DeleteFlag. +func NewTombstoneEntry(key []byte) *Entry { + now := uint64(time.Now().Unix()) + meta := NewMeta(). + WithTimeStamp(now). + WithKeySize(uint32(len(key))). + WithValueSize(0). + WithFlag(DeleteFlag) + return NewEntry().WithKey(key).WithValue(nil).WithMeta(meta) +} + func (e *Entry) Encode() []byte { size := e.Size() buf := make([]byte, size) @@ -46,7 +67,9 @@ func (e *Entry) Encode() []byte { binary.LittleEndian.PutUint32(buf[20:24], e.Meta.KeySize) binary.LittleEndian.PutUint32(buf[24:28], e.Meta.ValueSize) buf[28] = e.Meta.Flag - if e.Meta.Flag != DeleteFlag { + if e.Meta.Flag == DeleteFlag { + copy(buf[MetaSize:MetaSize+len(e.Key)], e.Key) + } else { copy(buf[MetaSize:MetaSize+len(e.Key)], e.Key) copy(buf[MetaSize+len(e.Key):MetaSize+len(e.Key)+len(e.Value)], e.Value) } @@ -68,6 +91,7 @@ func (e *Entry) DecodeMeta(bytes []byte) { e.Meta.TimeStamp = binary.LittleEndian.Uint64(bytes[12:20]) e.Meta.KeySize = binary.LittleEndian.Uint32(bytes[20:24]) e.Meta.ValueSize = binary.LittleEndian.Uint32(bytes[24:28]) + e.Meta.Flag = bytes[28] } func (e *Entry) Size() int64 { diff --git a/index/index.go b/index/index.go index 08f9edf..058232b 100644 --- a/index/index.go +++ b/index/index.go @@ -1,6 +1,10 @@ package index -import "tiny-bitcask/entity" +import ( + "sort" + + "tiny-bitcask/entity" +) const ( KeyNotFound = "key not found" @@ -59,18 +63,50 @@ type DataPosition struct { } func (kd *KeyDir) AddIndexByData(hint *entity.Hint, entry *entity.Entry) { - kd.AddIndexByRawInfo(hint.Fid, hint.Off, entry.Key, entry.Value) + kd.AddIndexByRawInfo(hint.Fid, hint.Off, entry.Key, entry.Value, entry.Meta.TimeStamp) } -func (kd *KeyDir) AddIndexByRawInfo(fid int, off int64, key, value []byte) { - index := newDataPosition(fid, off, key, value) +func (kd *KeyDir) AddIndexByRawInfo(fid int, off int64, key, value []byte, ts uint64) { + index := newDataPosition(fid, off, key, value, ts) kd.Add(string(key), index) } -func newDataPosition(fid int, off int64, key, value []byte) *DataPosition { +// Range visits every key in arbitrary map order until fn returns false. +func (kd *KeyDir) Range(fn func(key string, dp *DataPosition) bool) { + for k, dp := range kd.Index { + if !fn(k, dp) { + break + } + } +} + +// SortedKeys returns keys in lexicographic order (copy). +func (kd *KeyDir) SortedKeys() []string { + keys := make([]string, 0, len(kd.Index)) + for k := range kd.Index { + keys = append(keys, k) + } + sort.Strings(keys) + return keys +} + +// AddIndexBySizes records keydir metadata without reading the value (e.g. hint recovery). +func (kd *KeyDir) AddIndexBySizes(fid int, off int64, key []byte, keySize, valueSize int, ts uint64) { + dp := &DataPosition{ + Fid: fid, + Off: off, + Timestamp: ts, + KeySize: keySize, + ValueSize: valueSize, + } + kd.Add(string(key), dp) +} + +func newDataPosition(fid int, off int64, key, value []byte, ts uint64) *DataPosition { dp := &DataPosition{} dp.Fid = fid dp.Off = off + dp.Timestamp = ts dp.KeySize = len(key) dp.ValueSize = len(value) return dp diff --git a/options.go b/options.go index 3420d16..be9ff9e 100644 --- a/options.go +++ b/options.go @@ -8,12 +8,18 @@ const ( var ( DefaultOptions = &Options{ - Dir: "db", - SegmentSize: DefaultSegmentSize, + Dir: "db", + SegmentSize: DefaultSegmentSize, + VerifyCRC: true, + ExclusiveLock: true, } ) +// Options configures the database. Zero value is not valid; use DefaultOptions or set fields explicitly. type Options struct { - Dir string - SegmentSize int64 + Dir string + SegmentSize int64 + VerifyCRC bool // verify CRC32 on every read (default true when using DefaultOptions) + ReadOnly bool // open existing store read-only (ListKeys, Get, Fold allowed) + ExclusiveLock bool // advisory flock on .tiny-bitcask.lock (Unix); shared lock when ReadOnly } diff --git a/storage/datafiles.go b/storage/datafiles.go index 3fd40a1..ef832d1 100644 --- a/storage/datafiles.go +++ b/storage/datafiles.go @@ -6,6 +6,7 @@ import ( "io/ioutil" "os" "path" + "sort" "strconv" "strings" "tiny-bitcask/entity" @@ -40,6 +41,8 @@ type DataFiles struct { segmentSize int64 active *ActiveFile olds map[int]*OldFile + verifyCRC bool + readOnly bool } func (dfs *DataFiles) GetOldFiles() []int { @@ -53,7 +56,7 @@ func (dfs *DataFiles) RemoveReader(fid int) error { func (dfs *DataFiles) AddReader(fid int) error { path := getFilePath(dfs.dir, fid) - reader, err := NewOldFile(path) + reader, err := NewOldFile(path, dfs.verifyCRC) if err != nil { return err } @@ -62,29 +65,38 @@ func (dfs *DataFiles) AddReader(fid int) error { } // NewDataFileWithFiles create a DataFiles with existing dir -func NewDataFileWithFiles(dir string, segmentSize int64) (dfs *DataFiles, err error) { +func NewDataFileWithFiles(dir string, segmentSize int64, verifyCRC bool, readOnly bool) (dfs *DataFiles, err error) { dfs = &DataFiles{ dir: dir, olds: newOldFiles(), segmentSize: segmentSize, + verifyCRC: verifyCRC, + readOnly: readOnly, } fids, err := getFids(dir) if err != nil { return nil, err } + if len(fids) == 0 { + return nil, fmt.Errorf("storage: no %s files in %s", FileSuffix, dir) + } aFid := fids[len(fids)-1] - dfs.active, err = NewActiveFile(dir, aFid) + dfs.active, err = NewActiveFile(dir, aFid, readOnly, verifyCRC) if err != nil { return nil, err } + if len(fids) > 1 { + dfs.oIds = make([]int, len(fids)-1) + copy(dfs.oIds, fids[:len(fids)-1]) + } if len(fids) == 1 { return dfs, nil } oldFids := fids[:len(fids)-1] for _, fid := range oldFids { path := getFilePath(dir, fid) - reader, err := NewOldFile(path) + reader, err := NewOldFile(path, verifyCRC) if err != nil { return nil, err } @@ -95,12 +107,12 @@ func NewDataFileWithFiles(dir string, segmentSize int64) (dfs *DataFiles, err er } // NewDataFiles create a DataFiles Object with an empty dir -func NewDataFiles(path string, segmentSize int64) (dfs *DataFiles, err error) { +func NewDataFiles(path string, segmentSize int64, verifyCRC bool) (dfs *DataFiles, err error) { err = os.Mkdir(path, os.ModePerm) if err != nil { return nil, err } - af, err := NewActiveFile(path, 1) + af, err := NewActiveFile(path, 1, false, verifyCRC) if err != nil { return nil, err } @@ -110,6 +122,8 @@ func NewDataFiles(path string, segmentSize int64) (dfs *DataFiles, err error) { active: af, olds: map[int]*OldFile{}, segmentSize: segmentSize, + verifyCRC: verifyCRC, + readOnly: false, } return dfs, nil } @@ -121,15 +135,18 @@ func (dfs *DataFiles) rotate() error { if err != nil { return err } - r := &OldFile{fd: fd} + r := &OldFile{fd: fd, verifyCRC: dfs.verifyCRC} dfs.olds[dfs.active.fid] = r dfs.oIds = append(dfs.oIds, aFid) - af, err := NewActiveFile(dfs.dir, aFid+1) + af, err := NewActiveFile(dfs.dir, aFid+1, dfs.readOnly, dfs.verifyCRC) if err != nil { return err } dfs.active = af + if err := WriteHintFileForDataFile(dfs.dir, aFid, dfs.verifyCRC); err != nil { + return err + } return nil } @@ -145,6 +162,31 @@ func (dfs *DataFiles) ReadEntry(index *index.DataPosition) (e *entity.Entry, err return of.ReadEntity(index.Off, dataSize) } +// Sync flushes the active segment to stable storage. +func (dfs *DataFiles) Sync() error { + return dfs.active.fd.Sync() +} + +// Close releases file descriptors for the active and old segments. +func (dfs *DataFiles) Close() error { + var first error + if dfs.active != nil && dfs.active.fd != nil { + if err := dfs.active.fd.Close(); err != nil && first == nil { + first = err + } + } + for _, of := range dfs.olds { + if of != nil && of.fd != nil { + if err := of.fd.Close(); err != nil && first == nil { + first = err + } + } + } + dfs.olds = nil + dfs.active = nil + return first +} + func (dfs *DataFiles) GetOldFile(fid int) *OldFile { return dfs.olds[fid] } @@ -160,11 +202,21 @@ func (dfs *DataFiles) RemoveFile(fid int) error { if err != nil { return err } + RemoveHintFile(dfs.dir, fid) delete(dfs.olds, fid) + for i, id := range dfs.oIds { + if id == fid { + dfs.oIds = append(dfs.oIds[:i], dfs.oIds[i+1:]...) + break + } + } return nil } func (dfs *DataFiles) WriterEntity(e entity.Entity) (h *entity.Hint, err error) { + if dfs.readOnly { + return nil, errors.New("storage: read-only database") + } h, err = dfs.active.WriterEntity(e) if err != nil { return nil, err @@ -183,14 +235,19 @@ func (dfs *DataFiles) canRotate() bool { } type ActiveFile struct { - fid int - fd *os.File - off int64 + fid int + fd *os.File + off int64 + verifyCRC bool } -func NewActiveFile(dir string, fid int) (af *ActiveFile, err error) { +func NewActiveFile(dir string, fid int, readOnly, verifyCRC bool) (af *ActiveFile, err error) { path := getFilePath(dir, fid) - fd, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, os.ModePerm) + flag := os.O_CREATE | os.O_RDWR + if readOnly { + flag = os.O_RDONLY + } + fd, err := os.OpenFile(path, flag, os.ModePerm) if err != nil { return nil, err } @@ -199,9 +256,10 @@ func NewActiveFile(dir string, fid int) (af *ActiveFile, err error) { return nil, err } af = &ActiveFile{ - fd: fd, - off: fi.Size(), - fid: fid, + fd: fd, + off: fi.Size(), + fid: fid, + verifyCRC: verifyCRC, } return af, nil } @@ -221,52 +279,63 @@ func (af *ActiveFile) WriterEntity(e entity.Entity) (h *entity.Hint, err error) } func (af *ActiveFile) ReadEntity(off int64, length int) (e *entity.Entry, err error) { - return readEntry(af.fd, off, length) + return readEntry(af.fd, off, length, af.verifyCRC) } type OldFile struct { - fd *os.File + fd *os.File + verifyCRC bool } -func NewOldFile(path string) (of *OldFile, err error) { +func NewOldFile(path string, verifyCRC bool) (of *OldFile, err error) { fd, err := os.OpenFile(path, os.O_RDONLY, os.ModePerm) if err != nil { return nil, err } - of = &OldFile{fd: fd} + of = &OldFile{fd: fd, verifyCRC: verifyCRC} return of, nil } +func (of *OldFile) Close() error { + return of.fd.Close() +} + func (of *OldFile) ReadEntity(off int64, length int) (e *entity.Entry, err error) { - return readEntry(of.fd, off, length) + return readEntry(of.fd, off, length, of.verifyCRC) } func (of *OldFile) ReadEntityWithOutLength(off int64) (e *entity.Entry, err error) { e = entity.NewEntry().WithMeta(entity.NewMeta()) - buf := make([]byte, entity.MetaSize) - n, err := of.fd.ReadAt(buf, off) + metaBuf := make([]byte, entity.MetaSize) + n, err := of.fd.ReadAt(metaBuf, off) if err != nil { return nil, err } if n < entity.MetaSize { return nil, ReadMissDataErr } - off += entity.MetaSize - e.DecodeMeta(buf) + payloadOff := off + entity.MetaSize + e.DecodeMeta(metaBuf) payloadSize := e.Meta.KeySize + e.Meta.ValueSize - buf = make([]byte, payloadSize) - n, err = of.fd.ReadAt(buf, off) + payloadBuf := make([]byte, payloadSize) + n, err = of.fd.ReadAt(payloadBuf, payloadOff) if err != nil { return nil, err } if n < int(payloadSize) { return nil, ReadMissDataErr } - e.DecodePayload(buf) + if of.verifyCRC { + full := append(metaBuf, payloadBuf...) + if !entity.VerifyRecordCRC(full) { + return nil, CrcErr + } + } + e.DecodePayload(payloadBuf) return e, nil } -func readEntry(fd *os.File, off int64, length int) (e *entity.Entry, err error) { +func readEntry(fd *os.File, off int64, length int, verifyCRC bool) (e *entity.Entry, err error) { buf := make([]byte, length) n, err := fd.ReadAt(buf, off) if n < length { @@ -275,12 +344,25 @@ func readEntry(fd *os.File, off int64, length int) (e *entity.Entry, err error) if err != nil { return nil, err } - e = entity.NewEntry() + if verifyCRC && !entity.VerifyRecordCRC(buf) { + return nil, CrcErr + } + e = entity.NewEntry().WithMeta(entity.NewMeta()) e.DecodeMeta(buf[:entity.MetaSize]) e.DecodePayload(buf[entity.MetaSize:]) return e, nil } +// ListDataFileIDs returns numeric file IDs for all segment files in dir. +func ListDataFileIDs(dir string) (fids []int, err error) { + return getFids(dir) +} + +// DataFilePath returns the path to a segment file. +func DataFilePath(dir string, fid int) string { + return getFilePath(dir, fid) +} + func getFids(dir string) (fids []int, err error) { files, err := ioutil.ReadDir(dir) if err != nil { @@ -298,6 +380,7 @@ func getFids(dir string) (fids []int, err error) { fids = append(fids, fid) } } + sort.Ints(fids) return fids, nil }