diff --git a/.gitignore b/.gitignore index 2963b31..f3df9a6 100644 --- a/.gitignore +++ b/.gitignore @@ -3,4 +3,7 @@ #Ignore .txt and log files since these are for testing. *.txt -*.log \ No newline at end of file +*.log +*.sst +*.wal + diff --git a/go.mod b/go.mod index 3946826..3a9e041 100644 --- a/go.mod +++ b/go.mod @@ -1,3 +1,15 @@ module github.com/Vince-maple-byte/KeyData go 1.25.4 + +require ( + google.golang.org/grpc v1.81.1 + google.golang.org/protobuf v1.36.11 +) + +require ( + golang.org/x/net v0.51.0 // indirect + golang.org/x/sys v0.42.0 // indirect + golang.org/x/text v0.34.0 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260226221140-a57be14db171 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..44c671d --- /dev/null +++ b/go.sum @@ -0,0 +1,38 @@ +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/otel v1.43.0 h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I= +go.opentelemetry.io/otel v1.43.0/go.mod h1:JuG+u74mvjvcm8vj8pI5XiHy1zDeoCS2LB1spIq7Ay0= +go.opentelemetry.io/otel/metric v1.43.0 h1:d7638QeInOnuwOONPp4JAOGfbCEpYb+K6DVWvdxGzgM= +go.opentelemetry.io/otel/metric v1.43.0/go.mod h1:RDnPtIxvqlgO8GRW18W6Z/4P462ldprJtfxHxyKd2PY= +go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg= +go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg= +go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw= +go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A= +go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A= +go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0= +golang.org/x/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo= +golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y= +golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo= +golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk= +golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA= +gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260226221140-a57be14db171 h1:ggcbiqK8WWh6l1dnltU4BgWGIGo+EVYxCaAPih/zQXQ= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260226221140-a57be14db171/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.81.1 h1:VnnIIZ88UzOOKLukQi+ImGz8O1Wdp8nAGGnvOfEIWQQ= +google.golang.org/grpc v1.81.1/go.mod h1:xGH9GfzOyMTGIOXBJmXt+BX/V0kcdQbdcuwQ/zNw42I= +google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= +google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= diff --git a/internals/file/file.go b/internals/file/file.go deleted file mode 100644 index 86feb02..0000000 --- a/internals/file/file.go +++ /dev/null @@ -1,266 +0,0 @@ -package file - -import ( - "encoding/binary" - "errors" - "fmt" - "os" - "time" - - records "github.com/Vince-maple-byte/KeyData/internals/record" -) - -type File struct { - File *os.File - Size int64 - Index map[string]int64 -} - -/*This is how we are opening up the file, reading, and writing to the file -We are using os.OpenFile that returns an os.File pointer that allows us to use methods such as -file.Seek(offset int64, whence int) which sets the byte offset in our program to point to where in the file each record is located, - and whence just specifies if we should start in the beginning of the file, the current offset, or the end of the file - -file.Read(b []byte) which would read from the file starting at the current byte offset the os.File is pointing to, -and then adding n bytes where n is the underlining size of the byte slice (remember a slice is just a dynamic array in go); - -file.OpenFile() allows us to open the file and return the *os.File, with this we can run the methods mentioned above, - and pass in some arguments such as os.O_APPEND to specify what we can do in the file; - the prem argument is only valid if the OpenFile is creating the file for the first time - and specifies the special bits(setuid, sticky bit), the owner permission, group permissions, and the global permissions in this format - 0664 - Where for the last three digits - 4 = read permission - 2 = write permission - 1 = execute permission - 0 = no permission - -*/ - -// If the file doesn't exist we will just return an empty File object and an network error code since we don't want the -// program to stop running for an invalid file path. This would help us in the future when we accept network calls. -func OpenFile(fileName string) (File, error) { - //Just created fileEmpty in here so that I can reuse it in other areas where an error can happen - fileEmpty := File{ - File: nil, - Size: 0, - Index: nil, - } - file, err := os.OpenFile(fileName, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0644) - - if err != nil { - return fileEmpty, err - } - - fileInfo, err := file.Stat() - - if err != nil { - return fileEmpty, err - } - - size := fileInfo.Size() - - f := File{ - File: file, - Size: size, - Index: make(map[string]int64), - } - - if size > 0 { - fileContents, errf := f.ReadFile(0, int(f.Size)) - - if errf != nil { - return fileEmpty, errf - } - - f.Index = f.PopulateMap(fileContents) - } - - return f, err -} - -// Were are basically going to go through the entire file record by record decoding the each record -// We are also going to checking the checksum for each record -// And if we encounter an incorrect checksum we will stop from there -// Potential solution: We can make a new file writing the values from the beginning until the records were valid and then just update the file struct to be only pointing to the valid log and deleting the old log -// NOTE: The value part of the map is for the byte offset to locate the record -func (f *File) PopulateMap(fileContents []byte) map[string]int64 { - m := make(map[string]int64) - index := 0 - - for index < len(fileContents) { - - keySizeSection := index + 13 - - if keySizeSection >= len(fileContents) { - break - } - - keySize := binary.BigEndian.Uint32(fileContents[keySizeSection : index+17]) - - payloadSizeSection := index + 17 - - if payloadSizeSection >= len(fileContents) { - break - } - - payloadSize := binary.BigEndian.Uint32(fileContents[payloadSizeSection : index+21]) - - payloadByte := fileContents[int(keySize)+(index+21) : int(keySize)+(index+21)+int(payloadSize)] - - checksum := binary.BigEndian.Uint32(fileContents[index+8 : index+12]) - - var validChecksum bool = records.ChecksumChecker(payloadByte, checksum) - - if validChecksum { - key := string(fileContents[index+21 : int(keySize)+(index+21)]) - - m[key] = int64(index) - - //The length of the record will always be 21 + keySize + payloadSize - index += int(21 + keySize + payloadSize) - } else { - f.trimFile(index) - break - } - - } - - return m -} - -// This method gets executed when a record in the database is found out to be invalid through checksum. -// When this is the case we just take all of the values before the invalid one and save it to a new file. -func (f *File) trimFile(endPoint int) { - newFileContents, err := f.ReadFile(0, endPoint) - - if err != nil { - panic("Something went wrong with trying to read the file") - } - - fileName := f.File.Name() - - err = os.Remove(fileName) - - if err != nil { - panic("Unable to delete the file " + fileName) - } - - f.File, err = os.OpenFile(fileName, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0644) - - if err != nil { - panic("Unable to open/create file " + fileName) - } - - size, err := f.File.Write(newFileContents) - - if err != nil { - panic("Was not able to write into " + fileName) - } - - f.Size = int64(size) - -} - -// If we get -1 that means that the file was not successfully written. -func (f *File) PutContents(key, payload, operation string) (amountAdded int, err error) { - var recordBytes []byte - recordBytes, err = records.CreateRecord(key, payload, operation) - - if err != nil { - return -1, err - } - - amountAdded, err = f.File.Write(recordBytes) - - if err != nil { - return -1, err - } - - f.UpdateMap(recordBytes) - - f.Size += int64(amountAdded) - - return amountAdded, nil -} - -// In the case where the key is deleted, we can just return the offset of where the deleted record exists just like a regular read -// and from there we can just read from the tombstone whether it was deleted or not -func (f *File) UpdateMap(contents []byte) { - var offset int64 = f.Size - - keySize := binary.BigEndian.Uint32(contents[13:17]) - - key := string(contents[21 : keySize+21]) - - f.Index[key] = offset -} - -// We can now read from the file from the starting byte offset to the ending byte -func (f *File) ReadFile(startingPoint, endingPoint int) ([]byte, error) { - if startingPoint < 0 { - panic("Incorrect starting point") - } - - file := f.File - - //We would change the number for seek to be the specific byte offset in the map from the file struct - _, err := file.Seek(int64(startingPoint), 0) - if err != nil { - fmt.Println(err) - return nil, err - } - - b := make([]byte, endingPoint) - - _, error := file.Read(b) - - if error != nil { - fmt.Println(error) - return nil, error - } - - //fmt.Println(string(f)); - return b, nil - -} - -func (f *File) GetContents(key string) (deleted bool, payload string, timestamp time.Time, err error) { - keyOffset := f.Index[key] - - content, e := f.ReadFile(int(keyOffset), int(f.Size)) - - if e != nil { - err = e - return - } - - timeByte := content[:8] - tombstone := content[12] - keySize := binary.BigEndian.Uint32(content[13:17]) - payloadSize := binary.BigEndian.Uint32(content[17:21]) - timestamp = time.Unix(0, int64(binary.BigEndian.Uint64(timeByte))) - - if tombstone == 1 { - deleted = true - payload = "" - return - } else { - deleted = false - } - - if err != nil { - return - } - - keySaved := string(content[21 : keySize+21]) - - if key != keySaved { - err = errors.New("Invalid key given") - return - } - - payload = string(content[keySize+21 : (keySize+21)+payloadSize]) - - return -} diff --git a/internals/file/file_test.go b/internals/file/file_test.go deleted file mode 100644 index a96f12c..0000000 --- a/internals/file/file_test.go +++ /dev/null @@ -1,343 +0,0 @@ -package file_test - -import ( - "maps" - "testing" - - "github.com/Vince-maple-byte/KeyData/internals/file" - records "github.com/Vince-maple-byte/KeyData/internals/record" -) - -//TODO: Make test cases for the missing methods - -func TestReadFile(t *testing.T) { - //Testing table - tests := []struct { - testName string - fileName string - key string - payload string - operation string - expected int - }{ - { - testName: "TestingInitFile", - fileName: "./init.txt", - key: "a", - payload: "a", - operation: "PUT", - expected: 23, //The length of the byte array - }, - } - - for _, test := range tests { - - t.Run(test.testName, func(t *testing.T) { - //When - f, err := file.OpenFile(test.fileName) - if err != nil { - t.Errorf("%v was not able to be opened", test.fileName) - } - - var byteContent []byte - byteContent, err = f.ReadFile(0, 23) - - if err != nil { - t.Error(err) - } - - //Actual - if len(byteContent) != test.expected { - t.Errorf("For %v, this amount was added %v instead of %v", test.testName, len(byteContent), test.expected) - } - - }) - - } - -} - -func TestPutContents(t *testing.T) { - //Testing table - tests := []struct { - testName string - fileName string - key string - payload string - operation string - expected int - }{ - { - testName: "TestingPut", - fileName: "./init.txt", - key: "a", - payload: "a", - operation: "PUT", - expected: 23, - }, - { - testName: "TestingDelete", - fileName: "./init.txt", - key: "a", - payload: "", - operation: "DELETE", - expected: 22, - }, - { - testName: "TestingDelete", - fileName: "./init.txt", - key: "b", - payload: "b", - operation: "PUT", - expected: 23, - }, - } - - for _, test := range tests { - - t.Run(test.testName, func(t *testing.T) { - //When - f, err := file.OpenFile(test.fileName) - if err != nil { - t.Errorf("%v was not able to be opened", test.fileName) - } - - var amountAdded int - amountAdded, err = f.PutContents(test.key, test.payload, test.operation) - - if err != nil { - t.Error(err) - } - - //Actual - if amountAdded != test.expected { - t.Errorf("For %v, this amount was added %v instead of %v", test.testName, amountAdded, test.expected) - } - - }) - - } -} - -func TestGetContents(t *testing.T) { - //Testing table - tests := []struct { - testName string - fileName string - key string - deleted bool - expectedPayload string - }{ - { - testName: "TestingInitFileA", - fileName: "./init.txt", - key: "a", - deleted: true, - expectedPayload: "", - }, - { - testName: "TestingInitFileB", - fileName: "./init.txt", - key: "b", - deleted: false, - expectedPayload: "b", - }, - } - - for _, test := range tests { - - t.Run(test.testName, func(t *testing.T) { - //When - f, err := file.OpenFile(test.fileName) - if err != nil { - t.Errorf("%v was not able to be opened", test.fileName) - } - - delete, payload, _, errf := f.GetContents(test.key) - - if errf != nil { - t.Error(errf) - } - - //Actual - if delete != test.deleted || string(payload) != test.expectedPayload { - t.Errorf("For %v, this record %v returns an invalid value %v and %v", test.testName, test.key, delete, string(payload)) - } - - }) - - } -} - -func TestOpenFile(t *testing.T) { - //Testing table - tests := []struct { - testName string - fileName string - expectedSize int64 - expectedErr bool - }{ - { - testName: "IfFileDoesExists", - fileName: "./init.txt", - expectedSize: 0, - expectedErr: false, - }, - { - testName: "IfFileDoesNotExists", - fileName: "./internals/file/fake.txt", - expectedSize: 0, - expectedErr: true, - }, - } - - for _, test := range tests { - - t.Run(test.testName, func(t *testing.T) { - //When - f, err := file.OpenFile(test.fileName) - - //Actual - if test.expectedErr && err == nil { - t.Errorf("Expected error, got nil") - } - - if !test.expectedErr && err != nil { - t.Errorf("Unexpected error: %v", err) - } - - if f.Size != test.expectedSize { - t.Errorf( - "incorrect file size: got %d, want %d", - f.Size, - test.expectedSize, - ) - } - - if !test.expectedErr && f.File == nil { - t.Errorf("expected file handle to be non-nil") - } - - }) - - } -} - -func TestPopulateMap(t *testing.T) { - record1, _ := records.CreateRecord("foo", "1", "PUT") - recordF, _ := records.CreateRecord("foo", "4", "PUT") - record2, _ := records.CreateRecord("bar", "1", "PUT") - record3, _ := records.CreateRecord("n", "2", "PUT") - record4, _ := records.CreateRecord("b", "3", "PUT") - combinedUnique := append(append([]byte{}, record1...), record2...) - combinedUnique = append(combinedUnique, record3...) - combinedUnique = append(combinedUnique, record4...) - combinedSame := append(append([]byte{}, record1...), recordF...) - tests := []struct { - testName string - ByteSlice []byte - expectedMap map[string]int64 - }{ - { - testName: "IfByteSliceIsEmpty", - ByteSlice: make([]byte, 0), - expectedMap: make(map[string]int64), - }, - { - testName: "IfByteSliceHasValues", - ByteSlice: record1, - expectedMap: map[string]int64{"foo": 0}, - }, - { - testName: "IfByteSliceHasMultipleValues", - ByteSlice: combinedUnique, - expectedMap: map[string]int64{"foo": 0, "bar": 25, "b": 73, "n": 50}, - }, - { - testName: "IfByteSliceHasMultiplePutsOftheSameKey", - ByteSlice: combinedSame, - expectedMap: map[string]int64{"foo": 25}, - }, - } - - for _, test := range tests { - //When - f := file.File{} - m := f.PopulateMap(test.ByteSlice) - - if !maps.Equal(test.expectedMap, m) { - t.Errorf("Map mismatch between\nexpected %v\nand\nactual %v", test.expectedMap, m) - } - } -} - -func TestUpdateMap(t *testing.T) { - record1, _ := records.CreateRecord("foo", "1", "PUT") - record2, _ := records.CreateRecord("foo", "2", "PUT") - record3, _ := records.CreateRecord("bar", "3", "PUT") - emptyMap := make(map[string]int64) - fileEmpty := file.File{ - File: nil, - Size: 0, - Index: emptyMap, - } - - fileNonEmpty := file.File{ - File: nil, - Size: 25, - Index: map[string]int64{"foo": 0}, - } - - tests := []struct { - testName string - ByteSlice []byte - fileStruct file.File - expectedFileStruct file.File - }{ - { - testName: "IfMapWasEmpty", - ByteSlice: record1, - fileStruct: fileEmpty, - expectedFileStruct: file.File{ - File: nil, - Size: 0, //Keeping track of the size doesn't matter since the write function will handle that - Index: map[string]int64{"foo": 0}, - }, - }, - { - testName: "IfMapIsNotEmptyAndWeAreAddingAnExistingKey", - ByteSlice: record2, - fileStruct: fileNonEmpty, - expectedFileStruct: file.File{ - File: nil, - Size: 0, //Keeping track of the size doesn't matter since the write function will handle that - Index: map[string]int64{"foo": 25}, - }, - }, - { - testName: "IfMapIsNotEmptyAndWeAreAddingANewKey", - ByteSlice: record3, - fileStruct: file.File{ - File: nil, - Size: 25, - Index: map[string]int64{"foo": 0}, - }, - expectedFileStruct: file.File{ - File: nil, - Size: 0, //Keeping track of the size doesn't matter since the write function will handle that - Index: map[string]int64{"foo": 0, "bar": 25}, - }, - }, - } - - for _, test := range tests { - //When - test.fileStruct.UpdateMap(test.ByteSlice) - - //Then - if !maps.Equal(test.fileStruct.Index, test.expectedFileStruct.Index) { - t.Errorf("Map mismatch between expected %v and actual %v for test %v", - test.fileStruct.Index, test.expectedFileStruct.Index, test.testName) - } - } -} diff --git a/internals/memtable/memtable.go b/internals/memtable/memtable.go new file mode 100644 index 0000000..455a772 --- /dev/null +++ b/internals/memtable/memtable.go @@ -0,0 +1,152 @@ +package memtable + +import ( + "encoding/binary" + "errors" + "os" + + "fmt" + "github.com/Vince-maple-byte/KeyData/internals/record" + "github.com/Vince-maple-byte/KeyData/internals/sstable" +) + +const MAX_SIZE = 3200 + +type Memtable struct { + list *Skiplist + size int + WalFilePath string + DataDir string +} + +func CreateMemtable(wal, dir string) *Memtable { + return &Memtable{ + list: CreateSkiplist(), + size: 0, + WalFilePath: wal, + DataDir: dir, + } +} + +func (m *Memtable) Write(key, value, operation string) (bool, error) { + record, err := record.CreateRecord(key, value, operation) + + if err != nil { + return false, err + } + + ok, err := m.writeToWal(record) + + if !ok { + return false, err + } + + m.list.Insert(key, record) + m.size += 1 + + 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) + + if err != nil { + return false, err + } + + m.list.EmptyList() + m.size = 0 + os.Remove(m.WalFilePath) + + } + + return true, nil +} + +func (m *Memtable) Get(key string) ([]byte, error) { + records, err := m.list.Search(key) + + if err != nil { + return nil, err + } + + content := record.GetContents(records) + + if content.Tombstone != 0 { + return nil, errors.New("Key/Value pair doesn't exist") + } + + return records, nil +} + +func (m Memtable) writeToWal(record []byte) (bool, error) { + file, err := os.OpenFile(m.WalFilePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644) + + if err != nil { + errMessage := fmt.Sprintf("Not able to create/open the wal file %s", m.WalFilePath) + return false, errors.New(errMessage); + } + + defer file.Close() + + _, err = file.Write(record) + + if err != nil { + return false, err + } + + return true, nil +} + +func (m *Memtable) MemtableStartUp() (bool, error) { + file, err := os.OpenFile(m.WalFilePath, os.O_APPEND|os.O_CREATE|os.O_RDONLY, 0644) + + defer file.Close() + + if err != nil { + return false, err + } + + + + fileInfo, err := file.Stat() + + if err != nil { + return false, err + } + + for i := int64(0); i < fileInfo.Size(); { + header := make([]byte, 21) + file.ReadAt(header, i) + + keySize := binary.BigEndian.Uint32(header[13:17]) + payloadSize := binary.BigEndian.Uint32(header[17:21]) + + keyValuePair := make([]byte, keySize+payloadSize) + file.ReadAt(keyValuePair, i+21) + + ok := record.ChecksumChecker( + keyValuePair[:keySize], + keyValuePair[keySize:], + binary.BigEndian.Uint64(header[0:8]), + binary.BigEndian.Uint32(header[8:12]), + ) + + if !ok { + break + } + + rec := append(header, keyValuePair...) + m.list.Insert(string(rec[21: keySize + 21]), rec) + + m.size += 1 + i += int64(len(rec)) + } + + return true, nil +} diff --git a/internals/memtable/memtable_test.go b/internals/memtable/memtable_test.go new file mode 100644 index 0000000..649c90c --- /dev/null +++ b/internals/memtable/memtable_test.go @@ -0,0 +1,200 @@ +package memtable + +import ( + "os" + "path/filepath" + "testing" + "fmt" + "github.com/Vince-maple-byte/KeyData/internals/record" +) + +func TestWriteForSingleEntryInMemtable(t *testing.T) { + + baseDir := t.TempDir() + + walDir := filepath.Join(baseDir, "wal") + os.Mkdir(walDir, 0700) + walFile := filepath.Join(walDir, "mem1.wal") + dataDir := filepath.Join(baseDir, "data") + os.Mkdir(dataDir, 0700) + + mem := CreateMemtable(walFile, dataDir) + 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") + } +} + +func TestWriteForMultipleEntriesInMemtable(t *testing.T) { + baseDir := t.TempDir() + + walDir := filepath.Join(baseDir, "dir1") + os.Mkdir(walDir, 0700) + walFile := filepath.Join(walDir, "mem1.wal") + + dataDir := filepath.Join(baseDir, "dir2") + os.Mkdir(dataDir, 0700) + mem := CreateMemtable(walFile, dataDir) + + tests := []struct { + key, value, op string + }{ + {"key1", "val1", "PUT"}, + {"key2", "val2", "PUT"}, + {"key3", "val3", "PUT"}, + {"key4", "val4", "PUT"}, + {"key5", "val5", "PUT"}, + } + + for _, test := range tests { + 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 { + t.Errorf("The PUT operation did not successfully increase the memtable size") + } + } + +} + +func TestGetForMemtable(t *testing.T) { + baseDir := t.TempDir() + + walDir := filepath.Join(baseDir, "dir1") + os.Mkdir(walDir, 0700) + walFile := filepath.Join(walDir, "mem1.wal") + + dataDir := filepath.Join(baseDir, "dir2") + os.Mkdir(dataDir, 0700) + mem := CreateMemtable(walFile, dataDir) + + mem.Write("key", "val", "PUT") + actual, err := mem.Get("key") + + if err != nil { + t.Errorf("The read was not able to complete (could mean that the key was not saved into the memtable)") + } + + if record.GetContents(actual).Key != "key" { + t.Errorf("The key that was retrieved was not the same as the key/value pair that was written to the memtable") + } +} + +func TestDeleteForMemtable(t *testing.T) { + baseDir := t.TempDir() + + walDir := filepath.Join(baseDir, "dir1") + os.Mkdir(walDir, 0700) + walFile := filepath.Join(walDir, "mem1.wal") + + dataDir := filepath.Join(baseDir, "dir2") + os.Mkdir(dataDir, 0700) + mem := CreateMemtable(walFile, dataDir) + + mem.Write("key", "val", "PUT") + 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 { + t.Errorf("The delete operation did not successfully increase the memtable size") + } + +} + +func TestWrite_FlushResetsState(t *testing.T) { + // t.Skip("requires sstable wiring and writable data dir — run as integration test") + + baseDir := t.TempDir() + + walDir := filepath.Join(baseDir, "dir1") + os.Mkdir(walDir, 0700) + walFile := filepath.Join(walDir, "mem1.wal") + + dataDir := filepath.Join(baseDir, "dir2") + os.Mkdir(dataDir, 0700) + 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 { + t.Fatalf("unexpected error at write %d: %v", i, err) + } + } + + if mt.size != 0 { + t.Errorf("expected size to reset to 0 after flush, got %d", mt.size) + } +} + +func TestMemtableStartUp(t *testing.T) { + baseDir := t.TempDir() + + walDir := filepath.Join(baseDir, "dir1") + os.Mkdir(walDir, 0700) + walFile := filepath.Join(walDir, "mem1.wal") + + dataDir := filepath.Join(baseDir, "dir2") + os.Mkdir(dataDir, 0700) + mt:= CreateMemtable(walFile, dataDir) + + ok,err := mt.Write("key", "val23", "PUT"); + + if !ok { + 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") + } + + fileInfo,_ := os.Lstat(walFile) + + fmt.Println(fileInfo.Size()) + + mm := CreateMemtable(walFile, dataDir) + _,err = mm.MemtableStartUp() + + if err != nil { + t.Errorf("From trying to startup the memtable:%s\n", err.Error()) + } + + data, err := mm.Get("key") + + if err != nil { + t.Error(err.Error()) + } + + 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" { + t.Errorf("Did not return the valid key/value pair\nKey:%s\nValue:%s", contents.Key, contents.Payload) + } + + for i := range 10 { + content := record.GetContents(data) + + 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/memtable/skiplist.go b/internals/memtable/skiplist.go new file mode 100644 index 0000000..28c3eb8 --- /dev/null +++ b/internals/memtable/skiplist.go @@ -0,0 +1,165 @@ +package memtable + +import ( + "errors" + "math/rand/v2" + + "github.com/Vince-maple-byte/KeyData/internals/record" +) + +const MAX_LEVEL = 32 + +type Node struct { + key string + value []byte + levels []*Node +} + +type Skiplist struct { + head *Node + size int +} + +func CreateSkiplist() *Skiplist { + head := &Node{ + key: "", + value: nil, + levels: make([]*Node, MAX_LEVEL), + } + return &Skiplist{ + head: head, + size: 0, + } +} + +func (list *Skiplist) Search(key string) ([]byte, error) { + curr := list.head + var next *Node + + for i := len(list.head.levels) - 1; i >= 0; i-- { + + for next = curr.levels[i]; next != nil && key > next.key; next = curr.levels[i] { + curr = next + } + if next != nil && next.key == key { + return next.value, nil + } + } + + return nil, errors.New("Element doesn't exist") +} + +// The value is the entire record including all of the information on the timestamp, etc. +func (list *Skiplist) Insert(key string, value []byte) { + var update [MAX_LEVEL]*Node + curr := list.head + + for i := MAX_LEVEL - 1; i >= 0; i-- { + for curr.levels[i] != nil && key > curr.levels[i].key { + curr = curr.levels[i] + } + + update[i] = curr + } + + //This code block checks if the key is already inside of the skiplist + // If the key is already inside then we can check for whichever key is the newest by time and then have that value in the skiplist + next := curr.levels[0] + if next != nil && next.key == key { + newTime := record.GetContents(value).Timestamp.UTC() + prevTime := record.GetContents(next.value).Timestamp.UTC() + + if newTime.After(prevTime) || newTime.Equal(prevTime) { + next.value = value + } + + return + } + + newNode := &Node{ + key: key, + value: value, + levels: make([]*Node, randLevel()), + } + + for i := 0; i < len(newNode.levels); i++ { + newNode.levels[i] = update[i].levels[i] + update[i].levels[i] = newNode + } + + list.size++ +} + +func (list *Skiplist) Delete(key string) ([]byte, error) { + deleteNode, err := list.searchNode(key) + + if err != nil { + return nil, err + } + + update := make([]*Node, len(deleteNode.levels)) + curr := list.head + val := deleteNode.value + + for i := len(update) - 1; i >= 0; i-- { + for curr.levels[i] != nil && key > curr.levels[i].key { + curr = curr.levels[i] + } + + update[i] = curr + } + + for i := range len(update) { + update[i].levels[i] = deleteNode.levels[i] + } + + list.size-- + return val, nil +} + +func (list Skiplist) searchNode(key string) (*Node, error) { + curr := list.head + var next *Node + + for i := len(list.head.levels) - 1; i >= 0; i-- { + + for next = curr.levels[i]; next != nil && key > next.key; next = curr.levels[i] { + curr = next + } + if next != nil && next.key == key { + return next, nil + } + } + + return nil, errors.New("Element doesn't exist") +} + +func (list *Skiplist) EmptyList() { + list.head.key = "" + list.head.value = nil + list.head.levels = make([]*Node, MAX_LEVEL) + + list.size = 0 +} + +func (list *Skiplist) EntireList() [][]byte { + curr := list.head.levels[0] + res := make([][]byte, 0, list.size) + + for range list.size { + res = append(res, curr.value) + curr = curr.levels[0] + } + + return res +} + +func randLevel() int { + var level int = 1 + + for rand.IntN(2) == 1 && level < MAX_LEVEL { + level++ + } + + return level +} diff --git a/internals/memtable/skiplist_test.go b/internals/memtable/skiplist_test.go new file mode 100644 index 0000000..b49d8f5 --- /dev/null +++ b/internals/memtable/skiplist_test.go @@ -0,0 +1,301 @@ +package memtable + +import ( + "testing" + "time" + + "github.com/Vince-maple-byte/KeyData/internals/record" +) + +// --- helpers --- + +func newList(t *testing.T) *Skiplist { + t.Helper() + return CreateSkiplist() +} + +// --- CreateSkiplist --- + +func TestCreateSkiplist_NotNil(t *testing.T) { + list := newList(t) + if list == nil { + t.Fatal("expected non-nil Skiplist") + } +} + +func TestCreateSkiplist_InitialSizeZero(t *testing.T) { + list := newList(t) + if list.size != 0 { + t.Errorf("expected size 0, got %d", list.size) + } +} + +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)) + } +} + +// --- Insert --- + +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) + } +} + +func TestInsert_MultipleElements(t *testing.T) { + list := newList(t) + keys := []string{"banana", "apple", "cherry"} + 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) + } +} + +func TestInsert_UpdateExistingKeyForDifferentTimesItWasInserted(t *testing.T) { + list := newList(t) + old, _ := record.CreateRecord("key", "old", "PUT") + time.Sleep(2 * time.Millisecond) + new, _ := record.CreateRecord("key", "new", "PUT") + list.Insert("key", old) + 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) + } + + val, err := list.Search("key") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if record.GetContents(val).Payload != record.GetContents(new).Payload { + t.Errorf("expected 'new', got %q\nTime for old:%v\nTime for new:%v", + val, record.GetContents(old).Timestamp, record.GetContents(new).Timestamp) + + } +} + +func TestInsert_UpdateExistingKeyWhenInsertedAtTheSameTime(t *testing.T) { + list := newList(t) + old, _ := record.CreateRecord("key", "old", "PUT") + new, _ := record.CreateRecord("key", "new", "PUT") + list.Insert("key", old) + 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) + } + + val, err := list.Search("key") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if record.GetContents(val).Payload != record.GetContents(new).Payload { + t.Errorf("expected 'new', got %q\nTime for old:%v\nTime for new:%v", + val, record.GetContents(old).Timestamp, record.GetContents(new).Timestamp) + + } +} + +func TestInsert_NilValue(t *testing.T) { + list := newList(t) + list.Insert("nilkey", nil) + val, err := list.Search("nilkey") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if val != nil { + t.Errorf("expected nil value, got %v", val) + } +} + +func TestInsert_EmptyStringKey(t *testing.T) { + list := newList(t) + list.Insert("", []byte("empty-key-value")) + val, err := list.Search("") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if string(val) != "empty-key-value" { + t.Errorf("expected 'empty-key-value', got %q", val) + } +} + +func TestInsert_PreservesOrder(t *testing.T) { + list := newList(t) + // Insert out of alphabetical order. + list.Insert("c", []byte("c")) + list.Insert("a", []byte("a")) + list.Insert("b", []byte("b")) + + // Level-0 linked list must be sorted. + curr := list.head.levels[0] + var prev string + for curr != nil { + if curr.key < prev { + t.Errorf("out of order: %q after %q", curr.key, prev) + } + prev = curr.key + curr = curr.levels[0] + } +} + +// --- Search --- + +func TestSearch_ExistingKey(t *testing.T) { + list := newList(t) + list.Insert("hello", []byte("world")) + val, err := list.Search("hello") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if string(val) != "world" { + t.Errorf("expected 'world', got %q", val) + } +} + +func TestSearch_MissingKey(t *testing.T) { + list := newList(t) + _, err := list.Search("ghost") + if err == nil { + t.Error("expected error for missing key, got nil") + } +} + +func TestSearch_EmptyList(t *testing.T) { + list := newList(t) + _, err := list.Search("anything") + if err == nil { + t.Error("expected error on empty list, got nil") + } +} + +func TestSearch_AfterMultipleInserts(t *testing.T) { + list := newList(t) + pairs := map[string]string{ + "foo": "1", + "bar": "2", + "baz": "3", + } + for k, v := range pairs { + list.Insert(k, []byte(v)) + } + for k, want := range pairs { + got, err := list.Search(k) + if err != nil { + t.Errorf("Search(%q): unexpected error: %v", k, err) + continue + } + if string(got) != want { + t.Errorf("Search(%q): expected %q, got %q", k, want, got) + } + } +} + +// --- Delete --- + +func TestDelete_ExistingKey(t *testing.T) { + list := newList(t) + list.Insert("del-me", []byte("value")) + val, err := list.Delete("del-me") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if string(val) != "value" { + t.Errorf("expected returned value 'value', got %q", val) + } +} + +func TestDelete_DecreasesSize(t *testing.T) { + list := newList(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) + } +} + +func TestDelete_MakesKeyUnsearchable(t *testing.T) { + list := newList(t) + list.Insert("gone", []byte("poof")) + list.Delete("gone") + _, err := list.Search("gone") + if err == nil { + t.Error("expected error searching deleted key, got nil") + } +} + +func TestDelete_MissingKey(t *testing.T) { + list := newList(t) + _, err := list.Delete("nope") + if err == nil { + t.Error("expected error deleting missing key, got nil") + } +} + +func TestDelete_OnEmptyList(t *testing.T) { + list := newList(t) + _, err := list.Delete("anything") + if err == nil { + t.Error("expected error deleting from empty list, got nil") + } +} + +func TestDelete_CorrectValueReturned(t *testing.T) { + list := newList(t) + list.Insert("k", []byte("expected-val")) + val, err := list.Delete("k") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if string(val) != "expected-val" { + t.Errorf("expected 'expected-val', got %q", val) + } +} + +// --- EntireList --- + +func TestEntireList_EmptyList(t *testing.T) { + list := newList(t) + res := list.EntireList() + // Result should not be nil; contents may vary by implementation. + if res == nil { + t.Error("expected non-nil slice from EntireList on empty list") + } +} + +func TestEntireList_ReturnsAllValues(t *testing.T) { + list := newList(t) + list.Insert("a", []byte("1")) + list.Insert("b", []byte("2")) + list.Insert("c", []byte("3")) + res := list.EntireList() + if len(res) < 3 { + t.Errorf("expected at least 3 entries, got %d", len(res)) + } +} + +// --- EmptyList --- + +func TestEmptyList_SizeAfterEmpty(t *testing.T) { + list := newList(t) + list.Insert("a", []byte("1")) + list.Insert("b", []byte("2")) + list.EmptyList() + // 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 + if size != 0 { + t.Error("Expected an empty list", size) + } +} diff --git a/internals/record/record.go b/internals/record/record.go index b1eef14..aa6c588 100644 --- a/internals/record/record.go +++ b/internals/record/record.go @@ -1,8 +1,9 @@ -package records +package record import ( "encoding/binary" "errors" + "fmt" "hash/crc32" "time" //"fmt" @@ -13,18 +14,32 @@ import ( // The int value for the checksum is different depending on the data that we provide // As long as the data is the same between two variables they should always have the same checksum. -func checkSum(data string) uint32 { - return crc32.ChecksumIEEE([]byte(data)) + +type Content struct { + Timestamp time.Time + Checksum uint32 + Tombstone uint8 + Keysize uint32 + Payloadsize uint32 + Key string + Payload string } -func ChecksumChecker(data []byte, checksum uint32) bool { - valid := crc32.ChecksumIEEE(data) +func checkSum(key, payload []byte, timestamp uint64) uint32 { + buf := make([]byte, 16+len(key)+len(payload)) - if valid == checksum { - return true - } else { - return false - } + binary.BigEndian.PutUint64(buf[0:8], timestamp) + binary.BigEndian.PutUint32(buf[8:12], uint32(len(key))) + binary.BigEndian.PutUint32(buf[12:16], uint32(len(payload))) + + copy(buf[16:], key) + copy(buf[16+len(key):], payload) + + return crc32.ChecksumIEEE(buf) +} + +func ChecksumChecker(key, payload []byte, timestamp uint64, checksum uint32) bool { + return checkSum(key, payload, timestamp) == checksum } func createTimeStamp() int64 { @@ -40,7 +55,7 @@ func CreateRecord(key, payload, operation string) ([]byte, error) { //Don't need to check for the GET operation since that should automatically go to the ReadFile section. if operation == "" || (operation != "PUT" && operation != "DELETE") { - return nil, errors.New("Invalid Operation") + return nil, errors.New(fmt.Sprintf("Invalid operation: %v", operation)) } //Make sure to use time.AppendBinary https://pkg.go.dev/time#example-Time.AppendBinary @@ -53,7 +68,7 @@ func CreateRecord(key, payload, operation string) ([]byte, error) { //fmt.Println(len(b), time); //Remember to use BigEndian for reverting it back to a uint32 - b = binary.BigEndian.AppendUint32(b, checkSum(payload)) + b = binary.BigEndian.AppendUint32(b, checkSum([]byte(key), []byte(payload), uint64(time))) //We are adding the tombstone value into the record here var Tombstone uint8 @@ -90,3 +105,20 @@ func CreateRecord(key, payload, operation string) ([]byte, error) { return b, nil } + +func GetContents(content []byte) (result Content) { + timeByte := content[:8] + result.Timestamp = time.Unix(0, int64(binary.BigEndian.Uint64(timeByte))) + + result.Checksum = binary.BigEndian.Uint32(content[8:12]) + + result.Tombstone = uint8(content[12]) + + result.Keysize = binary.BigEndian.Uint32(content[13:17]) + result.Payloadsize = binary.BigEndian.Uint32(content[17:21]) + + result.Key = string(content[21 : result.Keysize+21]) + result.Payload = string(content[result.Keysize+21 : (result.Keysize+21)+result.Payloadsize]) + + return +} diff --git a/internals/record/record_test.go b/internals/record/record_test.go index 87e8591..a851d94 100644 --- a/internals/record/record_test.go +++ b/internals/record/record_test.go @@ -1,4 +1,4 @@ -package records_test +package record_test import ( "encoding/binary" @@ -6,7 +6,7 @@ import ( "testing" "time" - records "github.com/Vince-maple-byte/KeyData/internals/record" + "github.com/Vince-maple-byte/KeyData/internals/record" ) func TestRecordTimeStamp(t *testing.T) { @@ -44,7 +44,7 @@ func TestRecordTimeStamp(t *testing.T) { } for _, test := range tests { - record, err := records.CreateRecord(test.key, test.payload, test.operation) + record, err := record.CreateRecord(test.key, test.payload, test.operation) now := time.Now() if err != nil { @@ -77,26 +77,53 @@ func TestRecordCheckSum(t *testing.T) { key: "a", payload: "abc", operation: "PUT", - expected: crc32.ChecksumIEEE([]byte("abc")), + expected: func() uint32 { + data, _ := record.CreateRecord("a", "abc", "PUT") + f := make([]byte, 20) + copy(f[0:8], data[:8]) + binary.BigEndian.PutUint32(f[8:12], uint32(len("a"))) + binary.BigEndian.PutUint32(f[12:16], uint32(len("abc"))) + copy(f[16:17], []byte("a")) + copy(f[17:], []byte("abc")) + return crc32.ChecksumIEEE(f) + }(), }, { testName: "Test 2", key: "a", payload: "", operation: "DELETE", - expected: crc32.ChecksumIEEE([]byte("")), + expected: func() uint32 { + data, _ := record.CreateRecord("a", "", "PUT") + f := make([]byte, 17) + copy(f[0:8], data[:8]) + binary.BigEndian.PutUint32(f[8:12], uint32(len("a"))) + binary.BigEndian.PutUint32(f[12:16], uint32(len(""))) + copy(f[16:17], []byte("a")) + copy(f[17:], []byte("")) + return crc32.ChecksumIEEE(f) + }(), }, { testName: "Test 3", key: "a", payload: "a", operation: "PUT", - expected: crc32.ChecksumIEEE([]byte("a")), + expected: func() uint32 { + data, _ := record.CreateRecord("a", "a", "PUT") + f := make([]byte, 18) + copy(f[0:8], data[:8]) + binary.BigEndian.PutUint32(f[8:12], uint32(len("a"))) + binary.BigEndian.PutUint32(f[12:16], uint32(len("a"))) + copy(f[16:17], []byte("a")) + copy(f[17:], []byte("a")) + return crc32.ChecksumIEEE(f) + }(), }, } for _, test := range tests { - record, err := records.CreateRecord(test.key, test.payload, test.operation) + record, err := record.CreateRecord(test.key, test.payload, test.operation) if err != nil { t.Errorf("Unable to create the record") @@ -106,7 +133,7 @@ func TestRecordCheckSum(t *testing.T) { checksum := binary.BigEndian.Uint32(record[8:12]) if checksum != test.expected { - t.Errorf("incorrect checksum for %v", test.testName) + t.Errorf("incorrect checksum for %v:\nExpected checksum %v\nActual checksum %v", test.testName, test.expected, checksum) } } } @@ -136,7 +163,7 @@ func TestRecordTombstone(t *testing.T) { } for _, test := range tests { - record, err := records.CreateRecord(test.key, test.payload, test.operation) + record, err := record.CreateRecord(test.key, test.payload, test.operation) if err != nil { t.Errorf("Unable to create the record") @@ -176,7 +203,7 @@ func TestRecordKeySize(t *testing.T) { } for _, test := range tests { - record, err := records.CreateRecord(test.key, test.payload, test.operation) + record, err := record.CreateRecord(test.key, test.payload, test.operation) if err != nil { t.Errorf("Unable to create the record") @@ -216,7 +243,7 @@ func TestRecordPayloadSize(t *testing.T) { } for _, test := range tests { - record, err := records.CreateRecord(test.key, test.payload, test.operation) + record, err := record.CreateRecord(test.key, test.payload, test.operation) if err != nil { t.Errorf("Unable to create the record") @@ -256,7 +283,7 @@ func TestRecordKey(t *testing.T) { } for _, test := range tests { - record, err := records.CreateRecord(test.key, test.payload, test.operation) + record, err := record.CreateRecord(test.key, test.payload, test.operation) if err != nil { t.Errorf("Unable to create the record") @@ -297,19 +324,67 @@ func TestRecordPayload(t *testing.T) { } for _, test := range tests { - record, err := records.CreateRecord(test.key, test.payload, test.operation) + records, err := record.CreateRecord(test.key, test.payload, test.operation) if err != nil { t.Errorf("Unable to create the record") } //var parseTime time.Time; - keySize := binary.BigEndian.Uint32(record[13:17]) + keySize := binary.BigEndian.Uint32(records[13:17]) //payloadSize := binary.BigEndian.Uint32(record[17:21]) - payload := string(record[keySize+21:]) + payload := string(records[keySize+21:]) if payload != test.expected { t.Errorf("incorrect checksum for %v", test.testName) } } } + +func TestGetContents(t *testing.T) { + key := "helllpp" + payload := "hhhh" + records, err := record.CreateRecord(key, payload, "PUT") + + if err != nil { + t.Errorf("Unable to create the record") + } + + getContents := record.GetContents(records) + + if time.Unix(0, int64(binary.BigEndian.Uint64(records[:8]))) != getContents.Timestamp { + t.Errorf("Was not able to retrieve the proper timestamp\nExpected:%v\nActual:%v", + time.Unix(0, int64(binary.BigEndian.Uint64(records[:8]))), + getContents.Timestamp) + } + + if getContents.Checksum != binary.BigEndian.Uint32(records[8:12]) { + t.Errorf("Was not able to retrieve the proper checksum\nExpected:%v\nActual:%v", + binary.BigEndian.Uint32(records[8:12]), getContents.Checksum) + } + + if getContents.Tombstone != uint8(records[12]) { + t.Errorf("Was not able to retrieve the proper tombstone\nExpected:%v\nActual:%v", + uint8(records[12]), getContents.Tombstone) + } + + if getContents.Keysize != binary.BigEndian.Uint32(records[13:17]) { + t.Errorf("Was not able to retrieve the proper key size\nExpected:%v\nActual:%v", + binary.BigEndian.Uint32(records[13:17]), getContents.Keysize) + } + + if getContents.Payloadsize != binary.BigEndian.Uint32(records[17:21]) { + t.Errorf("Was not able to retrieve the proper payload size\nExpected:%v\nActual:%v", + binary.BigEndian.Uint32(records[17:21]), getContents.Payloadsize) + } + + if getContents.Key != string(records[21:getContents.Keysize+21]) { + t.Errorf("Was not able to retrieve the proper key\nExpected:%v\nActual:%v", + binary.BigEndian.Uint32(records[21:getContents.Keysize+21]), getContents.Key) + } + + if getContents.Payload != string(records[getContents.Keysize+21:]) { + t.Errorf("Was not able to retrieve the proper payload\nExpected:%v\nActual:%v", + string(records[getContents.Keysize+21:]), getContents.Payload) + } +} diff --git a/internals/sstable/corruption_test.go b/internals/sstable/corruption_test.go new file mode 100644 index 0000000..0fc7e13 --- /dev/null +++ b/internals/sstable/corruption_test.go @@ -0,0 +1,432 @@ +package sstable_test + +import ( + "encoding/binary" + "os" + "path/filepath" + "testing" + + "github.com/Vince-maple-byte/KeyData/internals/memtable" + "github.com/Vince-maple-byte/KeyData/internals/sstable" +) + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +// buildValidRecord constructs a minimal, valid SSTable record byte slice. +// Layout (matches Compact reader): +// +// [0:8] timestamp (uint64 big-endian) +// [8:12] checksum (uint32 big-endian) -- intentionally zeroed for helpers +// [12] flags byte (1 byte, unused but present) +// [13:17] keySize (uint32 big-endian) +// [17:21] payloadSize(uint32 big-endian) +// [21:21+keySize] key +// [21+keySize : ...] payload +func buildValidRecord(key, payload string, timestamp uint64) []byte { + ks := uint32(len(key)) + ps := uint32(len(payload)) + + rec := make([]byte, 0, 21+ks+ps) + rec = binary.BigEndian.AppendUint64(rec, timestamp) // [0:8] timestamp + rec = binary.BigEndian.AppendUint32(rec, 0) // [8:12] checksum placeholder + rec = append(rec, 0x00) // [12] flags + rec = binary.BigEndian.AppendUint32(rec, ks) // [13:17] keySize + rec = binary.BigEndian.AppendUint32(rec, ps) // [17:21] payloadSize + rec = append(rec, []byte(key)...) + rec = append(rec, []byte(payload)...) + return rec +} + +// buildValidFooter builds a 24-byte footer. +// +// [0:8] reserved / bloom-filter offset (zeroed) +// [8:16] end-of-data-block offset +// [16:24] magic number 0xDEADBEEFDEADBEEF +func buildValidFooter(dataBlockSize uint64) []byte { + f := make([]byte, 0, 24) + f = binary.BigEndian.AppendUint64(f, 0) + f = binary.BigEndian.AppendUint64(f, dataBlockSize) + f = binary.BigEndian.AppendUint64(f, 0xDEADBEEFDEADBEEF) + return f +} + +// writeTempFile writes data to a uniquely-named temp file inside dir and +// returns the full path. +func writeTempFile(t *testing.T, dir string, data []byte) string { + t.Helper() + f, err := os.CreateTemp(dir, "kd_*.sst") + if err != nil { + t.Fatalf("writeTempFile: CreateTemp: %v", err) + } + defer f.Close() + if _, err := f.Write(data); err != nil { + t.Fatalf("writeTempFile: Write: %v", err) + } + return f.Name() +} + +// buildSSTFile assembles: records || indexBlock || footer +// and returns the raw bytes. +func buildSSTFile(records [][]byte, indexBlock []byte, footer []byte) []byte { + var out []byte + for _, r := range records { + out = append(out, r...) + } + out = append(out, indexBlock...) + out = append(out, footer...) + return out +} + +// --------------------------------------------------------------------------- +// createFooter tests +// --------------------------------------------------------------------------- + +func TestCreateFooter_EmptyList(t *testing.T) { + _, err := sstable.ExportFooter([][]byte{}) + if err == nil { + t.Fatal("expected error for empty record list, got nil") + } +} + +func TestCreateFooter_MagicNumber(t *testing.T) { + rec := buildValidRecord("hello", "world", 1) + footer, err := sstable.ExportFooter([][]byte{rec}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(footer) != 24 { + t.Fatalf("footer length: want 24, got %d", len(footer)) + } + magic := binary.BigEndian.Uint64(footer[16:24]) + const want = uint64(0xDEADBEEFDEADBEEF) + if magic != want { + t.Errorf("magic: want 0x%X, got 0x%X", want, magic) + } +} + +func TestCreateFooter_DataSizeMatchesInput(t *testing.T) { + r1 := buildValidRecord("key1", "val1", 1) + r2 := buildValidRecord("key2", "val2", 2) + list := [][]byte{r1, r2} + + footer, err := sstable.ExportFooter(list) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + wantSize := uint64(len(r1) + len(r2)) + gotSize := binary.BigEndian.Uint64(footer[8:16]) + if gotSize != wantSize { + t.Errorf("dataBlockSize: want %d, got %d", wantSize, gotSize) + } +} + +// --------------------------------------------------------------------------- +// Truncated-file tests +// (These simulate what Compact would encounter when reading a corrupt file.) +// --------------------------------------------------------------------------- + +func TestReadFile_TruncatedFooter(t *testing.T) { + rec := buildValidRecord("key", "value", 42) + footer := buildValidFooter(uint64(len(rec))) + full := buildSSTFile([][]byte{rec}, []byte{}, footer) + + // Chop the last 4 bytes so the footer is incomplete. + truncated := full[:len(full)-4] + + dir := t.TempDir() + path := writeTempFile(t, dir, truncated) + + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("ReadFile: %v", err) + } + + if len(data) < 24 { + // Correct: caller should detect the file is too short to hold a footer. + t.Log("correctly detected truncated footer: file shorter than 24 bytes") + return + } + + footer2 := data[len(data)-24:] + magic := binary.BigEndian.Uint64(footer2[16:24]) + if magic == 0xDEADBEEFDEADBEEF { + t.Error("truncated file should NOT produce a valid magic number") + } +} + +func TestReadFile_TruncatedMidRecord(t *testing.T) { + // Build a record but only write the first half of it. + rec := buildValidRecord("somekey", "somevalue", 99) + half := rec[:len(rec)/2] + + dir := t.TempDir() + path := writeTempFile(t, dir, half) + + data, _ := os.ReadFile(path) + + // Simulate the Compact inner loop: try to read keySize starting at offset 0. + const headerEnd = 17 + if len(data) < headerEnd { + t.Log("correctly detected truncated record: header incomplete") + return + } + + keySize := int(binary.BigEndian.Uint32(data[13:17])) + payloadSize := int(binary.BigEndian.Uint32(data[17:21])) + expectedEnd := 21 + keySize + payloadSize + + if len(data) < expectedEnd { + t.Logf("correctly detected truncated record: need %d bytes, have %d", expectedEnd, len(data)) + } else { + t.Error("expected truncation to be detected but full record appears readable") + } +} + +func TestReadFile_ZeroByteFile(t *testing.T) { + dir := t.TempDir() + path := writeTempFile(t, dir, []byte{}) + + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("ReadFile: %v", err) + } + if len(data) != 0 { + t.Fatalf("expected 0 bytes, got %d", len(data)) + } + if len(data) < 24 { + t.Log("correctly detected zero-byte file: cannot contain a valid footer") + } +} + +func TestReadFile_FooterOnlyNoData(t *testing.T) { + // A file containing only a footer (dataBlockSize = 0) — edge case. + footer := buildValidFooter(0) + dir := t.TempDir() + path := writeTempFile(t, dir, footer) + + data, _ := os.ReadFile(path) + footer2 := data[len(data)-24:] + magic := binary.BigEndian.Uint64(footer2[16:24]) + + if magic != 0xDEADBEEFDEADBEEF { + t.Errorf("magic mismatch: got 0x%X", magic) + } + + dataEnd := binary.BigEndian.Uint64(footer2[8:16]) + if dataEnd != 0 { + t.Errorf("expected dataBlockSize=0, got %d", dataEnd) + } +} + +func TestReadFile_TruncatedAtHeaderBoundary(t *testing.T) { + // Write exactly 21 bytes (header only, no key/payload). + header := make([]byte, 21) + binary.BigEndian.PutUint64(header[0:8], 12345) // timestamp + binary.BigEndian.PutUint32(header[8:12], 0) // checksum + header[12] = 0x00 // flags + binary.BigEndian.PutUint32(header[13:17], 5) // keySize = 5 + binary.BigEndian.PutUint32(header[17:21], 10) // payloadSize = 10 + + dir := t.TempDir() + path := writeTempFile(t, dir, header) + data, _ := os.ReadFile(path) + + keySize := int(binary.BigEndian.Uint32(data[13:17])) + payloadSize := int(binary.BigEndian.Uint32(data[17:21])) + need := 21 + keySize + payloadSize + + if len(data) < need { + t.Logf("correctly detected record truncated at header boundary: need %d, have %d", need, len(data)) + } else { + t.Error("expected boundary truncation to be detected") + } +} + +// --------------------------------------------------------------------------- +// Swapped-bytes / bit-flip tests +// --------------------------------------------------------------------------- + +func TestReadFile_SwappedMagicBytes(t *testing.T) { + rec := buildValidRecord("key", "val", 1) + footer := buildValidFooter(uint64(len(rec))) + + // Swap the first two bytes of the magic number region (footer[16] <-> footer[17]). + footer[16], footer[17] = footer[17], footer[16] + + full := buildSSTFile([][]byte{rec}, []byte{}, footer) + dir := t.TempDir() + path := writeTempFile(t, dir, full) + + data, _ := os.ReadFile(path) + magic := binary.BigEndian.Uint64(data[len(data)-8:]) + + if magic == 0xDEADBEEFDEADBEEF { + t.Error("swapped magic bytes should invalidate the magic number") + } else { + t.Logf("correctly detected swapped magic bytes: got 0x%X", magic) + } +} + +func TestReadFile_SwappedKeySizeBytes(t *testing.T) { + rec := buildValidRecord("hello", "world", 7) + + // Swap bytes [13] and [14] — the high bytes of keySize (uint32). + rec[13], rec[16] = rec[16], rec[13] + + dir := t.TempDir() + path := writeTempFile(t, dir, []byte{}) + os.WriteFile(path, rec, 0644) + + data, _ := os.ReadFile(path) + keySize := int(binary.BigEndian.Uint32(data[13:17])) + + // Original keySize was 5 ("hello"); swapped bytes should differ. + if keySize == 5 { + t.Error("swapped keySize bytes should produce a different (wrong) keySize") + } else { + t.Logf("correctly detected swapped keySize bytes: parsed keySize=%d (true=5)", keySize) + } +} + +func TestReadFile_SwappedTimestampBytes(t *testing.T) { + const originalTS uint64 = 0x0000000100000002 + rec := buildValidRecord("k", "v", originalTS) + + // Swap bytes [0] and [7] — MSB and LSB of the timestamp. + rec[0], rec[7] = rec[7], rec[0] + + ts := binary.BigEndian.Uint64(rec[0:8]) + if ts == originalTS { + t.Error("swapped timestamp bytes should produce a different value") + } else { + t.Logf("correctly detected swapped timestamp: got 0x%X (original 0x%X)", ts, originalTS) + } +} + +func TestReadFile_SwappedDataBlockSizeInFooter(t *testing.T) { + rec := buildValidRecord("testkey", "testval", 55) + correctSize := uint64(len(rec)) + footer := buildValidFooter(correctSize) + + // Swap bytes [8] and [9] inside the footer (high bytes of dataBlockSize). + footer[8], footer[15] = footer[15], footer[8] + + full := buildSSTFile([][]byte{rec}, []byte{}, footer) + dir := t.TempDir() + path := writeTempFile(t, dir, full) + + data, _ := os.ReadFile(path) + footerSlice := data[len(data)-24:] + parsedSize := binary.BigEndian.Uint64(footerSlice[8:16]) + + if parsedSize == correctSize { + t.Logf("incorrectly detected swapped dataBlockSize: got %d (correct %d)", parsedSize, correctSize) + t.Error("swapped dataBlockSize bytes should produce a wrong size") + } else { + t.Logf("correctly detected swapped dataBlockSize: got %d (correct %d)", parsedSize, correctSize) + } +} + +func TestReadFile_BitFlipInPayload(t *testing.T) { + rec := buildValidRecord("mykey", "myvalue", 100) + payloadStart := 21 + len("mykey") + + // Flip all bits in the first payload byte. + rec[payloadStart] ^= 0xFF + + payload := rec[payloadStart : payloadStart+len("myvalue")] + if string(payload) == "myvalue" { + t.Error("bit-flipped payload should not equal original") + } else { + t.Logf("correctly detected payload corruption: %q", payload) + } +} + +// --------------------------------------------------------------------------- +// fileOffset tests +// --------------------------------------------------------------------------- + +func TestFileOffset_CorrectOffsets(t *testing.T) { + r1 := []byte("abcde") // 5 bytes → offset 0 + r2 := []byte("fghij") // 5 bytes → offset 5 + r3 := []byte("klmnopqr") // 8 bytes → offset 10 + + offsets := sstable.ExportFileOffset([][]byte{r1, r2, r3}) + + expected := []uint64{0, 5, 10} + for i, want := range expected { + if offsets[i] != want { + t.Errorf("offset[%d]: want %d, got %d", i, want, offsets[i]) + } + } +} + +func TestFileOffset_EmptyList(t *testing.T) { + offsets := sstable.ExportFileOffset([][]byte{}) + if len(offsets) != 0 { + t.Errorf("expected empty offsets, got %v", offsets) + } +} + +// --------------------------------------------------------------------------- +// Compact-level integration: corrupt file in temp dir +// --------------------------------------------------------------------------- + +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) + } + + writeValidSST("kd_1.sst", "apple", "pie") + writeValidSST("kd_2.sst", "banana", "split") + writeValidSST("kd_3.sst", "cherry", "jam") + + // Fourth file: valid header, truncated body. + 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) + + // 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. + t.Logf("Compact returned: %v", err) +} + +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))) + if corruptMagic { + // Swap the last two bytes of the magic number. + footer[22], footer[23] = footer[23], footer[22] + } + data := buildSSTFile([][]byte{rec}, []byte{}, footer) + os.WriteFile(filepath.Join(dir, name), data, 0644) + } + + 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 + + err := sstable.Compact(dir) + t.Logf("Compact with swapped magic returned: %v", err) +} diff --git a/internals/sstable/filepath.go b/internals/sstable/filepath.go new file mode 100644 index 0000000..2e989c5 --- /dev/null +++ b/internals/sstable/filepath.go @@ -0,0 +1,9 @@ +package sstable + +type FilePath struct { + Path string +} + +func NewPath(path string) *FilePath { + return &FilePath{Path: path} +} diff --git a/internals/sstable/merger.go b/internals/sstable/merger.go new file mode 100644 index 0000000..4e3a751 --- /dev/null +++ b/internals/sstable/merger.go @@ -0,0 +1,8 @@ +package sstable + +type ListMerger interface { + Insert(key string, value []byte) + EntireList() [][]byte +} + +var MergeList func() ListMerger diff --git a/internals/sstable/reader.go b/internals/sstable/reader.go new file mode 100644 index 0000000..1cde974 --- /dev/null +++ b/internals/sstable/reader.go @@ -0,0 +1,171 @@ +package sstable + +import ( + "encoding/binary" + "errors" + "os" + "path/filepath" + "sort" + "strconv" + "strings" + + "github.com/Vince-maple-byte/KeyData/internals/record" +) + +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: +// We check if the key inside of the byte offset is greater than or equal to our key +// If it is greater: we go to the previous byte offset inside of the index and search between the previous byte offset +// and the byte offset were it is greater than the key. +// 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) { + file, err := os.Open(filePath) + + defer file.Close() + if err != nil { + return nil, err + } + fileInfo, _ := file.Stat() + footer := make([]byte, 24) + + file.ReadAt(footer, fileInfo.Size()-24) + + //Checking the magic number first + if binary.BigEndian.Uint64(footer[16:]) != uint64(0xDEADBEEFDEADBEEF) { + err = errors.New("Incorrect magic number inside of the file. The file is invalid") + return nil, err + } + + //Getting the index block offset + indexBlockLoc := binary.BigEndian.Uint64(footer[8:16]) + lowKeyOffset := uint64(0) + highKeyOffset := uint64(0) + + for i := indexBlockLoc; i <= uint64(fileInfo.Size()-24); { + keySize := make([]byte, 4) + file.ReadAt(keySize, int64(i)) + offsetKey := make([]byte, int64(binary.BigEndian.Uint32(keySize))) + file.ReadAt(offsetKey, int64(i)+4) + //This gives us the location of the offset where it is saved in the data portion of the file + offsetLoc := make([]byte, 8) + file.ReadAt(offsetLoc, int64(i+4+uint64(binary.BigEndian.Uint32(keySize)))) + keyOffset := binary.BigEndian.Uint64(offsetLoc) + + //Go to the next location in the index block + i = i + 4 + uint64(binary.BigEndian.Uint32(keySize)) + 8 + + if key == string(offsetKey) { + curr := make([]byte, 21) + file.ReadAt(curr, int64(keyOffset)) + payloadSize := binary.BigEndian.Uint32(curr[17:21]) + + entireRecord := make([]byte, 21+binary.BigEndian.Uint32(keySize)+payloadSize) + file.ReadAt(entireRecord, int64(keyOffset)) + return entireRecord, nil + } + + lowKeyOffset = highKeyOffset + highKeyOffset = keyOffset + + if key < string(offsetKey) { + break + } + } + + 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; { + curr := make([]byte, 21) + file.ReadAt(curr, int64(i)) + keySize := binary.BigEndian.Uint32(curr[13:17]) + payloadSize := binary.BigEndian.Uint32(curr[17:21]) + + entireRecord := make([]byte, 21+keySize+payloadSize) + file.ReadAt(entireRecord, int64(i)) + + currRecord := record.GetContents(entireRecord) + + if !record.ChecksumChecker( + []byte(currRecord.Key), + []byte(currRecord.Payload), + uint64(currRecord.Timestamp.UnixNano()), + currRecord.Checksum) { + return nil, errors.New("This section of the file is corrupted, can not retrieve the key/value pairs from here") + } + + if currRecord.Key == key { + return entireRecord, nil + } + + if currRecord.Key > key { + break + } + + i += 21 + uint64(keySize) + uint64(payloadSize) + } + + return nil, errors.New("Key could not be found") +} + +func ReadFromAllFiles(key string, dir string) ([]byte, error) { + fileDir, err := os.ReadDir(dir) + + var files []SSTFile + + if err != nil { + return nil, err + } + + for _, file := range fileDir { + gen, err := parseGeneration(filepath.Join("../data", 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 + }) + + for _, file := range files { + res, err := ReadFromFile(file.FileName, key) + + if err == nil { + contents := record.GetContents(res) + + if contents.Tombstone != 0 { + return nil, errors.New("Key/Value pair doesn't exist") + } + + return res, nil + } + } + + 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 new file mode 100644 index 0000000..60c5cb0 --- /dev/null +++ b/internals/sstable/reader_test.go @@ -0,0 +1,175 @@ +package sstable_test + +import ( + "os" + "path/filepath" + "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/sstable" +) + +func TestReadFile(t *testing.T) { + tests := []struct { + testName string + data [][]byte + key string + expected string + }{ + { + testName: "key_size_1", + data: func() [][]byte { + d := memtable.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: "8", + expected: "9", + }, + { + testName: "key_size_2", + data: func() [][]byte { + d := memtable.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: "31", + expected: "32", + }, + { + testName: "key_size_3", + data: func() [][]byte { + d := memtable.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: "432", + expected: "433", + }, + { + testName: "key_size_4", + data: func() [][]byte { + d := memtable.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: "3197", + expected: "3198", + }, + { + testName: "key_is_a_direct_match_to_the_index", + data: func() [][]byte { + d := memtable.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: "1140", + expected: "1141", + }, + { + testName: "key_if_file_truncated", + data: func() [][]byte { + d := memtable.CreateSkiplist() + for i := range 240 { + + r, _ := record.CreateRecord(strconv.Itoa(i), strconv.Itoa(i+1), "PUT") + d.Insert(strconv.Itoa(i), r) + } + return d.EntireList() + }(), + key: "104", + expected: "105", + }, + { + testName: "key_if_larger_than_normal", + data: func() [][]byte { + d := memtable.CreateSkiplist() + for i := range 7000 { + + r, _ := record.CreateRecord(strconv.Itoa(i), strconv.Itoa(i+1), "PUT") + d.Insert(strconv.Itoa(i), r) + } + return d.EntireList() + }(), + key: "4006", + expected: "4007", + }, + } + + for _, test := range tests { + t.Run(test.testName, func(t *testing.T) { + dir := t.TempDir() + _, err := sstable.WriteToFile(test.data, dir) + + if err != nil { + t.Fatalf("Error in writing the file: %s", err.Error()) + } + 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) + + if rec == nil || err != nil { + t.Fatalf("Error received: %v\nSize of the file: %d", err.Error(), size) + } + if record.GetContents(rec).Payload != test.expected { + t.Errorf("Test failed for %s\nExpected:%v\nActual:%v", + test.testName, test.expected, record.GetContents(rec).Payload) + } + + //tearDown("../test") + }) + + } +} + +func TestReadFromAllFiles(t *testing.T) { + + dir := t.TempDir() + for i := range 10 { + d := memtable.CreateSkiplist() + for i := range 3000 * (i + 1) { + + r, _ := record.CreateRecord(strconv.Itoa(i), strconv.Itoa(i+1), "PUT") + d.Insert(strconv.Itoa(i), r) + } + _, err := sstable.WriteToFile(d.EntireList(), dir) + + if err != nil { + t.Fatalf("Error in writing the file: %s", err.Error()) + } + } + + _, err := sstable.ReadFromAllFiles("29994", dir) + + if err != nil { + t.Errorf("Was not able to find the valid record") + } + + //tearDown("../test") +} diff --git a/internals/sstable/writer.go b/internals/sstable/writer.go new file mode 100644 index 0000000..49e622f --- /dev/null +++ b/internals/sstable/writer.go @@ -0,0 +1,328 @@ +package sstable + +import ( + "encoding/binary" + "errors" + "fmt" + "io/fs" + "os" + "path/filepath" + "slices" + "strconv" + "strings" + + "github.com/Vince-maple-byte/KeyData/internals/record" +) + +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: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) { + filename := "" + files, err := os.ReadDir(filePath) + + if err != nil { + return false, err + } + + 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] + idx, err := strconv.Atoi(str) + if err == nil && idx > maxIndex { + maxIndex = idx + } + } + filename = "kd_" + strconv.Itoa(maxIndex+1) + ".sst" + } + + file, err := os.Create(filepath.Join(filePath, filename)) + + + if err != nil { + return false, err + } + + defer file.Close() + + offset := fileOffset(list) + index := createIndexBlock(list, offset) + footer, err := createFooter(list) + + if err != nil { + return false, err + } + + list = append(list, index) + list = append(list, footer) + + content := slices.Concat(list...) + + _, err = file.Write(content) + + if err != nil { + return false, err + } + + file.Sync() + + return true, nil +} + +//Create the indexing block for the file +/* + The indexing block is responsible for allowing us to perform reads much more efficiently. + + How is it done: + The skiplist would be called to write into the file after 3200 entries are filled. + When this is done, we can categorize the file into 20 sections of 160 elements. + + In the indexing block, each entry would represent a specific block. + The entry will contain the key and the byte offset of the lowest key value in the block. + So it would look like this for example + keyA=0,keyB=202,keyC=403, etc. + + Since all of the records in the file are organized in sorted order, we can use the block to traverse through the file much quicker + + For example, if we have a key called N, we would start by comparing the key of the 20th block, + if the key is greater than or equal to the key N we would start from the byte offset of that block, + else we keep on going backwards by 1 block until we find a block where the key is greater than or equal to the block. + + Only issue with this method. In the case where the key is not located in the file, we would still need to do this traversal, + Solution using bloom filters. + + Format: + Size of index block: uint64 + Each entry will follow like this: + KeySize: uint32 bytes long; + Key: n bytes long (string) (n for keysize) + Byte offset: uint64 +*/ + +func createIndexBlock(contentList [][]byte, offset []uint64) []byte { + index := make([]byte, 0, INDEX_BLOCK*160) + + //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]) + + index = binary.BigEndian.AppendUint32(index, contents.Keysize) + + index = append(index, contents.Key...) + + index = binary.BigEndian.AppendUint64(index, offset[len(offset)/2]) + } else { + for i := 0; i < len(contentList); i = i + (len(contentList) / INDEX_BLOCK) { + contents := record.GetContents(contentList[i]) + + index = binary.BigEndian.AppendUint32(index, contents.Keysize) + + index = append(index, contents.Key...) + + index = binary.BigEndian.AppendUint64(index, offset[i]) + } + } + + //var size uint64 = uint64(len(index)) + //index = append(binary.BigEndian.AppendUint64([]byte{}, size), index...) + + return index +} + +func fileOffset(contentList [][]byte) []uint64 { + var offsetTracker uint64 = 0 + offset := make([]uint64, len(contentList)) + + for i, v := range contentList { + recordsize := len(v) + + offset[i] = offsetTracker + + offsetTracker += uint64(recordsize) + } + + return offset +} + +func createFooter(list [][]byte) ([]byte, error) { + if len(list) <= 0 { + return nil, errors.New("Not able to create the footer because the record list is too small") + } + footer := make([]byte, 0, 24) + footer = binary.BigEndian.AppendUint64(footer, 0) + var size uint64 + + for _, rec := range list { + size += uint64(len(rec)) + } + footer = binary.BigEndian.AppendUint64(footer, size) + footer = binary.BigEndian.AppendUint64(footer, uint64(0xDEADBEEFDEADBEEF)) + + 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 := MergeList() + for _, fileInfo := range val { + + 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 { + os.Remove(filepath.Join(filePath, fileInfo.Name())) + 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 := 0; i < int(fileBlockEnds); { + //header := fileData[i : i+21] + 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])) + + key := fileData[i+21 : i+keySize+21] + payload := fileData[i+keySize+21 : i+(keySize+21)+payloadSize] + + //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+(keySize+21)+payloadSize]) + i = i + (keySize + 21) + payloadSize + + } + } + + //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) +} + +func ExportFileOffset(contentList [][]byte) []uint64 { + return fileOffset(contentList) +} diff --git a/internals/sstable/writer_test.go b/internals/sstable/writer_test.go new file mode 100644 index 0000000..e0433bc --- /dev/null +++ b/internals/sstable/writer_test.go @@ -0,0 +1,241 @@ +package sstable_test + +import ( + "encoding/binary" + "fmt" + "os" + "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" +) + +func startUp(filePath string, data ...string) (int, error) { + + 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, filePath) + + if err != nil { + return 0, err + } + size++ + + } + + return size, 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 + data []string + }{ + { + testName: "Same size for each file", + data: []string{"data", "data", "data", "data", "data", "data", "data", "data", "data", "data"}, + }, + { + testName: "Size increasing for each file", + data: []string{"data1", "data11", "data111", "data1111", "data11111", + "data111111", "data1111111", "data11111111", "data111111111", "data1111111111"}, + }, + { + testName: "Size decreasing for each file", + data: []string{"data1111111111", "data111111111", "data11111111", "data1111111", "data111111", + "data11111", "data1111", "data111", "data11", "data1"}, + }, + { + testName: "Random order of size for each file", + data: []string{"data111", "data1111", "data1", "data1111111", "data111111111", + "data11111", "data1111", "data1111111111", "data11", "data11111111"}, + }, + } + + for _, test := range tests { + filePath := t.TempDir() + totalSize, err := startUp(filePath, test.data...) + + if err != nil || totalSize == 0 { + t.Fatalf("Could not start up the test") + } + + buckets := sstable.ExportBuckets(filePath) + average := totalSize / 10 + + for key, bucket := range buckets { + + if len(bucket) != 0 { + t.Logf("This bucket %v is %d length\n", key, len(bucket)) + + for _, item := range bucket { + 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()) + } + } + } + } + } + + //tearDown("../test") +} + +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 + 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) + + t.Logf("Size: %d", len(file)) + + if err != nil || len(file) == 0 { + t.Fatalf("Not to able properly make the file: size=%d", len(file)) + } + + } + + 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) + + if err != nil { + t.Fatalf("Not able to access the directory for testing: %v", dir) + } + + if len(files) != 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) { + fileContents := make([][]byte, 0, 10) + dir := t.TempDir() + + for range 1 { + r, _ := record.CreateRecord("a", "1", "PUT") + fileContents = append(fileContents, r) + } + + ok, err := sstable.WriteToFile(fileContents, dir) + + if err != nil { + t.Errorf("Error encountered: %v\n", err) + } + + if !ok { + t.Errorf("Not able to create the file") + } + + file, _ := os.Open(filepath.Join(dir, "kd_1.sst")) + defer file.Close() + info, _ := file.Stat() + + if info.Size() == 0 { + t.Error("Did not populate file") + } + + //tearDown("../test") +} + +func TestFooter(t *testing.T) { + fileContents := make([][]byte, 0) + dir := t.TempDir() + + for i := range 3200 { + c, _ := record.CreateRecord(fmt.Sprintf("d%v", i), fmt.Sprintf("d%v", i), "PUT") + fileContents = append(fileContents, c) + } + + _, err := sstable.WriteToFile(fileContents, dir) + + if err != nil { + t.Fatal(err.Error()) + } + + file, err := os.Open(filepath.Join(dir, "kd_1.sst")) + defer file.Close() + + if err != nil { + t.Fatal(err.Error()) + } + + footer := make([]byte, 24) + fileInfo, _ := file.Stat() + size := fileInfo.Size() + _, err = file.ReadAt(footer, int64(size-24)) + + if err != nil { + t.Fatal(err.Error()) + } + + if binary.BigEndian.Uint64(footer[:8]) != uint64(0) { + t.Errorf("Footer saved incorrect offset for the start of the file:%d", binary.BigEndian.Uint64(footer[:8])) + } + + indexBlockLoc := make([]byte, 8) + file.ReadAt(indexBlockLoc, int64(binary.BigEndian.Uint64(footer[8:16]))) + keySize := make([]byte, 4) + + file.ReadAt(keySize, int64(binary.BigEndian.Uint64(footer[8:16]))) + key := make([]byte, binary.BigEndian.Uint32(keySize)) + file.ReadAt(key, int64(binary.BigEndian.Uint64(footer[8:16]))+4) + offset := make([]byte, 8) + file.ReadAt(offset, int64(binary.BigEndian.Uint64(footer[8:16])+4+uint64(binary.BigEndian.Uint32(keySize)))) + //keyOffset := binary.BigEndian.Uint64(offset) + + //keySize := binary.BigEndian.Uint64(); + fmt.Printf("Footer bytes: %x\n", footer) + t.Logf("key size=%d", binary.BigEndian.Uint32(keySize)) + t.Logf("Index Block Loc=%v", binary.BigEndian.Uint64(footer[8:16])) + if binary.BigEndian.Uint64(offset) != 0 { + t.Errorf("Footer saved incorrect offset for the start of the index block:%d\nkey=%v", + binary.BigEndian.Uint64(offset), string(key)) + } + if binary.BigEndian.Uint64(footer[16:24]) != uint64(0xDEADBEEFDEADBEEF) { + t.Errorf("Footer saved incorrect magic number: %v", binary.BigEndian.Uint64(footer[16:24])) + } + +} + +func TestCorrectKeyValue(t *testing.T) { + +} diff --git a/main.go b/main.go index 99d8d48..27ef79f 100644 --- a/main.go +++ b/main.go @@ -2,20 +2,18 @@ package main import ( "fmt" - - "github.com/Vince-maple-byte/KeyData/internals/file" - //"time" - // "github.com/Vince-maple-byte/KeyData/internals/file" - //"github.com/Vince-maple-byte/KeyData/internals/record" - // "unsafe" + "log" + "net" + + "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" ) /* -TODO: - -Make the Write method - How the structure of the timestamp would look like Timestamp | CRC32 error checksum| Tombstone (It's one byte long; basically 0 and 1 to determine whether this is a deleted key or not) | Key Size | Payload(Value) Size | Key Value | Payload @@ -24,9 +22,9 @@ time stamp: 8 bytes CRC32: 4 bytes Tombstone: 1 byte key size: 4 bytes -payload size: 4 bytes +payload size: 4 bytes -> The header of the record is 21 bytes long key: (key size) bytes -payload: (payload size) bytes +payload: (payload size) bytes -> The rest of the record is 21 + keySize + payloadSize bytes long Have it so that in the Write method after we are done adding the record into the file, make sure to add flush(), we update the index map with the key and byte offset @@ -37,97 +35,46 @@ Make some unit tests to test the methods From there we can move on to SSTable, compaction, LSM Tables, and finally bloom tables +TODO: Fix the filePath for the project so that it doesn't have to be hardcoded. +./internals/data works for the production while ../data works for unit tests */ func main() { - var database string - fmt.Println("Welcome to KeyData") - fmt.Print("Enter your database name to get started\n") - - _, err := fmt.Scan(&database) - - for err != nil { - fmt.Println("Please enter a valid database name") - _, err = fmt.Scan(&database) + //We need to use this interface here to avoid an import cycle between sstable and memtable + sstable.MergeList = func() sstable.ListMerger { + return memtable.CreateSkiplist() } - database += ".log" + walPath := "./internals/wal/mem1.wal" + dataDir := "./internals/data" + mem := memtable.CreateMemtable(walPath, dataDir) - f, errf := file.OpenFile(database) - defer f.File.Close() + mem.MemtableStartUp() - if errf != nil { - panic(errf) + server := &network.Server{ + Memtable: mem, + DataDir: dataDir, } - fmt.Println("Database", database, "opened successfully") - willContinue := true - - for willContinue { - fmt.Println("Enter an operation: PUT, DELETE, GET") - - var operation string - _, err := fmt.Scan(&operation) - - if err != nil { - break - } - - if operation == "PUT" { - fmt.Println("Enter a key") - var key string - fmt.Scan(&key) - fmt.Println("Enter the data that you want to save") - var payload string - fmt.Scan(&payload) + port := 5773 - numAdded, _ := f.PutContents(key, payload, "PUT") + lis, err := net.Listen("tcp", fmt.Sprintf("localhost:%d", port)) - if numAdded > -1 { - fmt.Println("Was able to successfully add the key/value pair into the database") - } else { - fmt.Println("Was not able to successfully add the key/value pair into the database") - } - - } - if operation == "DELETE" { - fmt.Println("Enter a key") - var key string - fmt.Scan(&key) - - numAdded, _ := f.PutContents(key, "", "DELETE") - - if numAdded > -1 { - fmt.Println("Was able to successfully delete the key/value pair into the database") - } else { - fmt.Println("Was not able to successfully delete the key/value pair into the database") - } - - } - if operation == "GET" { - fmt.Println("Enter a key") - var key string - fmt.Scan(&key) - - deleted, payload, timestamp, err := f.GetContents(key) - - if err != nil { - fmt.Println("Was able not able to retrieve the file contents for", key) - } else { - fmt.Printf("For key %v:\nThe payload: %v\nDeleted: %v\nThe timestamp: %v\n", key, payload, deleted, timestamp) - } - - } - operation = "" + if err != nil { + log.Fatalf("Failed to listen: %v", err) + } - fmt.Println("Do you want to continue?(Y/N)") - var choice string - fmt.Scan(&choice) + var opts []grpc.ServerOption - if choice == "N" || choice == "n" { - break - } + grpcServer := grpc.NewServer(opts...) + //network.pb.RegisterDataServer(grpcServer, newServer()) + network.RegisterDataServer(grpcServer, server) + reflection.Register(grpcServer) + err = grpcServer.Serve(lis) + if err != nil { + log.Fatalf("Failed to serve: %v", err) } + fmt.Printf("logging in port: %d", port) } diff --git a/network/data.pb.go b/network/data.pb.go new file mode 100644 index 0000000..ab59dfc --- /dev/null +++ b/network/data.pb.go @@ -0,0 +1,311 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.11 +// protoc v7.35.0 +// source: network/data.proto + +package network + +import ( + reflect "reflect" + sync "sync" + unsafe "unsafe" + + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + timestamppb "google.golang.org/protobuf/types/known/timestamppb" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +type CreateRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + Operation string `protobuf:"bytes,1,opt,name=operation,proto3" json:"operation,omitempty"` + Key string `protobuf:"bytes,2,opt,name=key,proto3" json:"key,omitempty"` + Payload string `protobuf:"bytes,3,opt,name=payload,proto3" json:"payload,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *CreateRequest) Reset() { + *x = CreateRequest{} + mi := &file_network_data_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *CreateRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*CreateRequest) ProtoMessage() {} + +func (x *CreateRequest) ProtoReflect() protoreflect.Message { + mi := &file_network_data_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use CreateRequest.ProtoReflect.Descriptor instead. +func (*CreateRequest) Descriptor() ([]byte, []int) { + return file_network_data_proto_rawDescGZIP(), []int{0} +} + +func (x *CreateRequest) GetOperation() string { + if x != nil { + return x.Operation + } + return "" +} + +func (x *CreateRequest) GetKey() string { + if x != nil { + return x.Key + } + return "" +} + +func (x *CreateRequest) GetPayload() string { + if x != nil { + return x.Payload + } + return "" +} + +type CreateResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + Success bool `protobuf:"varint,1,opt,name=success,proto3" json:"success,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *CreateResponse) Reset() { + *x = CreateResponse{} + mi := &file_network_data_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *CreateResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*CreateResponse) ProtoMessage() {} + +func (x *CreateResponse) ProtoReflect() protoreflect.Message { + mi := &file_network_data_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use CreateResponse.ProtoReflect.Descriptor instead. +func (*CreateResponse) Descriptor() ([]byte, []int) { + return file_network_data_proto_rawDescGZIP(), []int{1} +} + +func (x *CreateResponse) GetSuccess() bool { + if x != nil { + return x.Success + } + return false +} + +type SearchRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + Key string `protobuf:"bytes,2,opt,name=key,proto3" json:"key,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *SearchRequest) Reset() { + *x = SearchRequest{} + mi := &file_network_data_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *SearchRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*SearchRequest) ProtoMessage() {} + +func (x *SearchRequest) ProtoReflect() protoreflect.Message { + mi := &file_network_data_proto_msgTypes[2] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use SearchRequest.ProtoReflect.Descriptor instead. +func (*SearchRequest) Descriptor() ([]byte, []int) { + return file_network_data_proto_rawDescGZIP(), []int{2} +} + +func (x *SearchRequest) GetKey() string { + if x != nil { + return x.Key + } + return "" +} + +type SearchResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + CreatedAt *timestamppb.Timestamp `protobuf:"bytes,1,opt,name=created_at,json=createdAt,proto3" json:"created_at,omitempty"` + Key string `protobuf:"bytes,2,opt,name=key,proto3" json:"key,omitempty"` + Payload string `protobuf:"bytes,3,opt,name=payload,proto3" json:"payload,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *SearchResponse) Reset() { + *x = SearchResponse{} + mi := &file_network_data_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *SearchResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*SearchResponse) ProtoMessage() {} + +func (x *SearchResponse) ProtoReflect() protoreflect.Message { + mi := &file_network_data_proto_msgTypes[3] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use SearchResponse.ProtoReflect.Descriptor instead. +func (*SearchResponse) Descriptor() ([]byte, []int) { + return file_network_data_proto_rawDescGZIP(), []int{3} +} + +func (x *SearchResponse) GetCreatedAt() *timestamppb.Timestamp { + if x != nil { + return x.CreatedAt + } + return nil +} + +func (x *SearchResponse) GetKey() string { + if x != nil { + return x.Key + } + return "" +} + +func (x *SearchResponse) GetPayload() string { + if x != nil { + return x.Payload + } + return "" +} + +var File_network_data_proto protoreflect.FileDescriptor + +const file_network_data_proto_rawDesc = "" + + "\n" + + "\x12network/data.proto\x1a\x1fgoogle/protobuf/timestamp.proto\"Y\n" + + "\rCreateRequest\x12\x1c\n" + + "\toperation\x18\x01 \x01(\tR\toperation\x12\x10\n" + + "\x03key\x18\x02 \x01(\tR\x03key\x12\x18\n" + + "\apayload\x18\x03 \x01(\tR\apayload\"*\n" + + "\x0eCreateResponse\x12\x18\n" + + "\asuccess\x18\x01 \x01(\bR\asuccess\"!\n" + + "\rSearchRequest\x12\x10\n" + + "\x03key\x18\x02 \x01(\tR\x03key\"w\n" + + "\x0eSearchResponse\x129\n" + + "\n" + + "created_at\x18\x01 \x01(\v2\x1a.google.protobuf.TimestampR\tcreatedAt\x12\x10\n" + + "\x03key\x18\x02 \x01(\tR\x03key\x12\x18\n" + + "\apayload\x18\x03 \x01(\tR\apayload2`\n" + + "\x04Data\x12+\n" + + "\x06Search\x12\x0e.SearchRequest\x1a\x0f.SearchResponse\"\x00\x12+\n" + + "\x06Create\x12\x0e.CreateRequest\x1a\x0f.CreateResponse\"\x00B-Z+github.com/Vince-maple-byte/KeyData/networkb\x06proto3" + +var ( + file_network_data_proto_rawDescOnce sync.Once + file_network_data_proto_rawDescData []byte +) + +func file_network_data_proto_rawDescGZIP() []byte { + file_network_data_proto_rawDescOnce.Do(func() { + file_network_data_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_network_data_proto_rawDesc), len(file_network_data_proto_rawDesc))) + }) + return file_network_data_proto_rawDescData +} + +var file_network_data_proto_msgTypes = make([]protoimpl.MessageInfo, 4) +var file_network_data_proto_goTypes = []any{ + (*CreateRequest)(nil), // 0: CreateRequest + (*CreateResponse)(nil), // 1: CreateResponse + (*SearchRequest)(nil), // 2: SearchRequest + (*SearchResponse)(nil), // 3: SearchResponse + (*timestamppb.Timestamp)(nil), // 4: google.protobuf.Timestamp +} +var file_network_data_proto_depIdxs = []int32{ + 4, // 0: SearchResponse.created_at:type_name -> google.protobuf.Timestamp + 2, // 1: Data.Search:input_type -> SearchRequest + 0, // 2: Data.Create:input_type -> CreateRequest + 3, // 3: Data.Search:output_type -> SearchResponse + 1, // 4: Data.Create:output_type -> CreateResponse + 3, // [3:5] is the sub-list for method output_type + 1, // [1:3] is the sub-list for method input_type + 1, // [1:1] is the sub-list for extension type_name + 1, // [1:1] is the sub-list for extension extendee + 0, // [0:1] is the sub-list for field type_name +} + +func init() { file_network_data_proto_init() } +func file_network_data_proto_init() { + if File_network_data_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_network_data_proto_rawDesc), len(file_network_data_proto_rawDesc)), + NumEnums: 0, + NumMessages: 4, + NumExtensions: 0, + NumServices: 1, + }, + GoTypes: file_network_data_proto_goTypes, + DependencyIndexes: file_network_data_proto_depIdxs, + MessageInfos: file_network_data_proto_msgTypes, + }.Build() + File_network_data_proto = out.File + file_network_data_proto_goTypes = nil + file_network_data_proto_depIdxs = nil +} diff --git a/network/data.proto b/network/data.proto new file mode 100644 index 0000000..479ec53 --- /dev/null +++ b/network/data.proto @@ -0,0 +1,30 @@ +syntax = "proto3"; + +import "google/protobuf/timestamp.proto"; + +option go_package = "github.com/Vince-maple-byte/KeyData/network"; + +service Data { + rpc Search(SearchRequest) returns (SearchResponse) {} + rpc Create(CreateRequest) returns (CreateResponse) {} +} + +message CreateRequest { + string operation = 1; + string key = 2; + string payload = 3; +} + +message CreateResponse { + bool success = 1; +} + +message SearchRequest { + string key = 2; +} + +message SearchResponse { + google.protobuf.Timestamp created_at = 1; + string key = 2; + string payload = 3; +} diff --git a/network/data_grpc.pb.go b/network/data_grpc.pb.go new file mode 100644 index 0000000..01ff269 --- /dev/null +++ b/network/data_grpc.pb.go @@ -0,0 +1,159 @@ +// Code generated by protoc-gen-go-grpc. DO NOT EDIT. +// versions: +// - protoc-gen-go-grpc v1.6.2 +// - protoc v7.35.0 +// source: network/data.proto + +package network + +import ( + context "context" + grpc "google.golang.org/grpc" + codes "google.golang.org/grpc/codes" + status "google.golang.org/grpc/status" +) + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the grpc package it is being compiled against. +// Requires gRPC-Go v1.64.0 or later. +const _ = grpc.SupportPackageIsVersion9 + +const ( + Data_Search_FullMethodName = "/Data/Search" + Data_Create_FullMethodName = "/Data/Create" +) + +// DataClient is the client API for Data service. +// +// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. +type DataClient interface { + Search(ctx context.Context, in *SearchRequest, opts ...grpc.CallOption) (*SearchResponse, error) + Create(ctx context.Context, in *CreateRequest, opts ...grpc.CallOption) (*CreateResponse, error) +} + +type dataClient struct { + cc grpc.ClientConnInterface +} + +func NewDataClient(cc grpc.ClientConnInterface) DataClient { + return &dataClient{cc} +} + +func (c *dataClient) Search(ctx context.Context, in *SearchRequest, opts ...grpc.CallOption) (*SearchResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(SearchResponse) + err := c.cc.Invoke(ctx, Data_Search_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *dataClient) Create(ctx context.Context, in *CreateRequest, opts ...grpc.CallOption) (*CreateResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(CreateResponse) + err := c.cc.Invoke(ctx, Data_Create_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +// DataServer is the server API for Data service. +// All implementations must embed UnimplementedDataServer +// for forward compatibility. +type DataServer interface { + Search(context.Context, *SearchRequest) (*SearchResponse, error) + Create(context.Context, *CreateRequest) (*CreateResponse, error) + mustEmbedUnimplementedDataServer() +} + +// UnimplementedDataServer must be embedded to have +// forward compatible implementations. +// +// NOTE: this should be embedded by value instead of pointer to avoid a nil +// pointer dereference when methods are called. +type UnimplementedDataServer struct{} + +func (UnimplementedDataServer) Search(context.Context, *SearchRequest) (*SearchResponse, error) { + return nil, status.Error(codes.Unimplemented, "method Search not implemented") +} +func (UnimplementedDataServer) Create(context.Context, *CreateRequest) (*CreateResponse, error) { + return nil, status.Error(codes.Unimplemented, "method Create not implemented") +} +func (UnimplementedDataServer) mustEmbedUnimplementedDataServer() {} +func (UnimplementedDataServer) testEmbeddedByValue() {} + +// UnsafeDataServer may be embedded to opt out of forward compatibility for this service. +// Use of this interface is not recommended, as added methods to DataServer will +// result in compilation errors. +type UnsafeDataServer interface { + mustEmbedUnimplementedDataServer() +} + +func RegisterDataServer(s grpc.ServiceRegistrar, srv DataServer) { + // If the following call panics, it indicates UnimplementedDataServer was + // embedded by pointer and is nil. This will cause panics if an + // unimplemented method is ever invoked, so we test this at initialization + // time to prevent it from happening at runtime later due to I/O. + if t, ok := srv.(interface{ testEmbeddedByValue() }); ok { + t.testEmbeddedByValue() + } + s.RegisterService(&Data_ServiceDesc, srv) +} + +func _Data_Search_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(SearchRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(DataServer).Search(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: Data_Search_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(DataServer).Search(ctx, req.(*SearchRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _Data_Create_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(CreateRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(DataServer).Create(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: Data_Create_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(DataServer).Create(ctx, req.(*CreateRequest)) + } + return interceptor(ctx, in, info, handler) +} + +// Data_ServiceDesc is the grpc.ServiceDesc for Data service. +// It's only intended for direct use with grpc.RegisterService, +// and not to be introspected or modified (even as a copy) +var Data_ServiceDesc = grpc.ServiceDesc{ + ServiceName: "Data", + HandlerType: (*DataServer)(nil), + Methods: []grpc.MethodDesc{ + { + MethodName: "Search", + Handler: _Data_Search_Handler, + }, + { + MethodName: "Create", + Handler: _Data_Create_Handler, + }, + }, + Streams: []grpc.StreamDesc{}, + Metadata: "network/data.proto", +} diff --git a/network/internals.go b/network/internals.go new file mode 100644 index 0000000..9a3dad3 --- /dev/null +++ b/network/internals.go @@ -0,0 +1,16 @@ +package network + +// This is the memtable +type InternalMemtable interface { + Write(key, value, operation string) (bool, error) + Get(key string) ([]byte, error) +} + +var NetworkMemtable func() InternalMemtable + +// This is the filepath struct +type Dir struct { + Path string +} + +var FileDir Dir diff --git a/network/request_handler.go b/network/request_handler.go new file mode 100644 index 0000000..df18243 --- /dev/null +++ b/network/request_handler.go @@ -0,0 +1,59 @@ +package network + +import ( + context "context" + + "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" +) + +type Server struct { + UnimplementedDataServer + + Memtable InternalMemtable + DataDir string +} + +func (s *Server) Search(ctx context.Context, search *SearchRequest) (*SearchResponse, error) { + res, err := s.searchHelper(search.GetKey()) + + if err != nil { + return &SearchResponse{CreatedAt: nil, Key: "", Payload: ""}, status.Error(codes.NotFound, err.Error()) + } + + contents := record.GetContents(res) + + return &SearchResponse{ + CreatedAt: timestamppb.New(contents.Timestamp), + Key: contents.Key, + Payload: contents.Payload, + }, nil +} + +func (s *Server) Create(ctx context.Context, create *CreateRequest) (*CreateResponse, error) { + //mem := NetworkMemtable() + + ok, err := s.Memtable.Write(create.GetKey(), create.GetPayload(), create.GetOperation()) + + if !ok { + return &CreateResponse{Success: ok}, status.Error(codes.Unknown, err.Error()) + } + + 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 +}