fix: compact_map get error mismatching cokie (#3748)

* fix: compact_map get error

* fix: CompactSection delete lock and move test to compact_map

Co-authored-by: shibinbin <shibinbin@megvii.com>

weed/storage/needle_map/compact_map.go

@@ -144,9 +144,13 @@ func (cs *CompactSection) deleteOverflowEntry(key SectionalNeedleId) {
 
 // return old entry size
 func (cs *CompactSection) Delete(key NeedleId) Size {
-	skey := SectionalNeedleId(key - cs.start)
 	cs.Lock()
+	defer cs.Unlock()
 	ret := Size(0)
+	if key > cs.end {
+		return ret
+	}
+	skey := SectionalNeedleId(key - cs.start)
 	if i := cs.binarySearchValues(skey); i >= 0 {
 		if cs.values[i].Size > 0 && cs.values[i].Size.IsValid() {
 			ret = cs.values[i].Size
@@ -157,23 +161,23 @@ func (cs *CompactSection) Delete(key NeedleId) Size {
 		cs.deleteOverflowEntry(skey)
 		ret = v.Size
 	}
-	cs.Unlock()
 	return ret
 }
 func (cs *CompactSection) Get(key NeedleId) (*NeedleValue, bool) {
 	cs.RLock()
+	defer cs.RUnlock()
+	if key > cs.end {
+		return nil, false
+	}
 	skey := SectionalNeedleId(key - cs.start)
 	if ve, v, ok := cs.findOverflowEntry(skey); ok {
-		cs.RUnlock()
 		nv := toNeedleValue(ve, v, cs)
 		return &nv, true
 	}
 	if i := cs.binarySearchValues(skey); i >= 0 {
-		cs.RUnlock()
 		nv := toNeedleValue(cs.valuesExtra[i], cs.values[i], cs)
 		return &nv, true
 	}
-	cs.RUnlock()
 	return nil, false
 }
 func (cs *CompactSection) binarySearchValues(key SectionalNeedleId) int {

weed/storage/needle_map/compact_map_test.go

@@ -4,6 +4,8 @@ import (
 	"fmt"
 	"github.com/seaweedfs/seaweedfs/weed/sequence"
 	. "github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"log"
+	"os"
 	"testing"
 )
 
@@ -179,3 +181,40 @@ func TestOverflow(t *testing.T) {
 	println()
 
 }
+
+func TestCompactSection_Get(t *testing.T) {
+	var maps []*CompactMap
+	totalRowCount := uint64(0)
+	indexFile, ie := os.OpenFile("../../../test/data/sample.idx",
+		os.O_RDWR|os.O_RDONLY, 0644)
+	defer indexFile.Close()
+	if ie != nil {
+		log.Fatalln(ie)
+	}
+
+	m, rowCount := loadNewNeedleMap(indexFile)
+	maps = append(maps, m)
+	totalRowCount += rowCount
+	m.Set(1574318345753513987, ToOffset(10002), 10002)
+	nv,ok := m.Get(1574318345753513987)
+	if ok {
+		t.Log(uint64(nv.Key))
+	}
+
+	nv1, ok := m.Get(1574318350048481283)
+	if ok {
+		t.Error(uint64(nv1.Key))
+	}
+
+	m.Set(1574318350048481283, ToOffset(10002), 10002)
+	nv2,ok1 := m.Get(1574318350048481283)
+	if ok1 {
+		t.Log(uint64(nv2.Key))
+	}
+
+	m.Delete(nv2.Key)
+	nv3,has := m.Get(nv2.Key)
+	if has && nv3.Size > 0 {
+		t.Error(uint64(nv3.Size))
+	}
+}