Skip to content

Commit

Permalink
flashback: retry getStoreGlobalMinSafeTS during execute flashback (#…
Browse files Browse the repository at this point in the history 
…41100)

close #41086
  • Loading branch information
@Defined2014
Defined2014 authored Feb 7, 2023
1 parent 98aff8c commit c996d64
Show file tree
Hide file tree
Showing 5 changed files with 115 additions and 101 deletions.
39 changes: 35 additions & 4 deletions ddl/cluster.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,6 @@ import (
"github.com/pingcap/kvproto/pkg/kvrpcpb"
"github.com/pingcap/tidb/ddl/util"
"github.com/pingcap/tidb/domain/infosync"
"github.com/pingcap/tidb/expression"
"github.com/pingcap/tidb/infoschema"
"github.com/pingcap/tidb/kv"
"github.com/pingcap/tidb/meta"
Expand Down Expand Up @@ -99,6 +98,16 @@ func recoverPDSchedule(pdScheduleParam map[string]interface{}) error {
return infosync.SetPDScheduleConfig(context.Background(), pdScheduleParam)
}

func getStoreGlobalMinSafeTS(s kv.Storage) time.Time {
minSafeTS := s.GetMinSafeTS(kv.GlobalTxnScope)
// Inject mocked SafeTS for test.
failpoint.Inject("injectSafeTS", func(val failpoint.Value) {
injectTS := val.(int)
minSafeTS = uint64(injectTS)
})
return oracle.GetTimeFromTS(minSafeTS)
}

// ValidateFlashbackTS validates that flashBackTS in range [gcSafePoint, currentTS).
func ValidateFlashbackTS(ctx context.Context, sctx sessionctx.Context, flashBackTS uint64) error {
currentTS, err := sctx.GetStore().GetOracle().GetStaleTimestamp(ctx, oracle.GlobalTxnScope, 0)
Expand All @@ -111,12 +120,34 @@ func ValidateFlashbackTS(ctx context.Context, sctx sessionctx.Context, flashBack
}
currentTS = currentVer.Ver
}
if oracle.GetTimeFromTS(flashBackTS).After(oracle.GetTimeFromTS(currentTS)) {
oracleFlashbackTS := oracle.GetTimeFromTS(flashBackTS)
if oracleFlashbackTS.After(oracle.GetTimeFromTS(currentTS)) {
return errors.Errorf("cannot set flashback timestamp to future time")
}
if oracle.GetTimeFromTS(flashBackTS).After(expression.GetMinSafeTime(sctx)) {
return errors.Errorf("cannot set flashback timestamp to too close to present time")

flashbackGetMinSafeTimeTimeout := time.Minute
failpoint.Inject("changeFlashbackGetMinSafeTimeTimeout", func(val failpoint.Value) {
t := val.(int)
flashbackGetMinSafeTimeTimeout = time.Duration(t)
})

start := time.Now()
minSafeTime := getStoreGlobalMinSafeTS(sctx.GetStore())
ticker := time.NewTicker(time.Second)
defer ticker.Stop()
for oracleFlashbackTS.After(minSafeTime) {
if time.Since(start) >= flashbackGetMinSafeTimeTimeout {
return errors.Errorf("cannot set flashback timestamp after min-resolved-ts(%s)", minSafeTime)
}
select {
case <-ticker.C:
minSafeTime = getStoreGlobalMinSafeTS(sctx.GetStore())
break
case <-ctx.Done():
return ctx.Err()
}
}

gcSafePoint, err := gcutil.GetGCSafePoint(sctx)
if err != nil {
return err
Expand Down
28 changes: 8 additions & 20 deletions ddl/cluster_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -71,9 +71,7 @@ func TestFlashbackCloseAndResetPDSchedule(t *testing.T) {

injectSafeTS := oracle.GoTimeToTS(time.Now().Add(10 * time.Second))
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/mockFlashbackTest", `return(true)`))
require.NoError(t, failpoint.Enable("tikvclient/injectSafeTS",
fmt.Sprintf("return(%v)", injectSafeTS)))
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/expression/injectSafeTS",
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/injectSafeTS",
fmt.Sprintf("return(%v)", injectSafeTS)))

oldValue := map[string]interface{}{
Expand Down Expand Up @@ -111,8 +109,7 @@ func TestFlashbackCloseAndResetPDSchedule(t *testing.T) {
require.EqualValues(t, finishValue["merge-schedule-limit"], 1)

require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/mockFlashbackTest"))
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/expression/injectSafeTS"))
require.NoError(t, failpoint.Disable("tikvclient/injectSafeTS"))
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/injectSafeTS"))
}

func TestAddDDLDuringFlashback(t *testing.T) {
Expand All @@ -128,9 +125,7 @@ func TestAddDDLDuringFlashback(t *testing.T) {

injectSafeTS := oracle.GoTimeToTS(oracle.GetTimeFromTS(ts).Add(10 * time.Second))
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/mockFlashbackTest", `return(true)`))
require.NoError(t, failpoint.Enable("tikvclient/injectSafeTS",
fmt.Sprintf("return(%v)", injectSafeTS)))
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/expression/injectSafeTS",
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/injectSafeTS",
fmt.Sprintf("return(%v)", injectSafeTS)))

timeBeforeDrop, _, safePointSQL, resetGC := MockGC(tk)
Expand All @@ -150,8 +145,7 @@ func TestAddDDLDuringFlashback(t *testing.T) {

dom.DDL().SetHook(originHook)
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/mockFlashbackTest"))
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/expression/injectSafeTS"))
require.NoError(t, failpoint.Disable("tikvclient/injectSafeTS"))
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/injectSafeTS"))
}

func TestGlobalVariablesOnFlashback(t *testing.T) {
Expand All @@ -167,9 +161,7 @@ func TestGlobalVariablesOnFlashback(t *testing.T) {

injectSafeTS := oracle.GoTimeToTS(oracle.GetTimeFromTS(ts).Add(10 * time.Second))
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/mockFlashbackTest", `return(true)`))
require.NoError(t, failpoint.Enable("tikvclient/injectSafeTS",
fmt.Sprintf("return(%v)", injectSafeTS)))
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/expression/injectSafeTS",
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/injectSafeTS",
fmt.Sprintf("return(%v)", injectSafeTS)))

timeBeforeDrop, _, safePointSQL, resetGC := MockGC(tk)
Expand Down Expand Up @@ -232,8 +224,7 @@ func TestGlobalVariablesOnFlashback(t *testing.T) {

dom.DDL().SetHook(originHook)
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/mockFlashbackTest"))
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/expression/injectSafeTS"))
require.NoError(t, failpoint.Disable("tikvclient/injectSafeTS"))
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/injectSafeTS"))
}

func TestCancelFlashbackCluster(t *testing.T) {
Expand All @@ -247,9 +238,7 @@ func TestCancelFlashbackCluster(t *testing.T) {

injectSafeTS := oracle.GoTimeToTS(oracle.GetTimeFromTS(ts).Add(10 * time.Second))
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/mockFlashbackTest", `return(true)`))
require.NoError(t, failpoint.Enable("tikvclient/injectSafeTS",
fmt.Sprintf("return(%v)", injectSafeTS)))
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/expression/injectSafeTS",
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/injectSafeTS",
fmt.Sprintf("return(%v)", injectSafeTS)))

timeBeforeDrop, _, safePointSQL, resetGC := MockGC(tk)
Expand Down Expand Up @@ -284,6 +273,5 @@ func TestCancelFlashbackCluster(t *testing.T) {
dom.DDL().SetHook(originHook)

require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/mockFlashbackTest"))
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/expression/injectSafeTS"))
require.NoError(t, failpoint.Disable("tikvclient/injectSafeTS"))
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/injectSafeTS"))
}
7 changes: 2 additions & 5 deletions ddl/tiflashtest/ddl_tiflash_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -458,9 +458,7 @@ func TestTiFlashFlashbackCluster(t *testing.T) {

injectSafeTS := oracle.GoTimeToTS(oracle.GetTimeFromTS(ts).Add(10 * time.Second))
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/mockFlashbackTest", `return(true)`))
require.NoError(t, failpoint.Enable("tikvclient/injectSafeTS",
fmt.Sprintf("return(%v)", injectSafeTS)))
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/expression/injectSafeTS",
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/injectSafeTS",
fmt.Sprintf("return(%v)", injectSafeTS)))

ChangeGCSafePoint(tk, time.Now().Add(-10*time.Second), "true", "10m0s")
Expand All @@ -473,8 +471,7 @@ func TestTiFlashFlashbackCluster(t *testing.T) {
tk.MustGetErrMsg(fmt.Sprintf("flashback cluster to timestamp '%s'", oracle.GetTimeFromTS(ts)), errorMsg)

require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/mockFlashbackTest"))
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/expression/injectSafeTS"))
require.NoError(t, failpoint.Disable("tikvclient/injectSafeTS"))
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/injectSafeTS"))
}

func CheckTableAvailableWithTableName(dom *domain.Domain, t *testing.T, count uint64, labels []string, db string, table string) {
Expand Down
61 changes: 47 additions & 14 deletions executor/recover_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -307,9 +307,7 @@ func TestRecoverClusterMeetError(t *testing.T) {

injectSafeTS := oracle.GoTimeToTS(flashbackTs.Add(10 * time.Second))
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/mockFlashbackTest", `return(true)`))
require.NoError(t, failpoint.Enable("tikvclient/injectSafeTS",
fmt.Sprintf("return(%v)", injectSafeTS)))
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/expression/injectSafeTS",
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/injectSafeTS",
fmt.Sprintf("return(%v)", injectSafeTS)))

// Get GC safe point error.
Expand Down Expand Up @@ -346,8 +344,7 @@ func TestRecoverClusterMeetError(t *testing.T) {
errorMsg = fmt.Sprintf("[ddl:-1]Detected TiDB upgrade during [%s, now), can't do flashback", oracle.GetTimeFromTS(nowTS).String())
tk.MustGetErrMsg(fmt.Sprintf("flashback cluster to timestamp '%s'", oracle.GetTimeFromTS(nowTS)), errorMsg)

require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/expression/injectSafeTS"))
require.NoError(t, failpoint.Disable("tikvclient/injectSafeTS"))
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/injectSafeTS"))
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/mockFlashbackTest"))
}

Expand All @@ -356,6 +353,7 @@ func TestFlashbackWithSafeTs(t *testing.T) {
tk := testkit.NewTestKit(t, store)

require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/mockFlashbackTest", `return(true)`))
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/changeFlashbackGetMinSafeTimeTimeout", `return(0)`))

timeBeforeDrop, _, safePointSQL, resetGC := MockGC(tk)
defer resetGC()
Expand All @@ -380,9 +378,8 @@ func TestFlashbackWithSafeTs(t *testing.T) {
compareWithSafeTS: 0,
},
{
name: "10 seconds ago to now, safeTS 5 secs ago",
// Add flashbackTs.Add(-500*time.Millisecond) to avoid flashback time range overlapped.
sql: fmt.Sprintf("flashback cluster to timestamp '%s'", flashbackTs.Add(-500*time.Millisecond)),
name: "10 seconds ago to now, safeTS 5 secs ago",
sql: fmt.Sprintf("flashback cluster to timestamp '%s'", flashbackTs),
injectSafeTS: oracle.GoTimeToTS(flashbackTs.Add(10 * time.Second)),
compareWithSafeTS: -1,
},
Expand All @@ -395,19 +392,55 @@ func TestFlashbackWithSafeTs(t *testing.T) {
}
for _, testcase := range testcases {
t.Log(testcase.name)
require.NoError(t, failpoint.Enable("tikvclient/injectSafeTS",
fmt.Sprintf("return(%v)", testcase.injectSafeTS)))
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/expression/injectSafeTS",
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/injectSafeTS",
fmt.Sprintf("return(%v)", testcase.injectSafeTS)))
if testcase.compareWithSafeTS == 1 {
start := time.Now()
tk.MustContainErrMsg(testcase.sql,
"cannot set flashback timestamp to too close to present time")
"cannot set flashback timestamp after min-resolved-ts")
// When set `flashbackGetMinSafeTimeTimeout` = 0, no retry for `getStoreGlobalMinSafeTS`.
require.Less(t, time.Since(start), time.Second)
} else {
tk.MustExec(testcase.sql)
}
}
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/expression/injectSafeTS"))
require.NoError(t, failpoint.Disable("tikvclient/injectSafeTS"))
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/injectSafeTS"))
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/mockFlashbackTest"))
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/changeFlashbackGetMinSafeTimeTimeout"))
}

func TestFlashbackRetryGetMinSafeTime(t *testing.T) {
store := testkit.CreateMockStore(t)
tk := testkit.NewTestKit(t, store)

require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/mockFlashbackTest", `return(true)`))

timeBeforeDrop, _, safePointSQL, resetGC := MockGC(tk)
defer resetGC()

// Set GC safe point.
tk.MustExec(fmt.Sprintf(safePointSQL, timeBeforeDrop))

time.Sleep(time.Second)
ts, _ := tk.Session().GetStore().GetOracle().GetTimestamp(context.Background(), &oracle.Option{})
flashbackTs := oracle.GetTimeFromTS(ts)

require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/injectSafeTS",
fmt.Sprintf("return(%v)", oracle.GoTimeToTS(flashbackTs.Add(-10*time.Minute)))))

go func() {
time.Sleep(2 * time.Second)
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/injectSafeTS",
fmt.Sprintf("return(%v)", oracle.GoTimeToTS(flashbackTs.Add(10*time.Minute)))))
}()

start := time.Now()
tk.MustExec(fmt.Sprintf("flashback cluster to timestamp '%s'", flashbackTs))
duration := time.Since(start)
require.Greater(t, duration, 2*time.Second)
require.Less(t, duration, 5*time.Second)

require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/injectSafeTS"))
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/mockFlashbackTest"))
}

Expand Down
Loading

0 comments on commit c996d64

Please sign in to comment.