*: handle region error for GetMvccByEncodedKey API (#47811)

close #47807

pkg/executor/BUILD.bazel

@@ -385,6 +385,7 @@ go_test(
         "//pkg/expression/aggregation",
         "//pkg/infoschema",
         "//pkg/kv",
+        "//pkg/meta",
         "//pkg/meta/autoid",
         "//pkg/metrics",
         "//pkg/parser",

pkg/executor/executor_failpoint_test.go

@@ -27,9 +27,11 @@ import (
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/pkg/config"
 	"github.com/pingcap/tidb/pkg/ddl"
+	"github.com/pingcap/tidb/pkg/meta"
 	"github.com/pingcap/tidb/pkg/parser/terror"
 	"github.com/pingcap/tidb/pkg/session"
 	"github.com/pingcap/tidb/pkg/store/copr"
+	"github.com/pingcap/tidb/pkg/store/helper"
 	"github.com/pingcap/tidb/pkg/testkit"
 	"github.com/pingcap/tidb/pkg/util/dbterror/exeerrors"
 	"github.com/pingcap/tidb/pkg/util/deadlockhistory"
@@ -622,3 +624,31 @@ func TestTiKVClientReadTimeout(t *testing.T) {
 	explain = fmt.Sprintf("%v", rows[0])
 	require.Regexp(t, ".*TableReader.* root  time:.*, loops:.* cop_task: {num: 1, .* rpc_num: 2.*", explain)
 }
+
+func TestGetMvccByEncodedKeyRegionError(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	h := helper.NewHelper(store.(helper.Storage))
+	txn, err := store.Begin()
+	require.NoError(t, err)
+	m := meta.NewMeta(txn)
+	schemaVersion := tk.Session().GetDomainInfoSchema().SchemaMetaVersion()
+	key := m.EncodeSchemaDiffKey(schemaVersion)
+
+	resp, err := h.GetMvccByEncodedKey(key)
+	require.NoError(t, err)
+	require.NotNil(t, resp.Info)
+	require.Equal(t, 1, len(resp.Info.Writes))
+	require.Less(t, uint64(0), resp.Info.Writes[0].CommitTs)
+	commitTs := resp.Info.Writes[0].CommitTs
+
+	require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/pkg/store/mockstore/unistore/epochNotMatch", "2*return(true)"))
+	defer func() {
+		require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/pkg/store/mockstore/unistore/epochNotMatch"))
+	}()
+	resp, err = h.GetMvccByEncodedKey(key)
+	require.NoError(t, err)
+	require.NotNil(t, resp.Info)
+	require.Equal(t, 1, len(resp.Info.Writes))
+	require.Equal(t, commitTs, resp.Info.Writes[0].CommitTs)
+}

pkg/store/helper/helper.go

@@ -98,23 +98,45 @@ func NewHelper(store Storage) *Helper {
 
 // GetMvccByEncodedKey get the MVCC value by the specific encoded key.
 func (h *Helper) GetMvccByEncodedKey(encodedKey kv.Key) (*kvrpcpb.MvccGetByKeyResponse, error) {
-	keyLocation, err := h.RegionCache.LocateKey(tikv.NewBackofferWithVars(context.Background(), 500, nil), encodedKey)
-	if err != nil {
-		return nil, derr.ToTiDBErr(err)
-	}
-
+	bo := tikv.NewBackofferWithVars(context.Background(), 5000, nil)
 	tikvReq := tikvrpc.NewRequest(tikvrpc.CmdMvccGetByKey, &kvrpcpb.MvccGetByKeyRequest{Key: encodedKey})
-	kvResp, err := h.Store.SendReq(tikv.NewBackofferWithVars(context.Background(), 500, nil), tikvReq, keyLocation.Region, time.Minute)
-	if err != nil {
-		logutil.BgLogger().Info("get MVCC by encoded key failed",
-			zap.Stringer("encodeKey", encodedKey),
-			zap.Reflect("region", keyLocation.Region),
-			zap.Stringer("keyLocation", keyLocation),
-			zap.Reflect("kvResp", kvResp),
-			zap.Error(err))
-		return nil, errors.Trace(err)
+	for {
+		keyLocation, err := h.RegionCache.LocateKey(bo, encodedKey)
+		if err != nil {
+			return nil, derr.ToTiDBErr(err)
+		}
+		kvResp, err := h.Store.SendReq(bo, tikvReq, keyLocation.Region, time.Minute)
+		if err != nil {
+			logutil.BgLogger().Info("get MVCC by encoded key failed",
+				zap.Stringer("encodeKey", encodedKey),
+				zap.Reflect("region", keyLocation.Region),
+				zap.Stringer("keyLocation", keyLocation),
+				zap.Reflect("kvResp", kvResp),
+				zap.Error(err))
+			return nil, errors.Trace(err)
+		}
+		regionErr, err := kvResp.GetRegionError()
+		if err != nil {
+			return nil, errors.Trace(err)
+		}
+		if regionErr != nil {
+			if err = bo.Backoff(tikv.BoRegionMiss(), errors.New(regionErr.String())); err != nil {
+				return nil, err
+			}
+			continue
+		}
+		mvccResp := kvResp.Resp.(*kvrpcpb.MvccGetByKeyResponse)
+		if errMsg := mvccResp.GetError(); errMsg != "" {
+			logutil.BgLogger().Info("get MVCC by encoded key failed",
+				zap.Stringer("encodeKey", encodedKey),
+				zap.Reflect("region", keyLocation.Region),
+				zap.Stringer("keyLocation", keyLocation),
+				zap.Reflect("kvResp", kvResp),
+				zap.String("error", errMsg))
+			return nil, errors.New(errMsg)
+		}
+		return mvccResp, nil
 	}
-	return kvResp.Resp.(*kvrpcpb.MvccGetByKeyResponse), nil
 }
 
 // MvccKV wraps the key's mvcc info in tikv.

pkg/store/mockstore/unistore/rpc.go

@@ -66,6 +66,11 @@ func (c *RPCClient) SendRequest(ctx context.Context, addr string, req *tikvrpc.R
 			failpoint.Return(tikvrpc.GenRegionErrorResp(req, &errorpb.Error{ServerIsBusy: &errorpb.ServerIsBusy{}}))
 		}
 	})
+	failpoint.Inject("epochNotMatch", func(val failpoint.Value) {
+		if val.(bool) {
+			failpoint.Return(tikvrpc.GenRegionErrorResp(req, &errorpb.Error{EpochNotMatch: &errorpb.EpochNotMatch{}}))
+		}
+	})
 
 	failpoint.Inject("unistoreRPCClientSendHook", func(val failpoint.Value) {
 		if fn := UnistoreRPCClientSendHook.Load(); val.(bool) && fn != nil {