txn-file: Get first key inside range of both region & txn chunk (#1455)

 

Signed-off-by: Ping Yu <[email protected]>

txnkv/transaction/txn_file.go

@@ -228,13 +228,18 @@ func (r *txnChunkRange) getOverlapRegions(c *locate.RegionCache, bo *retry.Backo
 	regions := make([]*locate.KeyLocation, 0)
 	firstKeys := make([][]byte, 0)
 	startKey := r.smallest
+	exclusiveBiggest := kv.NextKey(r.biggest)
 	for bytes.Compare(startKey, r.biggest) <= 0 {
 		loc, err := c.LocateKey(bo, startKey)
 		if err != nil {
 			logutil.Logger(bo.GetCtx()).Error("locate key failed", zap.Error(err), zap.String("startKey", kv.StrKey(startKey)))
 			return nil, nil, errors.Wrap(err, "locate key failed")
 		}
-		firstKey, ok := MutationsHasDataInRange(mutations, loc.StartKey, loc.EndKey)
+		firstKey, ok := MutationsHasDataInRange(
+			mutations,
+			util.GetMaxStartKey(r.smallest, loc.StartKey),
+			util.GetMinEndKey(exclusiveBiggest, loc.EndKey),
+		)
 		if ok {
 			regions = append(regions, loc)
 			firstKeys = append(firstKeys, firstKey)

util/misc.go

@@ -35,6 +35,7 @@
 package util
 
 import (
+	"bytes"
 	"context"
 	"encoding/hex"
 	"fmt"
@@ -207,3 +208,23 @@ func HexRegionKey(key []byte) []byte {
 func HexRegionKeyStr(key []byte) string {
 	return String(HexRegionKey(key))
 }
+
+func GetMaxStartKey(lhs []byte, rhs []byte) []byte {
+	if bytes.Compare(lhs, rhs) > 0 {
+		return lhs
+	}
+	return rhs
+}
+
+func GetMinEndKey(lhs []byte, rhs []byte) []byte {
+	if len(rhs) == 0 {
+		return lhs
+	}
+	if len(lhs) == 0 {
+		return rhs
+	}
+	if bytes.Compare(lhs, rhs) < 0 {
+		return lhs
+	}
+	return rhs
+}

util/misc_test.go

@@ -97,3 +97,38 @@ func TestBytesToString(t *testing.T) {
 		assert.Equal(str, String(bytes))
 	}
 }
+
+func TestGetMaxStartKey(t *testing.T) {
+	assert := assert.New(t)
+
+	cases := []struct {
+		lhs, rhs, expected string
+	}{
+		{"", "", ""},
+		{"", "a", "a"},
+		{"a", "a", "a"},
+	}
+
+	for _, c := range cases {
+		assert.Equal([]byte(c.expected), GetMaxStartKey([]byte(c.lhs), []byte(c.rhs)))
+		assert.Equal([]byte(c.expected), GetMaxStartKey([]byte(c.rhs), []byte(c.lhs)))
+	}
+}
+
+func TestGetMinEndKey(t *testing.T) {
+	assert := assert.New(t)
+
+	cases := []struct {
+		lhs, rhs, expected string
+	}{
+		{"", "", ""},
+		{"a", "", "a"},
+		{"a", "a", "a"},
+		{"a", "b", "a"},
+	}
+
+	for _, c := range cases {
+		assert.Equal([]byte(c.expected), GetMinEndKey([]byte(c.lhs), []byte(c.rhs)))
+		assert.Equal([]byte(c.expected), GetMinEndKey([]byte(c.rhs), []byte(c.lhs)))
+	}
+}