From 35fa840a57bb3d5a00f6f342923febe37ebd3df5 Mon Sep 17 00:00:00 2001 From: Ivers Date: Thu, 30 Jul 2026 18:17:32 -0400 Subject: [PATCH 1/9] Did some much need refactoring for the skiplist and memtable structs, to separate them into 2 different packages. This makes the code easier to read and removes the confusing interface that we had in the sstable that was used for compaction --- internals/sstable/corruption_test.go | 5 ----- internals/sstable/merger.go | 8 -------- internals/sstable/reader_test.go | 18 +++++++++--------- internals/sstable/writer.go | 3 ++- internals/sstable/writer_test.go | 12 ++---------- 5 files changed, 13 insertions(+), 33 deletions(-) delete mode 100644 internals/sstable/merger.go diff --git a/internals/sstable/corruption_test.go b/internals/sstable/corruption_test.go index 0fc7e13..e3bc574 100644 --- a/internals/sstable/corruption_test.go +++ b/internals/sstable/corruption_test.go @@ -6,7 +6,6 @@ import ( "path/filepath" "testing" - "github.com/Vince-maple-byte/KeyData/internals/memtable" "github.com/Vince-maple-byte/KeyData/internals/sstable" ) @@ -407,10 +406,6 @@ func TestCompact_IgnoresFileWithTruncatedRecord(t *testing.T) { func TestCompact_FileWithSwappedMagic(t *testing.T) { dir := t.TempDir() - sstable.MergeList = func() sstable.ListMerger { - return memtable.CreateSkiplist() - } - writeSST := func(name, key, val string, corruptMagic bool) { rec := buildValidRecord(key, val, 1) footer := buildValidFooter(uint64(len(rec))) diff --git a/internals/sstable/merger.go b/internals/sstable/merger.go deleted file mode 100644 index 4e3a751..0000000 --- a/internals/sstable/merger.go +++ /dev/null @@ -1,8 +0,0 @@ -package sstable - -type ListMerger interface { - Insert(key string, value []byte) - EntireList() [][]byte -} - -var MergeList func() ListMerger diff --git a/internals/sstable/reader_test.go b/internals/sstable/reader_test.go index dc93545..bfef5f1 100644 --- a/internals/sstable/reader_test.go +++ b/internals/sstable/reader_test.go @@ -7,8 +7,8 @@ import ( "strconv" "testing" - "github.com/Vince-maple-byte/KeyData/internals/memtable" "github.com/Vince-maple-byte/KeyData/internals/record" + "github.com/Vince-maple-byte/KeyData/internals/skiplist" "github.com/Vince-maple-byte/KeyData/internals/sstable" ) @@ -22,7 +22,7 @@ func TestReadFile(t *testing.T) { { testName: "key_size_1", data: func() [][]byte { - d := memtable.CreateSkiplist() + d := skiplist.CreateSkiplist() for i := range 3200 { r, _ := record.CreateRecord(strconv.Itoa(i), strconv.Itoa(i+1), "PUT") d.Insert(strconv.Itoa(i), r) @@ -36,7 +36,7 @@ func TestReadFile(t *testing.T) { { testName: "key_size_2", data: func() [][]byte { - d := memtable.CreateSkiplist() + d := skiplist.CreateSkiplist() for i := range 3200 { r, _ := record.CreateRecord(strconv.Itoa(i), strconv.Itoa(i+1), "PUT") @@ -50,7 +50,7 @@ func TestReadFile(t *testing.T) { { testName: "key_size_3", data: func() [][]byte { - d := memtable.CreateSkiplist() + d := skiplist.CreateSkiplist() for i := range 3200 { r, _ := record.CreateRecord(strconv.Itoa(i), strconv.Itoa(i+1), "PUT") @@ -64,7 +64,7 @@ func TestReadFile(t *testing.T) { { testName: "key_size_4", data: func() [][]byte { - d := memtable.CreateSkiplist() + d := skiplist.CreateSkiplist() for i := range 3200 { r, _ := record.CreateRecord(strconv.Itoa(i), strconv.Itoa(i+1), "PUT") @@ -78,7 +78,7 @@ func TestReadFile(t *testing.T) { { testName: "key_is_a_direct_match_to_the_index", data: func() [][]byte { - d := memtable.CreateSkiplist() + d := skiplist.CreateSkiplist() for i := range 3200 { r, _ := record.CreateRecord(strconv.Itoa(i), strconv.Itoa(i+1), "PUT") @@ -92,7 +92,7 @@ func TestReadFile(t *testing.T) { { testName: "key_if_file_truncated", data: func() [][]byte { - d := memtable.CreateSkiplist() + d := skiplist.CreateSkiplist() for i := range 240 { r, _ := record.CreateRecord(strconv.Itoa(i), strconv.Itoa(i+1), "PUT") @@ -106,7 +106,7 @@ func TestReadFile(t *testing.T) { { testName: "key_if_larger_than_normal", data: func() [][]byte { - d := memtable.CreateSkiplist() + d := skiplist.CreateSkiplist() for i := range 7000 { r, _ := record.CreateRecord(strconv.Itoa(i), strconv.Itoa(i+1), "PUT") @@ -154,7 +154,7 @@ func TestReadFromAllFiles(t *testing.T) { dir := t.TempDir() fmt.Println("File Directory:", dir) for i := range 10 { - d := memtable.CreateSkiplist() + d := skiplist.CreateSkiplist() for i := range 3000 * (i + 1) { r, _ := record.CreateRecord(strconv.Itoa(i), strconv.Itoa(i+1), "PUT") diff --git a/internals/sstable/writer.go b/internals/sstable/writer.go index 3f4a45b..cd6e26b 100644 --- a/internals/sstable/writer.go +++ b/internals/sstable/writer.go @@ -12,6 +12,7 @@ import ( "strings" "github.com/Vince-maple-byte/KeyData/internals/record" + "github.com/Vince-maple-byte/KeyData/internals/skiplist" "github.com/ccoveille/go-safecast/v2" ) @@ -248,7 +249,7 @@ func Compact(filePath string) error { bucketSize := len(val) if bucketSize >= minThreshold && bucketSize <= maxThreshold { - skiplist := MergeList() + skiplist := skiplist.CreateSkiplist() for _, fileInfo := range val { // #nosec G304 -- Reading SSTables discovered via os.ReadDir from the internal storage directory. fileData, err := os.ReadFile(filepath.Join(filePath, fileInfo.Name())) diff --git a/internals/sstable/writer_test.go b/internals/sstable/writer_test.go index fc16972..13ec5de 100644 --- a/internals/sstable/writer_test.go +++ b/internals/sstable/writer_test.go @@ -7,7 +7,6 @@ import ( "path/filepath" "testing" - "github.com/Vince-maple-byte/KeyData/internals/memtable" "github.com/Vince-maple-byte/KeyData/internals/record" "github.com/Vince-maple-byte/KeyData/internals/sstable" ) @@ -101,12 +100,6 @@ func TestBucketsForFiles(t *testing.T) { func TestCompactFiles(t *testing.T) { data := []string{"data", "data", "data", "data", "data", "data", "data", "data", "data", "data"} - //startUp(data...) - - sstable.MergeList = func() sstable.ListMerger { - return memtable.CreateSkiplist() - } - dir := t.TempDir() //size := 0 @@ -165,7 +158,7 @@ func TestWriteToFile(t *testing.T) { t.Errorf("Not able to create the file") } - file,err := os.Open(filepath.Join(dir, "kd_1.sst")) + file, err := os.Open(filepath.Join(dir, "kd_1.sst")) if err != nil { t.Fatalf("not able to open the file: %v", err) @@ -177,7 +170,6 @@ func TestWriteToFile(t *testing.T) { t.Error("Did not populate file") } - } func TestFooter(t *testing.T) { @@ -196,7 +188,7 @@ func TestFooter(t *testing.T) { } file, err := os.Open(filepath.Join(dir, "kd_1.sst")) - + if err != nil { t.Fatal(err.Error()) } From b40304d5f0ee00a9eb0da5d4670d2af5813c8161 Mon Sep 17 00:00:00 2001 From: Ivers Date: Thu, 30 Jul 2026 18:37:51 -0400 Subject: [PATCH 2/9] continuation of the last commit --- internals/bloom/bloom.go | 16 ++++++ internals/memtable/memtable.go | 32 ++++++----- internals/skiplist/node.go | 19 +++++++ internals/{memtable => skiplist}/skiplist.go | 18 +++--- .../{memtable => skiplist}/skiplist_test.go | 55 +++++++++++-------- main.go | 4 -- 6 files changed, 94 insertions(+), 50 deletions(-) create mode 100644 internals/bloom/bloom.go create mode 100644 internals/skiplist/node.go rename internals/{memtable => skiplist}/skiplist.go (95%) rename internals/{memtable => skiplist}/skiplist_test.go (83%) diff --git a/internals/bloom/bloom.go b/internals/bloom/bloom.go new file mode 100644 index 0000000..1df2839 --- /dev/null +++ b/internals/bloom/bloom.go @@ -0,0 +1,16 @@ +package bloom + +import "fmt" + +type Bloom struct { + Bits []byte + Hash int +} + +func NewBloom(numKeys int) (*Bloom, error) { + if numKeys <= 0 { + return nil, fmt.Errorf("Improrer size for the bloom") + } + + return &Bloom{}, nil +} diff --git a/internals/memtable/memtable.go b/internals/memtable/memtable.go index ffaebe8..6018eff 100644 --- a/internals/memtable/memtable.go +++ b/internals/memtable/memtable.go @@ -6,23 +6,25 @@ import ( "os" "fmt" + "github.com/Vince-maple-byte/KeyData/internals/record" + "github.com/Vince-maple-byte/KeyData/internals/skiplist" "github.com/Vince-maple-byte/KeyData/internals/sstable" ) const MAX_SIZE = 3200 type Memtable struct { - list *Skiplist - size int + list *skiplist.Skiplist + size int WalFilePath string DataDir string } func CreateMemtable(wal, dir string) *Memtable { return &Memtable{ - list: CreateSkiplist(), - size: 0, + list: skiplist.CreateSkiplist(), + size: 0, WalFilePath: wal, DataDir: dir, } @@ -52,7 +54,7 @@ func (m *Memtable) Write(key, value, operation string) (bool, error) { return false, errF } - //For now, when running test we just + //For now, when running test we just err = sstable.Compact(m.DataDir) if err != nil { @@ -61,9 +63,9 @@ func (m *Memtable) Write(key, value, operation string) (bool, error) { m.list.EmptyList() m.size = 0 - + if err := os.Remove(m.WalFilePath); err != nil { - return false, err; + return false, err } } @@ -89,10 +91,10 @@ func (m *Memtable) Get(key string) ([]byte, error) { func (m Memtable) writeToWal(record []byte) (bool, error) { file, err := os.OpenFile(m.WalFilePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0600) - + if err != nil { errMessage := fmt.Sprintf("Not able to create/open the wal file %s", m.WalFilePath) - return false, errors.New(errMessage); + return false, errors.New(errMessage) } defer file.Close() @@ -108,9 +110,9 @@ func (m Memtable) writeToWal(record []byte) (bool, error) { func (m *Memtable) MemtableStartUp() (bool, error) { file, err := os.OpenFile(m.WalFilePath, os.O_APPEND|os.O_CREATE|os.O_RDONLY, 0600) - + if err != nil { - return false, err; + return false, err } defer file.Close() @@ -126,7 +128,7 @@ func (m *Memtable) MemtableStartUp() (bool, error) { _, err := file.ReadAt(header, i) if err != nil { - return false, err; + return false, err } keySize := binary.BigEndian.Uint32(header[13:17]) @@ -136,7 +138,7 @@ func (m *Memtable) MemtableStartUp() (bool, error) { _, err = file.ReadAt(keyValuePair, i+21) if err != nil { - return false, err; + return false, err } ok := record.ChecksumChecker( @@ -150,8 +152,8 @@ func (m *Memtable) MemtableStartUp() (bool, error) { break } - rec := append(header, keyValuePair...) - m.list.Insert(string(rec[21: keySize + 21]), rec) + rec := append(header, keyValuePair...) + m.list.Insert(string(rec[21:keySize+21]), rec) m.size += 1 i += int64(len(rec)) diff --git a/internals/skiplist/node.go b/internals/skiplist/node.go new file mode 100644 index 0000000..81929e2 --- /dev/null +++ b/internals/skiplist/node.go @@ -0,0 +1,19 @@ +package skiplist + +type Node struct { + key string + value []byte + levels []*Node +} + +func (n *Node) GetKey() string { + return n.key +} + +func (n *Node) GetValue() []byte { + return n.value +} + +func (n *Node) GetLevels() []*Node { + return n.levels +} diff --git a/internals/memtable/skiplist.go b/internals/skiplist/skiplist.go similarity index 95% rename from internals/memtable/skiplist.go rename to internals/skiplist/skiplist.go index c70eb66..d726978 100644 --- a/internals/memtable/skiplist.go +++ b/internals/skiplist/skiplist.go @@ -1,4 +1,4 @@ -package memtable +package skiplist import ( "errors" @@ -9,15 +9,9 @@ import ( const MAX_LEVEL = 32 -type Node struct { - key string - value []byte - levels []*Node -} - type Skiplist struct { head *Node - size int + size uint64 } func CreateSkiplist() *Skiplist { @@ -32,6 +26,14 @@ func CreateSkiplist() *Skiplist { } } +func (list Skiplist) GetSize() uint64 { + return list.size +} + +func (list Skiplist) GetHead() *Node { + return list.head +} + func (list *Skiplist) Search(key string) ([]byte, error) { curr := list.head var next *Node diff --git a/internals/memtable/skiplist_test.go b/internals/skiplist/skiplist_test.go similarity index 83% rename from internals/memtable/skiplist_test.go rename to internals/skiplist/skiplist_test.go index b49d8f5..181928f 100644 --- a/internals/memtable/skiplist_test.go +++ b/internals/skiplist/skiplist_test.go @@ -1,17 +1,19 @@ -package memtable +package skiplist_test import ( "testing" "time" "github.com/Vince-maple-byte/KeyData/internals/record" + "github.com/Vince-maple-byte/KeyData/internals/skiplist" + "github.com/ccoveille/go-safecast/v2" ) // --- helpers --- -func newList(t *testing.T) *Skiplist { +func newList(t *testing.T) *skiplist.Skiplist { t.Helper() - return CreateSkiplist() + return skiplist.CreateSkiplist() } // --- CreateSkiplist --- @@ -25,15 +27,16 @@ func TestCreateSkiplist_NotNil(t *testing.T) { func TestCreateSkiplist_InitialSizeZero(t *testing.T) { list := newList(t) - if list.size != 0 { - t.Errorf("expected size 0, got %d", list.size) + if list.GetSize() != 0 { + t.Errorf("expected size 0, got %d", list.GetSize()) } } func TestCreateSkiplist_HeadHasMaxLevels(t *testing.T) { list := newList(t) - if len(list.head.levels) != MAX_LEVEL { - t.Errorf("expected head to have %d levels, got %d", MAX_LEVEL, len(list.head.levels)) + + if len(list.GetHead().GetLevels()) != skiplist.MAX_LEVEL { + t.Errorf("expected head to have %d levels, got %d", skiplist.MAX_LEVEL, len(list.GetHead().GetLevels())) } } @@ -42,8 +45,8 @@ func TestCreateSkiplist_HeadHasMaxLevels(t *testing.T) { func TestInsert_SingleElement(t *testing.T) { list := newList(t) list.Insert("a", []byte("val-a")) - if list.size != 1 { - t.Errorf("expected size 1, got %d", list.size) + if list.GetSize() != 1 { + t.Errorf("expected size 1, got %d", list.GetSize()) } } @@ -53,8 +56,14 @@ func TestInsert_MultipleElements(t *testing.T) { for _, k := range keys { list.Insert(k, []byte(k+"-value")) } - if list.size != len(keys) { - t.Errorf("expected size %d, got %d", len(keys), list.size) + keysSize, err := safecast.Convert[uint64](len(keys)) + + if err != nil { + t.Fatal("Not able to get the proper size of the keys") + } + + if list.GetSize() != keysSize { + t.Errorf("expected size %d, got %d", keysSize, list.GetSize()) } } @@ -67,8 +76,8 @@ func TestInsert_UpdateExistingKeyForDifferentTimesItWasInserted(t *testing.T) { list.Insert("key", new) // Size should not grow on update. - if list.size != 1 { - t.Errorf("expected size 1 after update, got %d", list.size) + if list.GetSize() != 1 { + t.Errorf("expected size 1 after update, got %d", list.GetSize()) } val, err := list.Search("key") @@ -90,8 +99,8 @@ func TestInsert_UpdateExistingKeyWhenInsertedAtTheSameTime(t *testing.T) { list.Insert("key", new) // Size should not grow on update. - if list.size != 1 { - t.Errorf("expected size 1 after update, got %d", list.size) + if list.GetSize() != 1 { + t.Errorf("expected size 1 after update, got %d", list.GetSize()) } val, err := list.Search("key") @@ -137,14 +146,14 @@ func TestInsert_PreservesOrder(t *testing.T) { list.Insert("b", []byte("b")) // Level-0 linked list must be sorted. - curr := list.head.levels[0] + curr := list.GetHead().GetLevels()[0] var prev string for curr != nil { - if curr.key < prev { - t.Errorf("out of order: %q after %q", curr.key, prev) + if curr.GetKey() < prev { + t.Errorf("out of order: %q after %q", curr.GetKey(), prev) } - prev = curr.key - curr = curr.levels[0] + prev = curr.GetKey() + curr = curr.GetLevels()[0] } } @@ -219,8 +228,8 @@ func TestDelete_DecreasesSize(t *testing.T) { list.Insert("x", []byte("x")) list.Insert("y", []byte("y")) list.Delete("x") - if list.size != 1 { - t.Errorf("expected size 1 after delete, got %d", list.size) + if list.GetSize() != 1 { + t.Errorf("expected size 1 after delete, got %d", list.GetSize()) } } @@ -294,7 +303,7 @@ func TestEmptyList_SizeAfterEmpty(t *testing.T) { // NOTE: EmptyList reassigns list internally; the caller's pointer is // unchanged. This test documents the current (possibly surprising) // behaviour — update if the implementation is fixed. - size := list.size // no panic expected + size := list.GetSize() // no panic expected if size != 0 { t.Error("Expected an empty list", size) } diff --git a/main.go b/main.go index 84fe7a4..7029f82 100644 --- a/main.go +++ b/main.go @@ -7,7 +7,6 @@ import ( "os" "github.com/Vince-maple-byte/KeyData/internals/memtable" - "github.com/Vince-maple-byte/KeyData/internals/sstable" "github.com/Vince-maple-byte/KeyData/network" "google.golang.org/grpc" "google.golang.org/grpc/reflection" @@ -42,9 +41,6 @@ TODO: Fix the filePath for the project so that it doesn't have to be hardcoded. func main() { //We need to use this interface here to avoid an import cycle between sstable and memtable - sstable.MergeList = func() sstable.ListMerger { - return memtable.CreateSkiplist() - } walPath := "./internals/wal/mem1.wal" dataDir := "./internals/data" From 1f2f129219104c9e433eaee16178093d07fcccfe Mon Sep 17 00:00:00 2001 From: Ivers Date: Sat, 8 Aug 2026 16:02:57 -0400 Subject: [PATCH 3/9] Almost finished with the code refactor. Just need to make some tests for the database struct --- internals/db/database.go | 160 +++++++++++++++++++++ internals/db/database_test.go | 1 + internals/memtable/memtable.go | 13 +- internals/sstable/compaction.go | 183 ++++++++++++++++++++++++ internals/sstable/corruption_test.go | 139 ++++++++++++++---- internals/sstable/reader.go | 174 ++++------------------- internals/sstable/reader_test.go | 72 ++++++++-- internals/sstable/sstfile.go | 133 +++++++++++++++++ internals/sstable/writer.go | 205 ++++----------------------- internals/sstable/writer_test.go | 73 +++++----- 10 files changed, 746 insertions(+), 407 deletions(-) create mode 100644 internals/db/database.go create mode 100644 internals/db/database_test.go create mode 100644 internals/sstable/compaction.go create mode 100644 internals/sstable/sstfile.go diff --git a/internals/db/database.go b/internals/db/database.go new file mode 100644 index 0000000..6c05be4 --- /dev/null +++ b/internals/db/database.go @@ -0,0 +1,160 @@ +package database + +import ( + "errors" + "fmt" + "os" + "path/filepath" + + "github.com/Vince-maple-byte/KeyData/internals/memtable" + "github.com/Vince-maple-byte/KeyData/internals/sstable" + "github.com/ccoveille/go-safecast/v2" +) + +type Database struct { + Memtable *memtable.Memtable + SSTFiles []*sstable.SSTFile + Dir string + WalPath string +} + +func CreateDatabase(dir, walPath string) (*Database, error) { + d := &Database{} + d.Dir = dir + d.WalPath = walPath + + err := d.Open() + + if err != nil { + return nil, err + } + + d.CreateMemtable() + + return d, nil +} + +func (db *Database) Open() error { + db.SSTFiles = make([]*sstable.SSTFile, 0) + directory, err := os.ReadDir(db.Dir) + + if err != nil { + return err + } + + for _, d := range directory { + sstFile := &sstable.SSTFile{} + fileName := filepath.Join(db.Dir, d.Name()) + fileInfo, err := d.Info() + + if err != nil { + return err + } + + err = sstFile.Open(fileName) + + if err != nil { + return err + } + + sstFile.Size = fileInfo.Size() + + err = sstFile.PopulateFooter() + + if err != nil { + return err + } + + indexOffsetI64, err := safecast.Convert[int64](sstFile.Footer.IndexOffset) + + if err != nil { + return err + } + + indexBlock := make([]byte, (sstFile.Size-24)-indexOffsetI64) + _, err = sstFile.File.ReadAt(indexBlock, indexOffsetI64) + + if err != nil { + return err + } + + sstFile.PopulateIndex(indexBlock) + } + + return nil +} + +func (db *Database) Close() error { + for _, file := range db.SSTFiles { + if err := file.Close(); err != nil { + return err + } + } + + return nil +} + +func (db *Database) CreateMemtable() { + db.Memtable = memtable.CreateMemtable(db.Dir, db.WalPath) +} + +func (db *Database) Get(key string) ([]byte, error) { + //We would check the memtable first + res, err := db.Memtable.Get(key) + + if err == nil { + return res, nil + } + // Then all of the SSTFiles, + // once we make the bloom filters, this will be faster since + // we will get a good answer as to whether the key is in the file + res, err = sstable.ReadFromAllFiles(key, db.SSTFiles) + + if err == nil { + return res, nil + } + + return nil, fmt.Errorf("Not able to find key/value pair for %s", key) +} + +func (db *Database) Put(key, val string) (bool, error) { + ok, err := db.Memtable.Write(key, val, "PUT") + + if !ok { + return ok, err + } + + if err == nil { + return true, err + } + + compactErr := fmt.Errorf("Need to compact the files") + + if errors.Is(err, compactErr) { + sstable.Compact(db.SSTFiles, db.Dir) + return true, nil + } + + return ok, err +} + +func (db *Database) Delete(key string) (bool, error) { + ok, err := db.Memtable.Write(key, "", "DELETE") + + if !ok { + return ok, err + } + + if err == nil { + return true, err + } + + compactErr := fmt.Errorf("Need to compact the files") + + if errors.Is(err, compactErr) { + sstable.Compact(db.SSTFiles, db.Dir) + return true, nil + } + + return ok, err +} diff --git a/internals/db/database_test.go b/internals/db/database_test.go new file mode 100644 index 0000000..0385802 --- /dev/null +++ b/internals/db/database_test.go @@ -0,0 +1 @@ +package database_test diff --git a/internals/memtable/memtable.go b/internals/memtable/memtable.go index 6018eff..76d0673 100644 --- a/internals/memtable/memtable.go +++ b/internals/memtable/memtable.go @@ -48,19 +48,15 @@ func (m *Memtable) Write(key, value, operation string) (bool, error) { if m.size >= MAX_SIZE { content := m.list.EntireList() - _, errF := sstable.WriteToFile(content, m.DataDir) - - if errF != nil { - return false, errF - } - - //For now, when running test we just - err = sstable.Compact(m.DataDir) + _, err := sstable.WriteToFile(content, m.DataDir) if err != nil { return false, err } + //For now, when running test we just + //err = sstable.Compact(m.DataDir) + m.list.EmptyList() m.size = 0 @@ -68,6 +64,7 @@ func (m *Memtable) Write(key, value, operation string) (bool, error) { return false, err } + return true, fmt.Errorf("Need to compact the files") } return true, nil diff --git a/internals/sstable/compaction.go b/internals/sstable/compaction.go new file mode 100644 index 0000000..36adf6c --- /dev/null +++ b/internals/sstable/compaction.go @@ -0,0 +1,183 @@ +package sstable + +import ( + "encoding/binary" + "fmt" + "os" + + "github.com/Vince-maple-byte/KeyData/internals/record" + "github.com/Vince-maple-byte/KeyData/internals/skiplist" + "github.com/ccoveille/go-safecast/v2" +) + +//Change these methods to be SSTFile complient + +// We are going to be doing size based compaction for compacting these files +// The amount of files that need to be a similar size +// TODO: Need to make the bucket map into a persistent map that is used throughout the entire state of the +// program.w +func buckets(files []*SSTFile) (map[file_buckets][]*SSTFile, error) { + + var average_size int64 + var total_size int64 + + buckets := make(map[file_buckets][]*SSTFile) + + buckets[SMALL] = make([]*SSTFile, 0) + buckets[MEDIUM] = make([]*SSTFile, 0) + buckets[LARGE] = make([]*SSTFile, 0) + buckets[OVERSIZE] = make([]*SSTFile, 0) + + for _, file := range files { + + total_size += file.Size + } + + average_size = total_size / int64(len(files)) + + for _, file := range files { + size := file.Size + + switch { + case float64(size) >= float64(average_size)*float64(OVERSIZE): + buckets[OVERSIZE] = append(buckets[OVERSIZE], file) + case float64(size) >= float64(average_size)*float64(LARGE): + buckets[LARGE] = append(buckets[LARGE], file) + case float64(size) >= float64(average_size)*float64(MEDIUM): + buckets[MEDIUM] = append(buckets[MEDIUM], file) + default: + buckets[SMALL] = append(buckets[SMALL], file) + } + } + + return buckets, nil +} + +// When we are doing the concurrency option. Compaction and the buckets will be done in a separate class, +// and in it's own separate thread that will be run periodically every time interval that we decide +func Compact(files []*SSTFile, dir string) ([]*SSTFile, error) { + bucketMap, err := buckets(files) + + if err != nil { + return files, err + } + + minThreshold := 4 + maxThreshold := 32 + + //Slight potential optimization: We can have this continue compacting each bucket of similar sizes, + // but we have to recalculate the average file size and bucket arrangement for Each of the files + for _, val := range bucketMap { + bucketSize := len(val) + + if bucketSize >= minThreshold && bucketSize <= maxThreshold { + skiplist := skiplist.CreateSkiplist() + for _, file := range val { + // #nosec G304 -- Reading SSTables discovered via os.ReadDir from the internal storage directory. + footer := make([]byte, 24) + _, err := file.File.ReadAt(footer, file.Size-24) + + if err != nil { + return files, err + } + //We do this so that we only take into account the file block, and not the index or footer + if binary.BigEndian.Uint64(footer[16:]) != 0xDEADBEEFDEADBEEF { + file.Close() + if err := os.Remove(file.FileName); err != nil { + return files, fmt.Errorf("warning: failed to remove corrupt SSTable %s: %v", + file.FileName, err) + } + return files, fmt.Errorf("invalid footer magic") + } + fileBlockEnds := binary.BigEndian.Uint64(footer[8:16]) + + //We are going through each file and from there we will save each key/value pair record into a skiplist + fileData := make([]byte, file.Footer.IndexOffset) + _, err = file.File.ReadAt(fileData, 0) + + if err != nil { + return files, err + } + + for i := uint64(0); i < fileBlockEnds; { + time := fileData[i : i+8] + + checksum := binary.BigEndian.Uint32(fileData[i+8 : i+12]) + + keySize := int(binary.BigEndian.Uint32(fileData[i+13 : i+17])) + payloadSize := int(binary.BigEndian.Uint32(fileData[i+17 : i+21])) + keySizeUi64, err := safecast.Convert[uint64](keySize) + + if err != nil { + return files, err + } + + payloadSizeUi64, err := safecast.Convert[uint64](payloadSize) + + if err != nil { + return files, err + } + + key := fileData[i+21 : i+keySizeUi64+21] + payload := fileData[i+keySizeUi64+21 : i+(keySizeUi64+21)+payloadSizeUi64] + + //If the checksum is invalid, we ignore the rest of the file + if !record.ChecksumChecker(key, payload, binary.BigEndian.Uint64(time), checksum) { + break + } + + skiplist.Insert(string(key), fileData[i:i+(keySizeUi64+21)+payloadSizeUi64]) + i = i + (keySizeUi64 + 21) + payloadSizeUi64 + + } + } + + //We take the entire skiplist and write it into a new file + fileContents := skiplist.EntireList() + + //Technically speaking we can just recall the write to file again to make the new file + // Since the Entire write operation is there. + // So I'm planning on calling the Compact function after the write to file function goes through in the Memtable class + newSSTable, err := WriteToFile(fileContents, dir) + + if err != nil { + return files, err + } + + newSlice := make([]*SSTFile, 0, len(files)-len(val)) + // Delete all of the old files once the new file is committed + for _, oldFile := range val { + err := oldFile.Delete() + if err != nil { + return files, err + } + } + + toDelete := make(map[*SSTFile]struct{}) + + for _, sst := range val { + toDelete[sst] = struct{}{} + } + + for _, sst := range files { + if _, exists := toDelete[sst]; exists { + continue + } + + newSlice = append(newSlice, sst) + } + + newSlice = append(newSlice, newSSTable) + return newSlice, nil + } + + } + + return files, nil +} + +func ExportBuckets(files []*SSTFile) (map[file_buckets][]*SSTFile, error) { + result, err := buckets(files) + + return result, err +} diff --git a/internals/sstable/corruption_test.go b/internals/sstable/corruption_test.go index e3bc574..02259e5 100644 --- a/internals/sstable/corruption_test.go +++ b/internals/sstable/corruption_test.go @@ -4,6 +4,7 @@ import ( "encoding/binary" "os" "path/filepath" + "strconv" "testing" "github.com/Vince-maple-byte/KeyData/internals/sstable" @@ -377,51 +378,129 @@ func TestFileOffset_EmptyList(t *testing.T) { func TestCompact_IgnoresFileWithTruncatedRecord(t *testing.T) { dir := t.TempDir() - // Write 4 files so Compact's minThreshold (4) is met. - // Three are valid; one has its record body truncated. - writeValidSST := func(name, key, val string) { - rec := buildValidRecord(key, val, 1) - footer := buildValidFooter(uint64(len(rec))) - data := buildSSTFile([][]byte{rec}, []byte{}, footer) - os.WriteFile(filepath.Join(dir, name), data, 0644) - } + sstFiles := make([]*sstable.SSTFile, 0, 4) + + // Create three valid SSTables using the real writer. + keys := []string{"apple", "banana", "cherry"} + + for _, key := range keys { + rec := buildValidRecord(key, "value", 1) - writeValidSST("kd_1.sst", "apple", "pie") - writeValidSST("kd_2.sst", "banana", "split") - writeValidSST("kd_3.sst", "cherry", "jam") + sst, err := sstable.WriteToFile([][]byte{rec}, dir) + if err != nil { + t.Fatal(err) + } + + sstFiles = append(sstFiles, sst) + } - // Fourth file: valid header, truncated body. + // Create one corrupted SSTable manually. rec := buildValidRecord("date", "fruit", 1) truncated := rec[:len(rec)-3] + footer := buildValidFooter(uint64(len(truncated))) - corrupt := buildSSTFile([][]byte{truncated}, []byte{}, footer) - os.WriteFile(filepath.Join(dir, "kd_4.sst"), corrupt, 0644) + data := buildSSTFile([][]byte{truncated}, nil, footer) + + corruptPath := filepath.Join(dir, "kd_999.sst") + if err := os.WriteFile(corruptPath, data, 0644); err != nil { + t.Fatal(err) + } + + corrupt := &sstable.SSTFile{} + if err := corrupt.Open(corruptPath); err != nil { + t.Fatal(err) + } + + info, err := corrupt.File.Stat() + if err != nil { + t.Fatal(err) + } + + corrupt.Size = info.Size() + + if err = corrupt.PopulateFooter(); err != nil { + t.Fatal(err) + } + + // The corrupted file has no valid index block. + corrupt.Index = nil + + sstFiles = append(sstFiles, corrupt) + + defer func() { + for _, s := range sstFiles { + s.Close() + } + }() + + _, err = sstable.Compact(sstFiles, dir) - // Compact should not panic; corruption causes the inner loop to break early. - err := sstable.Compact(dir) - // We accept either nil or an error — the key requirement is no panic / index OOB. + // The important property is that Compact doesn't panic. t.Logf("Compact returned: %v", err) } func TestCompact_FileWithSwappedMagic(t *testing.T) { dir := t.TempDir() - writeSST := func(name, key, val string, corruptMagic bool) { - rec := buildValidRecord(key, val, 1) - footer := buildValidFooter(uint64(len(rec))) - if corruptMagic { - // Swap the last two bytes of the magic number. - footer[22], footer[23] = footer[23], footer[22] + sstFiles := make([]*sstable.SSTFile, 0, 4) + + // Create three valid SSTables. + keys := []string{"alpha", "beta", "gamma"} + + for i, key := range keys { + rec := buildValidRecord(key, strconv.Itoa(i+1), 1) + + sst, err := sstable.WriteToFile([][]byte{rec}, dir) + if err != nil { + t.Fatal(err) } - data := buildSSTFile([][]byte{rec}, []byte{}, footer) - os.WriteFile(filepath.Join(dir, name), data, 0644) + + sstFiles = append(sstFiles, sst) + } + + // Create one SSTable with a corrupted footer magic. + rec := buildValidRecord("delta", "4", 1) + footer := buildValidFooter(uint64(len(rec))) + + // Corrupt the magic number. + footer[22], footer[23] = footer[23], footer[22] + + data := buildSSTFile([][]byte{rec}, nil, footer) + + corruptPath := filepath.Join(dir, "kd_999.sst") + if err := os.WriteFile(corruptPath, data, 0644); err != nil { + t.Fatal(err) + } + + corrupt := &sstable.SSTFile{} + if err := corrupt.Open(corruptPath); err != nil { + t.Fatal(err) + } + + info, err := corrupt.File.Stat() + if err != nil { + t.Fatal(err) } - writeSST("kd_1.sst", "alpha", "1", false) - writeSST("kd_2.sst", "beta", "2", false) - writeSST("kd_3.sst", "gamma", "3", false) - writeSST("kd_4.sst", "delta", "4", true) // corrupted magic + corrupt.Size = info.Size() + + if err := corrupt.PopulateFooter(); err != nil { + t.Fatal(err) + } + + // No valid index block. + corrupt.Index = nil + + sstFiles = append(sstFiles, corrupt) + + defer func() { + for _, s := range sstFiles { + s.Close() + } + }() + + _, err = sstable.Compact(sstFiles, dir) - err := sstable.Compact(dir) + // We only care that Compact detects the corruption and doesn't panic. t.Logf("Compact with swapped magic returned: %v", err) } diff --git a/internals/sstable/reader.go b/internals/sstable/reader.go index 4043e39..406280b 100644 --- a/internals/sstable/reader.go +++ b/internals/sstable/reader.go @@ -3,22 +3,11 @@ package sstable import ( "encoding/binary" "errors" - "fmt" - "os" - "path/filepath" - "sort" - "strconv" - "strings" "github.com/Vince-maple-byte/KeyData/internals/record" "github.com/ccoveille/go-safecast/v2" ) -type SSTFile struct { - Generation int - FileName string -} - // How to read from the file. // We first checking the magic number inside of the footer to see if the file is valid // We get the byte offset of the index block, and proceed to do: @@ -28,20 +17,11 @@ type SSTFile struct { // if it is equal than we can just return that key/value pair inside of the file at that byte offset // If the key is not inside of the range in which we stated before, than we can just return nil and an error message // stating that the key can't be found -func ReadFromFile(filePath, key string) ([]byte, error) { - // #nosec G304 - file, err := os.Open(filePath) +func ReadFromFile(key string, file *SSTFile) ([]byte, error) { - if err != nil { - return nil, err - } - - defer file.Close() - - fileInfo, _ := file.Stat() footer := make([]byte, 24) - _, err = file.ReadAt(footer, fileInfo.Size()-24) + _, err := file.File.ReadAt(footer, file.Size-24) if err != nil { return nil, err @@ -53,102 +33,28 @@ func ReadFromFile(filePath, key string) ([]byte, error) { } //Getting the index block offset - indexBlockLoc := binary.BigEndian.Uint64(footer[8:16]) - lowKeyOffset := uint64(0) - highKeyOffset := uint64(0) - fileSize64, err := safecast.Convert[uint64](fileInfo.Size()) - - if err != nil { - return nil, err - } - - for i := indexBlockLoc; i <= fileSize64-24; { - i64, err := safecast.Convert[int64](i) - if err != nil { - return nil, err - } - keySize := make([]byte, 4) - _, err = file.ReadAt(keySize, i64) - keySize32 := binary.BigEndian.Uint32(keySize) - - if err != nil { - return nil, err - } - - offsetKey := make([]byte, keySize32) - - _, err = file.ReadAt(offsetKey, i64+4) - if err != nil { - return nil, err - } - - //This gives us the location of the offset where it is saved in the data portion of the file - offsetLoc := make([]byte, 8) - keySize64, err := safecast.Convert[int64](keySize32) - - if err != nil { - return nil, err - } - - _, err = file.ReadAt(offsetLoc, i64+4+keySize64) - - if err != nil { - return nil, err - } + startOffset := uint64(0) + endOffset := uint64(0) - keyOffset := binary.BigEndian.Uint64(offsetLoc) + for i := 0; i < len(file.Index); i++ { - //Go to the next location in the index block - i = i + 4 + uint64(binary.BigEndian.Uint32(keySize)) + 8 + indexKey := file.Index[i].Key + keyOffset := file.Index[i].Offset - if key == string(offsetKey) { - curr := make([]byte, 21) - keyOffsetI64, err := safecast.Convert[int64](keyOffset) - - if err != nil { - return nil, err - } - - _, err = file.ReadAt(curr, keyOffsetI64) - - if err != nil { - return nil, err - } - - payloadSize := binary.BigEndian.Uint32(curr[17:21]) - - entireRecord := make([]byte, 21+binary.BigEndian.Uint32(keySize)+payloadSize) - keyOffsetConv, err := safecast.Convert[int64](keyOffset) - - if err != nil { - return nil, err - } - - _, err = file.ReadAt(entireRecord, keyOffsetConv) - - if err != nil { - return nil, err - } - - return entireRecord, nil + if i < len(file.Index)-1 { + endOffset = file.Index[i+1].Offset + } else { + endOffset = file.Footer.IndexOffset } - lowKeyOffset = highKeyOffset - highKeyOffset = keyOffset - - if key < string(offsetKey) { + if key <= string(indexKey) { break } + startOffset = keyOffset } - if lowKeyOffset == highKeyOffset { - return nil, errors.New("key is not inside of the file: Key is larger than any key in the file") - } - - //TODO: Make the range to - - for i := lowKeyOffset; i < highKeyOffset; { + for i := startOffset; i <= endOffset; { curr := make([]byte, 21) convertI, err := safecast.Convert[int64](i) @@ -156,7 +62,7 @@ func ReadFromFile(filePath, key string) ([]byte, error) { return nil, err } - _, err = file.ReadAt(curr, convertI) + _, err = file.File.ReadAt(curr, convertI) if err != nil { return nil, err @@ -167,7 +73,7 @@ func ReadFromFile(filePath, key string) ([]byte, error) { entireRecord := make([]byte, 21+keySize+payloadSize) - _, err = file.ReadAt(entireRecord, convertI) + _, err = file.File.ReadAt(entireRecord, convertI) if err != nil { return nil, err @@ -192,9 +98,9 @@ func ReadFromFile(filePath, key string) ([]byte, error) { return entireRecord, nil } - if currRecord.Key > key { - break - } + // if currRecord.Key > key { + // break + // } i += 21 + uint64(keySize) + uint64(payloadSize) } @@ -202,37 +108,19 @@ func ReadFromFile(filePath, key string) ([]byte, error) { return nil, errors.New("key could not be found") } -func ReadFromAllFiles(key string, dir string) ([]byte, error) { - fileDir, err := os.ReadDir(dir) +// We have to fix the ReadFromAllFiles to use the []SSTFile struct instead dir string +func ReadFromAllFiles(key string, files []*SSTFile) ([]byte, error) { - var files []SSTFile - - if err != nil { - return nil, err - } - - for _, file := range fileDir { - gen, err := parseGeneration(filepath.Join(dir, file.Name())) - - if err != nil { - continue - } - - files = append(files, SSTFile{ - Generation: gen, - FileName: file.Name(), - }) - } - - sort.Slice(files, func(i, j int) bool { - return files[i].Generation > files[j].Generation - }) + //By default we are going to assume that the SSTFiles are going to be sorted since the new files + // always get written in the back + // sort.Slice(files, func(i, j int) bool { + // return files[i].Generation > files[j].Generation + // }) var currContent record.Content var result []byte for _, file := range files { - fmt.Println(file.FileName) - res, err := ReadFromFile(filepath.Join(dir, file.FileName), key) + res, err := ReadFromFile(key, file) if err == nil { contents := record.GetContents(res) @@ -256,11 +144,3 @@ func ReadFromAllFiles(key string, dir string) ([]byte, error) { return nil, errors.New("the key does not exist in any of the files") } } - -func parseGeneration(fileName string) (int, error) { - str := strings.Split(fileName, "_")[1] - str = strings.Split(str, ".")[0] - idx, err := strconv.Atoi(str) - - return idx, err -} diff --git a/internals/sstable/reader_test.go b/internals/sstable/reader_test.go index bfef5f1..5c144d8 100644 --- a/internals/sstable/reader_test.go +++ b/internals/sstable/reader_test.go @@ -90,11 +90,10 @@ func TestReadFile(t *testing.T) { expected: "1141", }, { - testName: "key_if_file_truncated", + testName: "key_if_file_smaller_than_normal", data: func() [][]byte { d := skiplist.CreateSkiplist() for i := range 240 { - r, _ := record.CreateRecord(strconv.Itoa(i), strconv.Itoa(i+1), "PUT") d.Insert(strconv.Itoa(i), r) } @@ -104,7 +103,7 @@ func TestReadFile(t *testing.T) { expected: "105", }, { - testName: "key_if_larger_than_normal", + testName: "key_if_file_larger_than_normal", data: func() [][]byte { d := skiplist.CreateSkiplist() for i := range 7000 { @@ -122,18 +121,20 @@ func TestReadFile(t *testing.T) { for _, test := range tests { t.Run(test.testName, func(t *testing.T) { dir := t.TempDir() - _, err := sstable.WriteToFile(test.data, dir) + file, err := sstable.WriteToFile(test.data, dir) if err != nil { t.Fatalf("Error in writing the file: %s", err.Error()) } + + defer file.Close() + fileInfo, err := os.Lstat(filepath.Join(dir, "kd_1.sst")) size := fileInfo.Size() if err != nil { - tearDown("../test") t.Fatalf("Error in accessing the file stats: %s", err.Error()) } - rec, err := sstable.ReadFromFile(filepath.Join(dir, "kd_1.sst"), test.key) + rec, err := sstable.ReadFromFile(test.key, file) if rec == nil || err != nil { t.Fatalf("Error received: %v\nSize of the file: %d", err.Error(), size) @@ -149,10 +150,62 @@ func TestReadFile(t *testing.T) { } } +func TestReadFromFileIfKeyIsNotValid(t *testing.T) { + tests := []struct { + testName string + data [][]byte + key string + expected string + }{ + { + testName: "key_size_4_not", + data: func() [][]byte { + d := skiplist.CreateSkiplist() + for i := range 3200 { + + r, _ := record.CreateRecord(strconv.Itoa(i), strconv.Itoa(i+1), "PUT") + d.Insert(strconv.Itoa(i), r) + } + return d.EntireList() + }(), + key: "3888888", + expected: "3888889", + }, + } + + for _, test := range tests { + t.Run(test.testName, func(t *testing.T) { + dir := t.TempDir() + file, err := sstable.WriteToFile(test.data, dir) + + if err != nil { + t.Fatalf("Error in writing the file: %s", err.Error()) + } + + defer file.Close() + + fileInfo, err := os.Lstat(filepath.Join(dir, "kd_1.sst")) + size := fileInfo.Size() + if err != nil { + t.Fatalf("Error in accessing the file stats: %s", err.Error()) + } + _, err = sstable.ReadFromFile(test.key, file) + + if err == nil { + t.Fatalf("Expecting an error\nSize of the file: %d", size) + } + + fmt.Println(err.Error()) + //tearDown("../test") + }) + } +} + func TestReadFromAllFiles(t *testing.T) { dir := t.TempDir() fmt.Println("File Directory:", dir) + files := make([]*sstable.SSTFile, 0, 10) for i := range 10 { d := skiplist.CreateSkiplist() for i := range 3000 * (i + 1) { @@ -160,14 +213,17 @@ func TestReadFromAllFiles(t *testing.T) { r, _ := record.CreateRecord(strconv.Itoa(i), strconv.Itoa(i+1), "PUT") d.Insert(strconv.Itoa(i), r) } - _, err := sstable.WriteToFile(d.EntireList(), dir) + file, err := sstable.WriteToFile(d.EntireList(), dir) + defer file.Close() if err != nil { t.Fatalf("Error in writing the file: %s", err.Error()) } + + files = append(files, file) } - _, err := sstable.ReadFromAllFiles("29994", dir) + _, err := sstable.ReadFromAllFiles("29994", files) if err != nil { t.Errorf("Was not able to find the valid record\n%v", err) diff --git a/internals/sstable/sstfile.go b/internals/sstable/sstfile.go new file mode 100644 index 0000000..d5cf121 --- /dev/null +++ b/internals/sstable/sstfile.go @@ -0,0 +1,133 @@ +package sstable + +import ( + "encoding/binary" + "fmt" + "os" + "strconv" + "strings" + + "github.com/ccoveille/go-safecast/v2" +) + +type Footer struct { + Magic uint64 + IndexOffset uint64 + //BloomOffset uint64 +} + +type IndexBlock struct { + Key string + Offset uint64 +} + +type SSTFile struct { + File *os.File + Generation int + FileName string + Size int64 + Index []*IndexBlock + Footer *Footer + + //bloom filter goes here +} + +func (s *SSTFile) Open(filepath string) error { + if s.File != nil { + return fmt.Errorf("File is already open") + } + + file, err := os.OpenFile(filepath, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0666) + + if err != nil { + return err + } + + s.File = file + s.FileName = filepath + s.Generation, err = s.ParseGeneration() + + if err != nil { + return err + } + + return nil +} + +func (s *SSTFile) Close() error { + if s.File == nil { + return nil + } + + err := s.File.Close() + s.File = nil + + return err +} + +func (s *SSTFile) PopulateIndex(blocks []byte) { + indexBlocks := make([]*IndexBlock, 0) + + for i := 0; i < len(blocks); { + keySize := binary.BigEndian.Uint32(blocks[i : i+4]) + + keySizeInt, err := safecast.Convert[int](keySize) + if err != nil { + break + } + + if i+12+keySizeInt > len(blocks) { + break + } + + indexKey := string(blocks[i+4 : uint32(i+4)+keySize]) + + keyOffset := binary.BigEndian.Uint64(blocks[uint32(i+4)+keySize : uint32(i+12)+keySize]) + //fmt.Println(keyOffset) + index := &IndexBlock{ + Key: indexKey, + Offset: keyOffset, + } + + i += keySizeInt + 12 + indexBlocks = append(indexBlocks, index) + } + + s.Index = indexBlocks +} + +func (s *SSTFile) PopulateFooter() error { + footer := make([]byte, 24) + _, err := s.File.ReadAt(footer, s.Size-24) + + if err != nil { + return err + } + + s.Footer = &Footer{} + + s.Footer.Magic = binary.BigEndian.Uint64(footer[16:]) + s.Footer.IndexOffset = binary.BigEndian.Uint64(footer[8:16]) + + return nil +} + +func (s *SSTFile) Delete() error { + err := s.File.Close() + + if err != nil { + return err + } + + err = os.Remove(s.FileName) + + return err +} + +func (s *SSTFile) ParseGeneration() (int, error) { + str := strings.Split(s.FileName, "_")[1] + str = strings.Split(str, ".")[0] + idx, err := strconv.Atoi(str) + //s.Generation = idx + return idx, err +} diff --git a/internals/sstable/writer.go b/internals/sstable/writer.go index cd6e26b..1970bc8 100644 --- a/internals/sstable/writer.go +++ b/internals/sstable/writer.go @@ -3,8 +3,6 @@ package sstable import ( "encoding/binary" "errors" - "fmt" - "io/fs" "os" "path/filepath" "slices" @@ -12,7 +10,6 @@ import ( "strings" "github.com/Vince-maple-byte/KeyData/internals/record" - "github.com/Vince-maple-byte/KeyData/internals/skiplist" "github.com/ccoveille/go-safecast/v2" ) @@ -27,19 +24,19 @@ const ( OVERSIZE file_buckets = 2.0 ) -// TODO:Finish with the write method for the file i/o; use the diagram that I made as a guide. -func WriteToFile(list [][]byte, filePath string) (bool, error) { +// TODO: Change WriteToFile to be SSTFile struct complient +func WriteToFile(list [][]byte, filePath string) (*SSTFile, error) { filename := "" files, err := os.ReadDir(filePath) if err != nil { - return false, err + return nil, err } + maxIndex := 0 if len(files) < 1 { filename = "kd_1.sst" } else { - maxIndex := 0 for _, f := range files { str := strings.Split(f.Name(), "_")[1] str = strings.Split(str, ".")[0] @@ -51,21 +48,21 @@ func WriteToFile(list [][]byte, filePath string) (bool, error) { filename = "kd_" + strconv.Itoa(maxIndex+1) + ".sst" } + sstFile := &SSTFile{} + sstFile.Generation = maxIndex + 1 // #nosec G304 -- Creating SSTables enumerated from the internal storage directory. - file, err := os.Create(filepath.Join(filePath, filename)) + err = sstFile.Open(filepath.Join(filePath, filename)) if err != nil { - return false, err + return nil, err } - defer file.Close() - offset := fileOffset(list) index := createIndexBlock(list, offset) footer, err := createFooter(list) if err != nil { - return false, err + return nil, err } list = append(list, index) @@ -73,17 +70,28 @@ func WriteToFile(list [][]byte, filePath string) (bool, error) { content := slices.Concat(list...) - _, err = file.Write(content) + sstFile.PopulateIndex(index) + sstFile.Size, err = safecast.Convert[int64](len(content)) + if err != nil { + return nil, err + } + indexOffset, err := safecast.Convert[uint64](sstFile.Size - int64(len(index)+len(footer))) + sstFile.Footer = &Footer{ + Magic: uint64(0xDEADBEEFDEADBEEF), + IndexOffset: indexOffset, + } + + _, err = sstFile.File.Write(content) if err != nil { - return false, err + return nil, err } - if err := file.Sync(); err != nil { - return false, err + if err := sstFile.File.Sync(); err != nil { + return nil, err } - return true, nil + return sstFile, nil } //Create the indexing block for the file @@ -121,13 +129,13 @@ func createIndexBlock(contentList [][]byte, offset []uint64) []byte { //I made this specific change for the index block since in the if len(contentList) < INDEX_BLOCK*160 { - contents := record.GetContents(contentList[len(contentList)/2]) + contents := record.GetContents(contentList[0]) index = binary.BigEndian.AppendUint32(index, contents.Keysize) index = append(index, contents.Key...) - index = binary.BigEndian.AppendUint64(index, offset[len(offset)/2]) + index = binary.BigEndian.AppendUint64(index, offset[0]) } else { for i := 0; i < len(contentList); i = i + (len(contentList) / INDEX_BLOCK) { contents := record.GetContents(contentList[i]) @@ -178,165 +186,6 @@ func createFooter(list [][]byte) ([]byte, error) { return footer, nil } -// We are going to be doing size based compaction for compacting these files -// The amount of files that need to be a similar size -// TODO: Need to make the bucket map into a persistent map that is used throughout the entire state of the -// program.w -func buckets(filePath string) (map[file_buckets][]fs.FileInfo, error) { - files, err := os.ReadDir(filePath) - - if err != nil { - return nil, err - } - var average_size int64 - var total_size int64 - - buckets := make(map[file_buckets][]fs.FileInfo) - // - // - - buckets[SMALL] = make([]fs.FileInfo, 0) - buckets[MEDIUM] = make([]fs.FileInfo, 0) - buckets[LARGE] = make([]fs.FileInfo, 0) - buckets[OVERSIZE] = make([]fs.FileInfo, 0) - - for _, file := range files { - - fileInfo, err := file.Info() - - if err != nil { - continue - } - - total_size += fileInfo.Size() - } - - average_size = total_size / int64(len(files)) - - for _, file := range files { - fileInfo, _ := file.Info() - size := fileInfo.Size() - - switch { - case float64(size) >= float64(average_size)*float64(OVERSIZE): - buckets[OVERSIZE] = append(buckets[OVERSIZE], fileInfo) - case float64(size) >= float64(average_size)*float64(LARGE): - buckets[LARGE] = append(buckets[LARGE], fileInfo) - case float64(size) >= float64(average_size)*float64(MEDIUM): - buckets[MEDIUM] = append(buckets[MEDIUM], fileInfo) - default: - buckets[SMALL] = append(buckets[SMALL], fileInfo) - } - } - - return buckets, nil -} - -// When we are doing the concurrency option. Compaction and the buckets will be done in a separate class, -// and in it's own separate thread that will be run periodically every time interval that we decide -func Compact(filePath string) error { - bucketMap, err := buckets(filePath) - - if err != nil { - return err - } - minThreshold := 4 - maxThreshold := 32 - - //Slight potential optimization: We can have this continue compacting each bucket of similar sizes, - // but we have to recalculate the average file size and bucket arrangement for Each of the files - for _, val := range bucketMap { - bucketSize := len(val) - - if bucketSize >= minThreshold && bucketSize <= maxThreshold { - skiplist := skiplist.CreateSkiplist() - for _, fileInfo := range val { - // #nosec G304 -- Reading SSTables discovered via os.ReadDir from the internal storage directory. - fileData, err := os.ReadFile(filepath.Join(filePath, fileInfo.Name())) - - if err != nil { - return err - } - //We do this so that we only take into account the file block, and not the index or footer - footer := fileData[len(fileData)-24:] - if binary.BigEndian.Uint64(footer[16:]) != 0xDEADBEEFDEADBEEF { - - if err := os.Remove(filepath.Join(filePath, fileInfo.Name())); err != nil { - return fmt.Errorf("warning: failed to remove corrupt SSTable %s: %v", - fileInfo.Name(), err) - } - return fmt.Errorf("invalid footer magic") - } - fileBlockEnds := binary.BigEndian.Uint64(footer[8:16]) - - //We are going through each file and from there we will save each key/value pair record into a skiplist - for i := uint64(0); i < fileBlockEnds; { - time := fileData[i : i+8] - - checksum := binary.BigEndian.Uint32(fileData[i+8 : i+12]) - - keySize := int(binary.BigEndian.Uint32(fileData[i+13 : i+17])) - payloadSize := int(binary.BigEndian.Uint32(fileData[i+17 : i+21])) - keySizeUi64, err := safecast.Convert[uint64](keySize) - - if err != nil { - return err - } - - payloadSizeUi64, err := safecast.Convert[uint64](payloadSize) - - if err != nil { - return err - } - - key := fileData[i+21 : i+keySizeUi64+21] - payload := fileData[i+keySizeUi64+21 : i+(keySizeUi64+21)+payloadSizeUi64] - - //If the checksum is invalid, we ignore the rest of the file - if !record.ChecksumChecker(key, payload, binary.BigEndian.Uint64(time), checksum) { - break - } - - skiplist.Insert(string(key), fileData[i:i+(keySizeUi64+21)+payloadSizeUi64]) - i = i + (keySizeUi64 + 21) + payloadSizeUi64 - - } - } - - //We take the entire skiplist and write it into a new file - fileContents := skiplist.EntireList() - - //Technically speaking we can just recall the write to file again to make the new file - // Since the Entire write operation is there. - // So I'm planning on calling the Compact function after the write to file function goes through in the Memtable class - _, err := WriteToFile(fileContents, filePath) - - if err != nil { - return err - } - - // Delete all of the old files once the new file is committed - for _, fileInfo := range val { - err := os.Remove(filepath.Join(filePath, fileInfo.Name())) - - if err != nil { - return err - } - } - break - } - - } - - return nil -} - -func ExportBuckets(filePath string) map[file_buckets][]fs.FileInfo { - result, _ := buckets(filePath) - - return result -} - func ExportFooter(list [][]byte) ([]byte, error) { return createFooter(list) } diff --git a/internals/sstable/writer_test.go b/internals/sstable/writer_test.go index 13ec5de..57d4c9d 100644 --- a/internals/sstable/writer_test.go +++ b/internals/sstable/writer_test.go @@ -11,25 +11,25 @@ import ( "github.com/Vince-maple-byte/KeyData/internals/sstable" ) -func startUp(filePath string, data ...string) (int, error) { +func startUp(filePath string, data ...string) ([]*sstable.SSTFile, error) { - size := 0 + files := make([]*sstable.SSTFile, 0) for range 10 { file := make([][]byte, 0, 10) for i := range 10 { w, _ := record.CreateRecord(data[i], data[i], "PUT") file = append(file, w) } - _, err := sstable.WriteToFile(file, filePath) + sst, err := sstable.WriteToFile(file, filePath) if err != nil { - return 0, err + return files, err } - size++ + files = append(files, sst) } - return size, nil + return files, nil } func tearDown(filePath string) { @@ -69,14 +69,19 @@ func TestBucketsForFiles(t *testing.T) { for _, test := range tests { filePath := t.TempDir() - totalSize, err := startUp(filePath, test.data...) + files, err := startUp(filePath, test.data...) - if err != nil || totalSize == 0 { + if err != nil || len(files) == 0 { t.Fatalf("Could not start up the test") } - buckets := sstable.ExportBuckets(filePath) - average := totalSize / 10 + buckets, err := sstable.ExportBuckets(files) + + if err != nil { + t.Fatal(err) + } + + average := len(files) / 10 for key, bucket := range buckets { @@ -84,17 +89,19 @@ func TestBucketsForFiles(t *testing.T) { t.Logf("This bucket %v is %d length\n", key, len(bucket)) for _, item := range bucket { - if float64(item.Size()) < float64(average)*float64(key) { + if float64(item.Size) < float64(average)*float64(key) { t.Errorf("For test:%s\nThis file should not be in this bucket:%v\nFile name:%s\tFile size:%d", test.testName, - key, item.Name(), item.Size()) + key, item.FileName, item.Size) } } } } - } - //tearDown("../test") + for _, file := range files { + file.File.Close() + } + } } func TestCompactFiles(t *testing.T) { @@ -103,40 +110,32 @@ func TestCompactFiles(t *testing.T) { dir := t.TempDir() //size := 0 - for range 10 { - file := make([][]byte, 0, 10) - for i := range 10 { - w, _ := record.CreateRecord(data[i], data[i], "PUT") - file = append(file, w) - } - - _, err := sstable.WriteToFile(file, dir) + files, err := startUp(dir, data...) - t.Logf("Size: %d", len(file)) + if err != nil { + t.Fatalf("Not able to be create the SST Files for the test\n%v", err) + } - if err != nil || len(file) == 0 { - t.Fatalf("Not to able properly make the file: size=%d", len(file)) - } + files, err = sstable.Compact(files, dir) + fmt.Println(len(files)) + for _, file := range files { + file.File.Close() } - err := sstable.Compact(dir) - if err != nil { t.Errorf("Not able to complete the compaction\n Recieved this error code:\n%v", err) } - files, err := os.ReadDir(dir) + dirFiles, err := os.ReadDir(dir) if err != nil { t.Fatalf("Not able to access the directory for testing: %v", dir) } - if len(files) != 1 { + if len(dirFiles) != 1 { t.Errorf("Improper amount of files inside of the test directory:\nExpected: %d; Actual:%d", 1, len(files)) } - - //tearDown("../test") } func TestWriteToFile(t *testing.T) { @@ -148,13 +147,14 @@ func TestWriteToFile(t *testing.T) { fileContents = append(fileContents, r) } - ok, err := sstable.WriteToFile(fileContents, dir) + sstFile, err := sstable.WriteToFile(fileContents, dir) + sstFile.File.Close() if err != nil { t.Errorf("Error encountered: %v\n", err) } - if !ok { + if sstFile == nil { t.Errorf("Not able to create the file") } @@ -163,6 +163,7 @@ func TestWriteToFile(t *testing.T) { if err != nil { t.Fatalf("not able to open the file: %v", err) } + defer file.Close() info, _ := file.Stat() @@ -181,8 +182,8 @@ func TestFooter(t *testing.T) { fileContents = append(fileContents, c) } - _, err := sstable.WriteToFile(fileContents, dir) - + sstFile, err := sstable.WriteToFile(fileContents, dir) + sstFile.File.Close() if err != nil { t.Fatal(err.Error()) } From b4d51a63b9748f7b4bbed4a7f196d917a9cd8cff Mon Sep 17 00:00:00 2001 From: Ivers Date: Tue, 11 Aug 2026 00:26:50 -0400 Subject: [PATCH 4/9] Finished the testing, now need to fix the grpc and the main files --- internals/db/database.go | 26 ++- internals/db/database_test.go | 282 ++++++++++++++++++++++++++++ internals/memtable/memtable.go | 34 ++-- internals/memtable/memtable_test.go | 65 ++++--- internals/sstable/sstfile.go | 5 +- internals/sstable/writer.go | 1 - 6 files changed, 364 insertions(+), 49 deletions(-) diff --git a/internals/db/database.go b/internals/db/database.go index 6c05be4..de5283f 100644 --- a/internals/db/database.go +++ b/internals/db/database.go @@ -95,7 +95,7 @@ func (db *Database) Close() error { } func (db *Database) CreateMemtable() { - db.Memtable = memtable.CreateMemtable(db.Dir, db.WalPath) + db.Memtable = memtable.CreateMemtable(db.WalPath, db.Dir) } func (db *Database) Get(key string) ([]byte, error) { @@ -118,33 +118,47 @@ func (db *Database) Get(key string) ([]byte, error) { } func (db *Database) Put(key, val string) (bool, error) { - ok, err := db.Memtable.Write(key, val, "PUT") + ok, sstfile, err := db.Memtable.Write(key, val, "PUT") if !ok { return ok, err } + if sstfile != nil { + db.SSTFiles = append(db.SSTFiles, sstfile) + } + if err == nil { return true, err } compactErr := fmt.Errorf("Need to compact the files") - if errors.Is(err, compactErr) { - sstable.Compact(db.SSTFiles, db.Dir) - return true, nil + if err.Error() == compactErr.Error() { + newFiles, errF := sstable.Compact(db.SSTFiles, db.Dir) + + if errF != nil { + return false, errF + } + + db.SSTFiles = newFiles + return true, errF } return ok, err } func (db *Database) Delete(key string) (bool, error) { - ok, err := db.Memtable.Write(key, "", "DELETE") + ok, sstfile, err := db.Memtable.Write(key, "", "DELETE") if !ok { return ok, err } + if sstfile != nil { + db.SSTFiles = append(db.SSTFiles, sstfile) + } + if err == nil { return true, err } diff --git a/internals/db/database_test.go b/internals/db/database_test.go index 0385802..f08ee84 100644 --- a/internals/db/database_test.go +++ b/internals/db/database_test.go @@ -1 +1,283 @@ package database_test + +import ( + "fmt" + "os" + "path/filepath" + "testing" + + database "github.com/Vince-maple-byte/KeyData/internals/db" + "github.com/Vince-maple-byte/KeyData/internals/record" +) + +func TestDatabaseCreation(t *testing.T) { + temp := t.TempDir() + store := filepath.Join(temp, "store") + wal := filepath.Join(temp, "wal") + walPath := filepath.Join(wal, "memtable1.wal") + + err := os.Mkdir(store, 0600) + if err != nil { + t.Fatal(err) + } + + err = os.Mkdir(wal, 0600) + if err != nil { + t.Fatal(err) + } + + db, err := database.CreateDatabase(store, filepath.Join(wal, "memtable1.wal")) + + if err != nil { + t.Fatalf("Error in creating the database struct:\n%v", err) + } + + if db.Dir != store { + t.Errorf("Incorrect data directory saved\nExpected:%s\nActual:%s", store, db.Dir) + } + + if db.WalPath != walPath { + t.Errorf("Incorrect data directory saved\nExpected:%s\nActual:%s", wal, db.WalPath) + } + + if db.Memtable == nil { + t.Error("Unable to create the Memtable") + } + + if db.SSTFiles == nil { + t.Error("Unable to create the SSTable slice") + } +} + +func TestDatabaseCorrectNumberOfSSTFiles(t *testing.T) { + temp := t.TempDir() + store := filepath.Join(temp, "store") + wal := filepath.Join(temp, "wal") + + err := os.Mkdir(store, 0600) + if err != nil { + t.Fatal(err) + } + + err = os.Mkdir(wal, 0600) + if err != nil { + t.Fatal(err) + } + + db, err := database.CreateDatabase(store, filepath.Join(wal, "memtable1.wal")) + + if err != nil { + t.Fatalf("Error in creating the database struct:\n%v", err) + } + + for i := range 3200 * 3 { + ok, err := db.Put(fmt.Sprintf("key%d", i), fmt.Sprintf("val%d", i)) + + if !ok { + t.Fatalf("Something went wrong when adding data into the database:\n%v", err) + } + } + + dir, err := os.ReadDir(store) + + if err != nil { + t.Fatalf("Not able to read into the directory\n%v", err) + } + + if len(dir) != len(db.SSTFiles) { + t.Errorf("The length of the database struct: %d, does not match the length of the actual amount of file stored in the directory: %d", len(db.SSTFiles), len(dir)) + } + + for _, file := range db.SSTFiles { + file.Close() + } +} + +func TestDatabaseAfterCompactionHappens(t *testing.T) { + temp := t.TempDir() + store := filepath.Join(temp, "store") + wal := filepath.Join(temp, "wal") + + err := os.Mkdir(store, 0600) + if err != nil { + t.Fatal(err) + } + + err = os.Mkdir(wal, 0600) + if err != nil { + t.Fatal(err) + } + + db, err := database.CreateDatabase(store, filepath.Join(wal, "memtable1.wal")) + + if err != nil { + t.Fatalf("Error in creating the database struct:\n%v", err) + } + + for range 3200 * 4 { + ok, err := db.Put("key", "val") + + if !ok { + t.Fatalf("Something went wrong when adding data into the database:\n%v", err) + } + + if err != nil { + t.Errorf("Received an error here:\n%v", err) + } + } + + dir, err := os.ReadDir(store) + + if err != nil { + t.Fatalf("Not able to read into the directory\n%v", err) + } + + if len(db.SSTFiles) != 1 { + t.Errorf("The length of the database struct: %d, does not match the length of the actual amount of file stored in the directory: %d", len(db.SSTFiles), len(dir)) + } + + for _, file := range db.SSTFiles { + t.Logf("File name: %s", file.FileName) + file.Close() + } +} + +func TestDatabasePutValue(t *testing.T) { + + temp := t.TempDir() + store := filepath.Join(temp, "store") + wal := filepath.Join(temp, "wal") + + err := os.Mkdir(store, 0700) + if err != nil { + t.Fatal(err) + } + + err = os.Mkdir(wal, 0700) + if err != nil { + t.Fatal(err) + } + + db, err := database.CreateDatabase(store, filepath.Join(wal, "memtable1.wal")) + + if err != nil { + t.Fatalf("Error in creating the database struct:\n%v", err) + } + + ok, err := db.Put("key1", "val1") + + if !ok { + t.Fatalf("Unable to add a new key/value pair\nError:%v", err) + } + + if err != nil { + t.Fatalf("Error:%v", err) + } + + if db.Memtable.Size < 1 { + t.Errorf("Did not properly add the key/value pair into the memtable\nExpected Size:%d\nActual Size:%d", + 1, db.Memtable.Size) + } +} + +func TestDatabaseGetValue(t *testing.T) { + + temp := t.TempDir() + store := filepath.Join(temp, "store") + wal := filepath.Join(temp, "wal") + + err := os.Mkdir(store, 0700) + if err != nil { + t.Fatal(err) + } + + err = os.Mkdir(wal, 0700) + if err != nil { + t.Fatal(err) + } + + db, err := database.CreateDatabase(store, filepath.Join(wal, "memtable1.wal")) + + if err != nil { + t.Fatalf("Error in creating the database struct:\n%v", err) + } + + for i := range 10 { + ok, err := db.Put(fmt.Sprint("key", i), fmt.Sprint("val", i)) + + if !ok { + t.Fatalf("Unable to add a new key/value pair\nError:%v", err) + } + + if err != nil { + t.Fatalf("Error:%v", err) + } + } + + for i := range 10 { + value, err := db.Get(fmt.Sprint("key", i)) + + if err != nil { + t.Errorf("Unable to get the value for %s key\nError:%v", fmt.Sprint("key", i), err) + } + + if record.GetContents(value).Payload != fmt.Sprint("val", i) { + t.Errorf("Improper value saved\nExpected:%s\nActual:%s", fmt.Sprint("val", i), record.GetContents(value).Payload) + } + } + + db.Close() +} + +func TestDatabaseDeleteValue(t *testing.T) { + temp := t.TempDir() + store := filepath.Join(temp, "store") + wal := filepath.Join(temp, "wal") + + err := os.Mkdir(store, 0700) + if err != nil { + t.Fatal(err) + } + + err = os.Mkdir(wal, 0700) + if err != nil { + t.Fatal(err) + } + + db, err := database.CreateDatabase(store, filepath.Join(wal, "memtable1.wal")) + + if err != nil { + t.Fatalf("Error in creating the database struct:\n%v", err) + } + + for i := range 10 { + ok, err := db.Put(fmt.Sprint("key", i), fmt.Sprint("val", i)) + + if !ok { + t.Fatalf("Unable to add a new key/value pair\nError:%v", err) + } + + if err != nil { + t.Fatalf("Error:%v", err) + } + } + + for i := range 10 { + + _, err = db.Delete(fmt.Sprint("key", i)) + + if err != nil { + t.Errorf("Unable to get the value for %s key\nError:%v", fmt.Sprint("key", i), err) + } + + //The memtable should return an error stating that the record is deleted + _, err := db.Get(fmt.Sprint("key", i)) + + if err == nil { + t.Fatal("Expecting an error in which stating that the record has been deleted") + } + } +} + +//Test for restarting the Database with the Wal file being able to restart the Memtable +// Test for restarting the Database where the SSTables will be loaded into the SSTFile array diff --git a/internals/memtable/memtable.go b/internals/memtable/memtable.go index 76d0673..9552735 100644 --- a/internals/memtable/memtable.go +++ b/internals/memtable/memtable.go @@ -16,7 +16,7 @@ const MAX_SIZE = 3200 type Memtable struct { list *skiplist.Skiplist - size int + Size int WalFilePath string DataDir string } @@ -24,50 +24,50 @@ type Memtable struct { func CreateMemtable(wal, dir string) *Memtable { return &Memtable{ list: skiplist.CreateSkiplist(), - size: 0, + Size: 0, WalFilePath: wal, DataDir: dir, } } -func (m *Memtable) Write(key, value, operation string) (bool, error) { +func (m *Memtable) Write(key, value, operation string) (bool, *sstable.SSTFile, error) { record, err := record.CreateRecord(key, value, operation) if err != nil { - return false, err + return false, nil, err } ok, err := m.writeToWal(record) if !ok { - return false, err + return false, nil, err } m.list.Insert(key, record) - m.size += 1 + m.Size += 1 - if m.size >= MAX_SIZE { + if m.Size >= MAX_SIZE { content := m.list.EntireList() - _, err := sstable.WriteToFile(content, m.DataDir) + sstfile, err := sstable.WriteToFile(content, m.DataDir) if err != nil { - return false, err + return false, nil, err } //For now, when running test we just //err = sstable.Compact(m.DataDir) - m.list.EmptyList() - m.size = 0 - if err := os.Remove(m.WalFilePath); err != nil { - return false, err + return false, nil, err } - return true, fmt.Errorf("Need to compact the files") + m.list.EmptyList() + m.Size = 0 + + return true, sstfile, fmt.Errorf("Need to compact the files") } - return true, nil + return true, nil, nil } func (m *Memtable) Get(key string) ([]byte, error) { @@ -90,7 +90,7 @@ func (m Memtable) writeToWal(record []byte) (bool, error) { file, err := os.OpenFile(m.WalFilePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0600) if err != nil { - errMessage := fmt.Sprintf("Not able to create/open the wal file %s", m.WalFilePath) + errMessage := fmt.Sprintf("Not able to create/open the wal file %s\n%s", m.WalFilePath, err.Error()) return false, errors.New(errMessage) } @@ -152,7 +152,7 @@ func (m *Memtable) MemtableStartUp() (bool, error) { rec := append(header, keyValuePair...) m.list.Insert(string(rec[21:keySize+21]), rec) - m.size += 1 + m.Size += 1 i += int64(len(rec)) } diff --git a/internals/memtable/memtable_test.go b/internals/memtable/memtable_test.go index 649c90c..b5b3b89 100644 --- a/internals/memtable/memtable_test.go +++ b/internals/memtable/memtable_test.go @@ -1,10 +1,11 @@ package memtable import ( + "fmt" "os" "path/filepath" "testing" - "fmt" + "github.com/Vince-maple-byte/KeyData/internals/record" ) @@ -19,15 +20,15 @@ func TestWriteForSingleEntryInMemtable(t *testing.T) { os.Mkdir(dataDir, 0700) mem := CreateMemtable(walFile, dataDir) - prevSize := mem.size - _, err := mem.Write("key", "val", "PUT") + prevSize := mem.Size + _, _, err := mem.Write("key", "val", "PUT") if err != nil { t.Fatalf("Was not able to process the PUT operation for the memtable\n %v", err) } - if mem.size <= prevSize { - t.Errorf("The write operation did not successfully increase the memtable size") + if mem.Size <= prevSize { + t.Errorf("The write operation did not successfully increase the memtable Size") } } @@ -53,14 +54,14 @@ func TestWriteForMultipleEntriesInMemtable(t *testing.T) { } for _, test := range tests { - prevSize := mem.size - _, err := mem.Write(test.key, test.value, test.op) + prevSize := mem.Size + _, _, err := mem.Write(test.key, test.value, test.op) if err != nil { t.Fatalf("Was not able to process the PUT operation for the memtable") } - if mem.size <= prevSize { + if mem.Size <= prevSize { t.Errorf("The PUT operation did not successfully increase the memtable size") } } @@ -102,17 +103,24 @@ func TestDeleteForMemtable(t *testing.T) { mem := CreateMemtable(walFile, dataDir) mem.Write("key", "val", "PUT") - prevSize := mem.size - _, err := mem.Write("key", "", "DELETE") + prevSize := mem.Size + _, _, err := mem.Write("key", "", "DELETE") if err != nil { t.Fatalf("Was not able to process the delete operation for the memtable") } - if mem.size <= prevSize { + if mem.Size <= prevSize { t.Errorf("The delete operation did not successfully increase the memtable size") } + //The memtable should return an error stating that the record is deleted + _, err = mem.Get("key") + + if err == nil { + t.Fatal("Expecting an error in which stating that the record has been deleted") + } + } func TestWrite_FlushResetsState(t *testing.T) { @@ -126,18 +134,27 @@ func TestWrite_FlushResetsState(t *testing.T) { dataDir := filepath.Join(baseDir, "dir2") os.Mkdir(dataDir, 0700) - mt:= CreateMemtable(walFile, dataDir) + mt := CreateMemtable(walFile, dataDir) for i := 0; i < MAX_SIZE; i++ { key := "key" + string(rune(i)) - _, err := mt.Write(key, "value", "PUT") - if err != nil { + ok, sstfile, err := mt.Write(key, "value", "PUT") + if err != nil && !ok { t.Fatalf("unexpected error at write %d: %v", i, err) } + if sstfile != nil { + if err := sstfile.Close(); err != nil { + t.Fatalf("Unable to close the SSTable file:%v", err) + } + + if sstfile.FileName == "" { + t.Errorf("Unable to create a valid SSTable file") + } + } } - if mt.size != 0 { - t.Errorf("expected size to reset to 0 after flush, got %d", mt.size) + if mt.Size != 0 { + t.Errorf("expected Size to reset to 0 after flush, got %d", mt.Size) } } @@ -150,24 +167,24 @@ func TestMemtableStartUp(t *testing.T) { dataDir := filepath.Join(baseDir, "dir2") os.Mkdir(dataDir, 0700) - mt:= CreateMemtable(walFile, dataDir) + mt := CreateMemtable(walFile, dataDir) - ok,err := mt.Write("key", "val23", "PUT"); + ok, _, err := mt.Write("key", "val23", "PUT") if !ok { - t.Fatalf("Not able to write into the memtable, %v", err); + t.Fatalf("Not able to write into the memtable, %v", err) } for i := range 10 { - mt.Write(fmt.Sprintf("key%d",i), fmt.Sprintf("val%d", i), "PUT") + mt.Write(fmt.Sprintf("key%d", i), fmt.Sprintf("val%d", i), "PUT") } - fileInfo,_ := os.Lstat(walFile) + fileInfo, _ := os.Lstat(walFile) fmt.Println(fileInfo.Size()) mm := CreateMemtable(walFile, dataDir) - _,err = mm.MemtableStartUp() + _, err = mm.MemtableStartUp() if err != nil { t.Errorf("From trying to startup the memtable:%s\n", err.Error()) @@ -182,7 +199,7 @@ func TestMemtableStartUp(t *testing.T) { if len(data) < 21 { t.Fatalf("Was Not able to retrieve the data from the wal file, %v", data) } - + contents := record.GetContents(data) if contents.Payload != "val23" && contents.Key == "key" { @@ -192,7 +209,7 @@ func TestMemtableStartUp(t *testing.T) { for i := range 10 { content := record.GetContents(data) - if content.Payload != fmt.Sprintf("val%d", i) && contents.Key == fmt.Sprintf("key%d",i) { + if content.Payload != fmt.Sprintf("val%d", i) && contents.Key == fmt.Sprintf("key%d", i) { t.Errorf("Did not return the valid key/value pair\nKey:%s\nValue:%s", contents.Key, contents.Payload) } diff --git a/internals/sstable/sstfile.go b/internals/sstable/sstfile.go index d5cf121..ed688e0 100644 --- a/internals/sstable/sstfile.go +++ b/internals/sstable/sstfile.go @@ -4,6 +4,7 @@ import ( "encoding/binary" "fmt" "os" + "path/filepath" "strconv" "strings" @@ -125,7 +126,9 @@ func (s *SSTFile) Delete() error { } func (s *SSTFile) ParseGeneration() (int, error) { - str := strings.Split(s.FileName, "_")[1] + filename := filepath.Base(s.FileName) + + str := strings.Split(filename, "_")[1] str = strings.Split(str, ".")[0] idx, err := strconv.Atoi(str) //s.Generation = idx diff --git a/internals/sstable/writer.go b/internals/sstable/writer.go index 1970bc8..d635ef8 100644 --- a/internals/sstable/writer.go +++ b/internals/sstable/writer.go @@ -49,7 +49,6 @@ func WriteToFile(list [][]byte, filePath string) (*SSTFile, error) { } sstFile := &SSTFile{} - sstFile.Generation = maxIndex + 1 // #nosec G304 -- Creating SSTables enumerated from the internal storage directory. err = sstFile.Open(filepath.Join(filePath, filename)) From b0884098dd27d4aa82c65ff26d8cdb610d8ef33f Mon Sep 17 00:00:00 2001 From: Ivers Date: Tue, 11 Aug 2026 11:16:27 -0400 Subject: [PATCH 5/9] Fixed the errors that where happening in the grpc and main files. --- main.go | 11 ++++------- network/request_handler.go | 35 ++++++++++++++++------------------- 2 files changed, 20 insertions(+), 26 deletions(-) diff --git a/main.go b/main.go index 7029f82..026ee91 100644 --- a/main.go +++ b/main.go @@ -6,7 +6,7 @@ import ( "net" "os" - "github.com/Vince-maple-byte/KeyData/internals/memtable" + database "github.com/Vince-maple-byte/KeyData/internals/db" "github.com/Vince-maple-byte/KeyData/network" "google.golang.org/grpc" "google.golang.org/grpc/reflection" @@ -44,18 +44,15 @@ func main() { walPath := "./internals/wal/mem1.wal" dataDir := "./internals/data" - mem := memtable.CreateMemtable(walPath, dataDir) + db, err := database.CreateDatabase(dataDir, walPath) - ok, err := mem.MemtableStartUp() - - if !ok { + if err != nil { log.Fatal(err.Error()) os.Exit(1) } server := &network.Server{ - Memtable: mem, - DataDir: dataDir, + Database: db, } port := 5773 diff --git a/network/request_handler.go b/network/request_handler.go index df18243..0da4b6b 100644 --- a/network/request_handler.go +++ b/network/request_handler.go @@ -2,9 +2,10 @@ package network import ( context "context" + "fmt" + database "github.com/Vince-maple-byte/KeyData/internals/db" "github.com/Vince-maple-byte/KeyData/internals/record" - "github.com/Vince-maple-byte/KeyData/internals/sstable" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/types/known/timestamppb" @@ -13,12 +14,11 @@ import ( type Server struct { UnimplementedDataServer - Memtable InternalMemtable - DataDir string + Database *database.Database } func (s *Server) Search(ctx context.Context, search *SearchRequest) (*SearchResponse, error) { - res, err := s.searchHelper(search.GetKey()) + res, err := s.Database.Get(search.GetKey()) if err != nil { return &SearchResponse{CreatedAt: nil, Key: "", Payload: ""}, status.Error(codes.NotFound, err.Error()) @@ -35,8 +35,18 @@ func (s *Server) Search(ctx context.Context, search *SearchRequest) (*SearchResp func (s *Server) Create(ctx context.Context, create *CreateRequest) (*CreateResponse, error) { //mem := NetworkMemtable() - - ok, err := s.Memtable.Write(create.GetKey(), create.GetPayload(), create.GetOperation()) + var ok bool + var err error + + switch create.GetOperation() { + case "PUT": + ok, err = s.Database.Put(create.GetKey(), create.GetPayload()) + case "DELETE": + ok, err = s.Database.Delete(create.GetKey()) + default: + err = fmt.Errorf("Invalid operation") + ok = false + } if !ok { return &CreateResponse{Success: ok}, status.Error(codes.Unknown, err.Error()) @@ -44,16 +54,3 @@ func (s *Server) Create(ctx context.Context, create *CreateRequest) (*CreateResp return &CreateResponse{Success: ok}, nil } - -func (s Server) searchHelper(key string) ([]byte, error) { - //mem := NetworkMemtable() - - res, err := s.Memtable.Get(key) - - //Got to figure out how to pass in the filePath for the directory while it still being reusable - if err != nil { - return sstable.ReadFromAllFiles(key, s.DataDir) - } - - return res, err -} From ff7b642f84a3c70ee4ac845c5c36e88ab2896369 Mon Sep 17 00:00:00 2001 From: Ivers Date: Tue, 11 Aug 2026 20:08:29 -0400 Subject: [PATCH 6/9] Update of the code refactor --- internals/bloom/bloom.go | 2 +- internals/db/database.go | 6 +++--- internals/memtable/memtable.go | 6 +++--- internals/memtable/memtable_test.go | 6 +++--- internals/sstable/sstfile.go | 2 +- 5 files changed, 11 insertions(+), 11 deletions(-) diff --git a/internals/bloom/bloom.go b/internals/bloom/bloom.go index 1df2839..2dc4320 100644 --- a/internals/bloom/bloom.go +++ b/internals/bloom/bloom.go @@ -9,7 +9,7 @@ type Bloom struct { func NewBloom(numKeys int) (*Bloom, error) { if numKeys <= 0 { - return nil, fmt.Errorf("Improrer size for the bloom") + return nil, fmt.Errorf("improrer size for the bloom") } return &Bloom{}, nil diff --git a/internals/db/database.go b/internals/db/database.go index de5283f..81b2d7e 100644 --- a/internals/db/database.go +++ b/internals/db/database.go @@ -114,7 +114,7 @@ func (db *Database) Get(key string) ([]byte, error) { return res, nil } - return nil, fmt.Errorf("Not able to find key/value pair for %s", key) + return nil, fmt.Errorf("not able to find key/value pair for %s", key) } func (db *Database) Put(key, val string) (bool, error) { @@ -132,7 +132,7 @@ func (db *Database) Put(key, val string) (bool, error) { return true, err } - compactErr := fmt.Errorf("Need to compact the files") + compactErr := fmt.Errorf("need to compact the files") if err.Error() == compactErr.Error() { newFiles, errF := sstable.Compact(db.SSTFiles, db.Dir) @@ -163,7 +163,7 @@ func (db *Database) Delete(key string) (bool, error) { return true, err } - compactErr := fmt.Errorf("Need to compact the files") + compactErr := fmt.Errorf("need to compact the files") if errors.Is(err, compactErr) { sstable.Compact(db.SSTFiles, db.Dir) diff --git a/internals/memtable/memtable.go b/internals/memtable/memtable.go index 9552735..ab48aaf 100644 --- a/internals/memtable/memtable.go +++ b/internals/memtable/memtable.go @@ -64,7 +64,7 @@ func (m *Memtable) Write(key, value, operation string) (bool, *sstable.SSTFile, m.list.EmptyList() m.Size = 0 - return true, sstfile, fmt.Errorf("Need to compact the files") + return true, sstfile, fmt.Errorf("need to compact the files") } return true, nil, nil @@ -80,7 +80,7 @@ func (m *Memtable) Get(key string) ([]byte, error) { content := record.GetContents(records) if content.Tombstone != 0 { - return nil, errors.New("Key/Value pair doesn't exist") + return nil, errors.New("key/value pair doesn't exist") } return records, nil @@ -90,7 +90,7 @@ func (m Memtable) writeToWal(record []byte) (bool, error) { file, err := os.OpenFile(m.WalFilePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0600) if err != nil { - errMessage := fmt.Sprintf("Not able to create/open the wal file %s\n%s", m.WalFilePath, err.Error()) + errMessage := fmt.Sprintf("not able to create/open the wal file %s\n%s", m.WalFilePath, err.Error()) return false, errors.New(errMessage) } diff --git a/internals/memtable/memtable_test.go b/internals/memtable/memtable_test.go index b5b3b89..dc8a705 100644 --- a/internals/memtable/memtable_test.go +++ b/internals/memtable/memtable_test.go @@ -144,11 +144,11 @@ func TestWrite_FlushResetsState(t *testing.T) { } if sstfile != nil { if err := sstfile.Close(); err != nil { - t.Fatalf("Unable to close the SSTable file:%v", err) + t.Fatalf("unable to close the SSTable file:%v", err) } if sstfile.FileName == "" { - t.Errorf("Unable to create a valid SSTable file") + t.Errorf("unable to create a valid SSTable file") } } } @@ -172,7 +172,7 @@ func TestMemtableStartUp(t *testing.T) { ok, _, err := mt.Write("key", "val23", "PUT") if !ok { - t.Fatalf("Not able to write into the memtable, %v", err) + t.Fatalf("not able to write into the memtable, %v", err) } for i := range 10 { diff --git a/internals/sstable/sstfile.go b/internals/sstable/sstfile.go index ed688e0..4ae8a8f 100644 --- a/internals/sstable/sstfile.go +++ b/internals/sstable/sstfile.go @@ -35,7 +35,7 @@ type SSTFile struct { func (s *SSTFile) Open(filepath string) error { if s.File != nil { - return fmt.Errorf("File is already open") + return fmt.Errorf("file is already open") } file, err := os.OpenFile(filepath, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0666) From 26a2c0443c3db85360eca9938eb55199bc24652a Mon Sep 17 00:00:00 2001 From: Vince-maple-byte <73848683+Vince-maple-byte@users.noreply.github.com> Date: Tue, 11 Aug 2026 20:18:31 -0400 Subject: [PATCH 7/9] Changed the file permission of the test. The ci/cd pipeline for the go tests should work now. --- internals/db/database_test.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/internals/db/database_test.go b/internals/db/database_test.go index f08ee84..2cf6f60 100644 --- a/internals/db/database_test.go +++ b/internals/db/database_test.go @@ -54,12 +54,12 @@ func TestDatabaseCorrectNumberOfSSTFiles(t *testing.T) { store := filepath.Join(temp, "store") wal := filepath.Join(temp, "wal") - err := os.Mkdir(store, 0600) + err := os.Mkdir(store, 0700) if err != nil { t.Fatal(err) } - err = os.Mkdir(wal, 0600) + err = os.Mkdir(wal, 0700) if err != nil { t.Fatal(err) } @@ -98,12 +98,12 @@ func TestDatabaseAfterCompactionHappens(t *testing.T) { store := filepath.Join(temp, "store") wal := filepath.Join(temp, "wal") - err := os.Mkdir(store, 0600) + err := os.Mkdir(store, 0700) if err != nil { t.Fatal(err) } - err = os.Mkdir(wal, 0600) + err = os.Mkdir(wal, 0700) if err != nil { t.Fatal(err) } From 35fd895b47d962b7aa6af9a2730a0fdb97816248 Mon Sep 17 00:00:00 2001 From: Vince-maple-byte <73848683+Vince-maple-byte@users.noreply.github.com> Date: Wed, 12 Aug 2026 13:57:42 -0400 Subject: [PATCH 8/9] Final changes need to pass the CI/CD pipeline. --- internals/db/database.go | 9 ++++++++- internals/sstable/compaction.go | 16 ++++++++++++++-- internals/sstable/reader_test.go | 2 +- internals/sstable/sstfile.go | 13 +++++++++---- internals/sstable/writer.go | 17 ++++++----------- internals/sstable/writer_test.go | 17 ++--------------- network/request_handler.go | 2 +- 7 files changed, 41 insertions(+), 35 deletions(-) diff --git a/internals/db/database.go b/internals/db/database.go index 81b2d7e..803ee45 100644 --- a/internals/db/database.go +++ b/internals/db/database.go @@ -166,7 +166,14 @@ func (db *Database) Delete(key string) (bool, error) { compactErr := fmt.Errorf("need to compact the files") if errors.Is(err, compactErr) { - sstable.Compact(db.SSTFiles, db.Dir) + newFiles, err := sstable.Compact(db.SSTFiles, db.Dir) + + if err != nil { + return false, err; + } + + db.SSTFiles = newFiles; + return true, nil } diff --git a/internals/sstable/compaction.go b/internals/sstable/compaction.go index 36adf6c..f5ec25e 100644 --- a/internals/sstable/compaction.go +++ b/internals/sstable/compaction.go @@ -11,7 +11,16 @@ import ( ) //Change these methods to be SSTFile complient - +type file_buckets float64 + +const ( + COMPACTION_SIZE = 4 + INDEX_BLOCK = 20 + SMALL file_buckets = 0.5 + MEDIUM file_buckets = 1.0 + LARGE file_buckets = 1.5 + OVERSIZE file_buckets = 2.0 +) // We are going to be doing size based compaction for compacting these files // The amount of files that need to be a similar size // TODO: Need to make the bucket map into a persistent map that is used throughout the entire state of the @@ -82,7 +91,10 @@ func Compact(files []*SSTFile, dir string) ([]*SSTFile, error) { } //We do this so that we only take into account the file block, and not the index or footer if binary.BigEndian.Uint64(footer[16:]) != 0xDEADBEEFDEADBEEF { - file.Close() + if err := file.Close(); err != nil { + return files, fmt.Errorf("warning: not able to close the file %s: %v", + file.FileName, err) + } if err := os.Remove(file.FileName); err != nil { return files, fmt.Errorf("warning: failed to remove corrupt SSTable %s: %v", file.FileName, err) diff --git a/internals/sstable/reader_test.go b/internals/sstable/reader_test.go index 5c144d8..d8b9a80 100644 --- a/internals/sstable/reader_test.go +++ b/internals/sstable/reader_test.go @@ -214,12 +214,12 @@ func TestReadFromAllFiles(t *testing.T) { d.Insert(strconv.Itoa(i), r) } file, err := sstable.WriteToFile(d.EntireList(), dir) - defer file.Close() if err != nil { t.Fatalf("Error in writing the file: %s", err.Error()) } + defer file.Close() files = append(files, file) } diff --git a/internals/sstable/sstfile.go b/internals/sstable/sstfile.go index 4ae8a8f..f00a4b6 100644 --- a/internals/sstable/sstfile.go +++ b/internals/sstable/sstfile.go @@ -38,14 +38,14 @@ func (s *SSTFile) Open(filepath string) error { return fmt.Errorf("file is already open") } - file, err := os.OpenFile(filepath, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0666) + s.FileName = filepath + file, err := os.OpenFile(s.FileName, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0666) if err != nil { return err } s.File = file - s.FileName = filepath s.Generation, err = s.ParseGeneration() if err != nil { @@ -70,6 +70,11 @@ func (s *SSTFile) PopulateIndex(blocks []byte) { indexBlocks := make([]*IndexBlock, 0) for i := 0; i < len(blocks); { + iU32, err := safecast.Convert[uint32](i) + + if err != nil { + return + } keySize := binary.BigEndian.Uint32(blocks[i : i+4]) keySizeInt, err := safecast.Convert[int](keySize) @@ -81,9 +86,9 @@ func (s *SSTFile) PopulateIndex(blocks []byte) { break } - indexKey := string(blocks[i+4 : uint32(i+4)+keySize]) + indexKey := string(blocks[i+4 : iU32+4+keySize]) - keyOffset := binary.BigEndian.Uint64(blocks[uint32(i+4)+keySize : uint32(i+12)+keySize]) + keyOffset := binary.BigEndian.Uint64(blocks[iU32+4+keySize : iU32+12+keySize]) //fmt.Println(keyOffset) index := &IndexBlock{ Key: indexKey, diff --git a/internals/sstable/writer.go b/internals/sstable/writer.go index d635ef8..7a754f1 100644 --- a/internals/sstable/writer.go +++ b/internals/sstable/writer.go @@ -13,18 +13,8 @@ import ( "github.com/ccoveille/go-safecast/v2" ) -type file_buckets float64 - -const ( - COMPACTION_SIZE = 4 - INDEX_BLOCK = 20 - SMALL file_buckets = 0.5 - MEDIUM file_buckets = 1.0 - LARGE file_buckets = 1.5 - OVERSIZE file_buckets = 2.0 -) -// TODO: Change WriteToFile to be SSTFile struct complient + func WriteToFile(list [][]byte, filePath string) (*SSTFile, error) { filename := "" files, err := os.ReadDir(filePath) @@ -75,6 +65,11 @@ func WriteToFile(list [][]byte, filePath string) (*SSTFile, error) { return nil, err } indexOffset, err := safecast.Convert[uint64](sstFile.Size - int64(len(index)+len(footer))) + + if err != nil { + return nil, err; + } + sstFile.Footer = &Footer{ Magic: uint64(0xDEADBEEFDEADBEEF), IndexOffset: indexOffset, diff --git a/internals/sstable/writer_test.go b/internals/sstable/writer_test.go index 57d4c9d..1bf79ab 100644 --- a/internals/sstable/writer_test.go +++ b/internals/sstable/writer_test.go @@ -32,15 +32,6 @@ func startUp(filePath string, data ...string) ([]*sstable.SSTFile, error) { return files, nil } -func tearDown(filePath string) { - files, _ := os.ReadDir(filePath) - - for _, file := range files { - os.Remove(filepath.Join(filePath, file.Name())) - - } -} - func TestBucketsForFiles(t *testing.T) { tests := []struct { testName string @@ -148,16 +139,12 @@ func TestWriteToFile(t *testing.T) { } sstFile, err := sstable.WriteToFile(fileContents, dir) - sstFile.File.Close() if err != nil { - t.Errorf("Error encountered: %v\n", err) - } - - if sstFile == nil { - t.Errorf("Not able to create the file") + t.Fatalf("Error encountered: %v\n", err) } + sstFile.File.Close() file, err := os.Open(filepath.Join(dir, "kd_1.sst")) if err != nil { diff --git a/network/request_handler.go b/network/request_handler.go index 0da4b6b..2fae908 100644 --- a/network/request_handler.go +++ b/network/request_handler.go @@ -44,7 +44,7 @@ func (s *Server) Create(ctx context.Context, create *CreateRequest) (*CreateResp case "DELETE": ok, err = s.Database.Delete(create.GetKey()) default: - err = fmt.Errorf("Invalid operation") + err = fmt.Errorf("invalid operation") ok = false } From 250149d15a388593776555c7c5cdc092c86c126d Mon Sep 17 00:00:00 2001 From: Vince-maple-byte <73848683+Vince-maple-byte@users.noreply.github.com> Date: Wed, 12 Aug 2026 14:00:41 -0400 Subject: [PATCH 9/9] Fixed the file persmissions for the sstfile --- internals/sstable/sstfile.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internals/sstable/sstfile.go b/internals/sstable/sstfile.go index f00a4b6..c2c5766 100644 --- a/internals/sstable/sstfile.go +++ b/internals/sstable/sstfile.go @@ -39,7 +39,7 @@ func (s *SSTFile) Open(filepath string) error { } s.FileName = filepath - file, err := os.OpenFile(s.FileName, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0666) + file, err := os.OpenFile(s.FileName, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0600) if err != nil { return err