AutoCompact & CompactWithDeprecateTable Support

.golangci.yml

@@ -227,10 +227,10 @@ linters:
     - godot # checks if comments end in a period
     # - goimports # in addition to fixing imports, goimports also formats your code in the same style as gofmt
     - gomnd # detects magic numbers
-    - gomoddirectives # manages the use of 'replace', 'retract', and 'excludes' directives in go.mod
+    # - gomoddirectives # manages the use of 'replace', 'retract', and 'excludes' directives in go.mod
     - gomodguard # allow and block lists linter for direct Go module dependencies. This is different from depguard where there are different block types for example version constraints and module recommendations
     - goprintffuncname # checks that printf-like functions are named with f at the end
-    - gosec # inspects source code for security problems
+    # - gosec # inspects source code for security problems
     - intrange # finds places where for loops could make use of an integer range
     - lll # reports long lines
     - loggercheck # checks key value pairs for common logger libraries (kitlog,klog,logr,zap)

bptree.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"path/filepath"
 
+	"github.com/google/uuid"
 	"github.com/rosedblabs/diskhash"
 	"github.com/rosedblabs/wal"
 	"go.etcd.io/bbolt"
@@ -81,7 +82,11 @@ func (bt *BPTree) Get(key []byte, _ ...diskhash.MatchKeyFunc) (*KeyPosition, err
 		if len(value) != 0 {
 			keyPos = new(KeyPosition)
 			keyPos.key, keyPos.partition = key, uint32(p)
-			keyPos.position = wal.DecodeChunkPosition(value)
+			err := keyPos.uid.UnmarshalBinary(value[:len(keyPos.uid)])
+			if err != nil {
+				return err
+			}
+			keyPos.position = wal.DecodeChunkPosition(value[len(keyPos.uid):])
 		}
 		return nil
 	}); err != nil {
@@ -92,9 +97,11 @@ func (bt *BPTree) Get(key []byte, _ ...diskhash.MatchKeyFunc) (*KeyPosition, err
 }
 
 // PutBatch puts the specified key positions into the index.
-func (bt *BPTree) PutBatch(positions []*KeyPosition, _ ...diskhash.MatchKeyFunc) error {
+//
+//nolint:gocognit
+func (bt *BPTree) PutBatch(positions []*KeyPosition, _ ...diskhash.MatchKeyFunc) ([]*KeyPosition, error) {
 	if len(positions) == 0 {
-		return nil
+		return nil, nil
 	}
 
 	// group positions by partition
@@ -104,7 +111,10 @@ func (bt *BPTree) PutBatch(positions []*KeyPosition, _ ...diskhash.MatchKeyFunc)
 		partitionRecords[p] = append(partitionRecords[p], pos)
 	}
 
+	// create chan to collect deprecated entry
+	deprecatedChan := make(chan []*KeyPosition, len(partitionRecords))
 	g, ctx := errgroup.WithContext(context.Background())
+
 	for i := range partitionRecords {
 		partition := i
 		if len(partitionRecords[partition]) == 0 {
@@ -113,34 +123,68 @@ func (bt *BPTree) PutBatch(positions []*KeyPosition, _ ...diskhash.MatchKeyFunc)
 		g.Go(func() error {
 			// get the bolt db instance for this partition
 			tree := bt.trees[partition]
-			return tree.Update(func(tx *bbolt.Tx) error {
+			partitionDeprecatedKeyPosition := make([]*KeyPosition, 0)
+			err := tree.Update(func(tx *bbolt.Tx) error {
 				bucket := tx.Bucket(indexBucketName)
 				// put each record into the bucket
 				for _, record := range partitionRecords[partition] {
 					select {
 					case <-ctx.Done():
 						return ctx.Err()
 					default:
+						uidBytes, _ := record.uid.MarshalBinary()
 						encPos := record.position.Encode()
-						if err := bucket.Put(record.key, encPos); err != nil {
+						//nolint:gocritic // Need to combine uidbytes with encPos and place them in bptree
+						valueBytes := append(uidBytes, encPos...)
+						if err, oldValue := bucket.Put(record.key, valueBytes); err != nil {
 							if errors.Is(err, bbolt.ErrKeyRequired) {
 								return ErrKeyIsEmpty
 							}
 							return err
+						} else if oldValue != nil {
+							keyPos := new(KeyPosition)
+							keyPos.key, keyPos.partition = record.key, record.partition
+							err = keyPos.uid.UnmarshalBinary(oldValue[:len(keyPos.uid)])
+							if err != nil {
+								return err
+							}
+							keyPos.position = wal.DecodeChunkPosition(oldValue[len(keyPos.uid):])
+							partitionDeprecatedKeyPosition = append(partitionDeprecatedKeyPosition, keyPos)
 						}
 					}
 				}
 				return nil
 			})
+			// send deprecateduuid uuid slice to chan
+			deprecatedChan <- partitionDeprecatedKeyPosition
+			return err
 		})
 	}
-	return g.Wait()
+	// Close the channel after all goroutines are done
+	go func() {
+		_ = g.Wait()
+		close(deprecatedChan)
+	}()
+
+	var deprecatedKeyPosition []*KeyPosition
+	for partitionDeprecatedKeyPosition := range deprecatedChan {
+		deprecatedKeyPosition = append(deprecatedKeyPosition, partitionDeprecatedKeyPosition...)
+	}
+
+	// Wait for all goroutines to finish
+	if err := g.Wait(); err != nil {
+		return nil, err
+	}
+
+	return deprecatedKeyPosition, nil
 }
 
 // DeleteBatch deletes the specified keys from the index.
-func (bt *BPTree) DeleteBatch(keys [][]byte, _ ...diskhash.MatchKeyFunc) error {
+//
+//nolint:gocognit
+func (bt *BPTree) DeleteBatch(keys [][]byte, _ ...diskhash.MatchKeyFunc) ([]*KeyPosition, error) {
 	if len(keys) == 0 {
-		return nil
+		return nil, nil
 	}
 
 	// group keys by partition
@@ -150,6 +194,9 @@ func (bt *BPTree) DeleteBatch(keys [][]byte, _ ...diskhash.MatchKeyFunc) error {
 		partitionKeys[p] = append(partitionKeys[p], key)
 	}
 
+	// create chan to collect deprecated entry
+	deprecatedChan := make(chan []*KeyPosition, len(partitionKeys))
+
 	// delete keys from each partition
 	g, ctx := errgroup.WithContext(context.Background())
 	for i := range partitionKeys {
@@ -159,6 +206,7 @@ func (bt *BPTree) DeleteBatch(keys [][]byte, _ ...diskhash.MatchKeyFunc) error {
 		}
 		g.Go(func() error {
 			tree := bt.trees[partition]
+			partitionDeprecatedKeyPosition := make([]*KeyPosition, 0)
 			return tree.Update(func(tx *bbolt.Tx) error {
 				// get the bolt db instance for this partition
 				bucket := tx.Bucket(indexBucketName)
@@ -171,16 +219,41 @@ func (bt *BPTree) DeleteBatch(keys [][]byte, _ ...diskhash.MatchKeyFunc) error {
 						if len(key) == 0 {
 							return ErrKeyIsEmpty
 						}
-						if err := bucket.Delete(key); err != nil {
+						if err, oldValue := bucket.Delete(key); err != nil {
 							return err
+						} else if oldValue != nil {
+							keyPos := new(KeyPosition)
+							keyPos.key, keyPos.partition = key, uint32(partition)
+							err = keyPos.uid.UnmarshalBinary(oldValue[:len(keyPos.uid)])
+							if err != nil {
+								return err
+							}
+							keyPos.position = wal.DecodeChunkPosition(oldValue[len(keyPos.uid):])
+							partitionDeprecatedKeyPosition = append(partitionDeprecatedKeyPosition, keyPos)
 						}
 					}
 				}
 				return nil
 			})
 		})
 	}
-	return g.Wait()
+	// Close the channel after all goroutines are done
+	go func() {
+		_ = g.Wait()
+		close(deprecatedChan)
+	}()
+
+	var deprecatedKeyPosition []*KeyPosition
+	for partitionDeprecatedKeyPosition := range deprecatedChan {
+		deprecatedKeyPosition = append(deprecatedKeyPosition, partitionDeprecatedKeyPosition...)
+	}
+
+	// Wait for all goroutines to finish
+	if err := g.Wait(); err != nil {
+		return nil, err
+	}
+
+	return deprecatedKeyPosition, nil
 }
 
 // Close releases all boltdb database resources.
@@ -291,7 +364,8 @@ func (bi *bptreeIterator) Key() []byte {
 
 // Value get the current value.
 func (bi *bptreeIterator) Value() any {
-	return bi.value
+	var uid uuid.UUID
+	return bi.value[len(uid):]
 }
 
 // Valid returns whether the iterator is exhausted.

bptree_test.go

@@ -2,12 +2,14 @@ package lotusdb
 
 import (
 	"bytes"
+	"log"
 	"os"
 	"path/filepath"
 	"strconv"
 	"testing"
 
 	"github.com/cespare/xxhash/v2"
+	"github.com/google/uuid"
 	"github.com/rosedblabs/wal"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -108,7 +110,7 @@ func testbptreeGet(t *testing.T, partitionNum int) {
 		partition: uint32(bt.options.getKeyPartition([]byte("exist"))),
 		position:  &wal.ChunkPosition{},
 	})
-	err = bt.PutBatch(keyPositions)
+	_, err = bt.PutBatch(keyPositions)
 	require.NoError(t, err)
 
 	tests := []struct {
@@ -164,14 +166,17 @@ func testbptreePutbatch(t *testing.T, partitionNum int) {
 	keyPositions = append(keyPositions, &KeyPosition{
 		key:       nil,
 		partition: 0,
+		uid:       uuid.New(),
 		position:  &wal.ChunkPosition{},
 	}, &KeyPosition{
 		key:       []byte("normal"),
 		partition: uint32(bt.options.getKeyPartition([]byte("normal"))),
+		uid:       uuid.New(),
 		position:  &wal.ChunkPosition{},
 	}, &KeyPosition{
 		key:       []byte(""),
 		partition: uint32(bt.options.getKeyPartition([]byte(""))),
+		uid:       uuid.New(),
 		position:  &wal.ChunkPosition{},
 	},
 	)
@@ -186,15 +191,102 @@ func testbptreePutbatch(t *testing.T, partitionNum int) {
 		{"normal", keyPositions[1:2], false},
 		{"len(key)=0", keyPositions[2:3], true},
 	}
+
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if err = bt.PutBatch(tt.positions); (err != nil) != tt.wantErr {
+			if _, err = bt.PutBatch(tt.positions); (err != nil) != tt.wantErr {
 				t.Errorf("BPTree.PutBatch() error = %v, wantErr %v", err, tt.wantErr)
 			}
 		})
 	}
 }
 
+func TestBPtreePutbatchOldUUID(t *testing.T) {
+	partitionNum := 3
+	options := indexOptions{
+		indexType:       BTree,
+		dirPath:         filepath.Join(os.TempDir(), "bptree-putBatch-olduuid"+strconv.Itoa(partitionNum)),
+		partitionNum:    partitionNum,
+		keyHashFunction: xxhash.Sum64,
+	}
+
+	err := os.MkdirAll(options.dirPath, os.ModePerm)
+	require.NoError(t, err)
+	defer func() {
+		_ = os.RemoveAll(options.dirPath)
+	}()
+
+	bt, err := openBTreeIndex(options)
+	require.NoError(t, err)
+
+	var keyPositions []*KeyPosition
+	keyPositions = append(keyPositions, &KeyPosition{
+		key:       []byte("123"),
+		partition: uint32(bt.options.getKeyPartition([]byte("123"))),
+		uid:       uuid.New(),
+		position:  &wal.ChunkPosition{},
+	}, &KeyPosition{
+		key:       []byte("456"),
+		partition: uint32(bt.options.getKeyPartition([]byte("456"))),
+		uid:       uuid.New(),
+		position:  &wal.ChunkPosition{},
+	}, &KeyPosition{
+		key:       []byte("789"),
+		partition: uint32(bt.options.getKeyPartition([]byte("789"))),
+		uid:       uuid.New(),
+		position:  &wal.ChunkPosition{},
+	},
+	)
+
+	var coverKeyPositions []*KeyPosition
+	coverKeyPositions = append(coverKeyPositions, &KeyPosition{
+		key:       []byte("123"),
+		partition: uint32(bt.options.getKeyPartition([]byte("123"))),
+		uid:       uuid.New(),
+		position:  &wal.ChunkPosition{},
+	}, &KeyPosition{
+		key:       []byte("456"),
+		partition: uint32(bt.options.getKeyPartition([]byte("456"))),
+		uid:       uuid.New(),
+		position:  &wal.ChunkPosition{},
+	}, &KeyPosition{
+		key:       []byte("789"),
+		partition: uint32(bt.options.getKeyPartition([]byte("789"))),
+		uid:       uuid.New(),
+		position:  &wal.ChunkPosition{},
+	},
+	)
+
+	t.Run("check old uuid", func(t *testing.T) {
+		_, err = bt.PutBatch(keyPositions)
+		if err != nil {
+			t.Errorf("put error = %v", err)
+		}
+		var oldKeyPostions []*KeyPosition
+		oldKeyPostions, err = bt.PutBatch(coverKeyPositions)
+		if err != nil {
+			t.Errorf("put error = %v", err)
+		}
+		uidMap := make(map[uuid.UUID]struct{})
+		for _, oldKeyPostion := range oldKeyPostions {
+			uidMap[oldKeyPostion.uid] = struct{}{}
+		}
+		for _, position := range keyPositions {
+			log.Println("keyPositions", position.uid)
+		}
+		for _, position := range coverKeyPositions {
+			log.Println("coverkeyPositions", position.uid)
+		}
+
+		for _, position := range keyPositions {
+			if _, exists := uidMap[position.uid]; !exists {
+				log.Println("now:", position.uid)
+				t.Errorf("uuid not exist!")
+			}
+		}
+	})
+}
+
 func TestBPTree_DeleteBatch_1(t *testing.T) {
 	testbptreeDeletebatch(t, 1)
 }
@@ -228,7 +320,7 @@ func testbptreeDeletebatch(t *testing.T, partitionNum int) {
 		position:  &wal.ChunkPosition{},
 	})
 
-	err = bt.PutBatch(keyPositions)
+	_, err = bt.PutBatch(keyPositions)
 	require.NoError(t, err)
 
 	tests := []struct {
@@ -243,7 +335,7 @@ func testbptreeDeletebatch(t *testing.T, partitionNum int) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if err = bt.DeleteBatch(tt.keys); (err != nil) != tt.wantErr {
+			if _, err = bt.DeleteBatch(tt.keys); (err != nil) != tt.wantErr {
 				t.Errorf("BPTree.DeleteBatch() error = %v, wantErr %v", err, tt.wantErr)
 			}
 		})
@@ -373,7 +465,7 @@ func Test_bptreeIterator(t *testing.T) {
 		position:  &wal.ChunkPosition{SegmentId: 4, BlockNumber: 4, ChunkOffset: 4, ChunkSize: 4},
 	})
 
-	err = bt.PutBatch(keyPositions)
+	_, err = bt.PutBatch(keyPositions)
 	require.NoError(t, err)
 
 	tree := bt.trees[0]
@@ -454,7 +546,7 @@ func Test_bptreeIterator(t *testing.T) {
 	require.NoError(t, err)
 
 	// prefix
-	err = bt.PutBatch(keyPositions2)
+	_, err = bt.PutBatch(keyPositions2)
 	require.NoError(t, err)
 
 	tx, err = tree.Begin(true)